fix(diagnostics): preserve telemetry during service shutdown (#119705)

* fix(diagnostics): preserve otel service shutdown lifecycle

* test(diagnostics): type OTEL unsubscribe callback

* test(diagnostics): normalize async stop assertion

* fix(plugins): preserve ordinary stop handling

extensions/diagnostics-otel/src/service.test.ts

@@ -1027,6 +1027,27 @@ describe("diagnostics-otel service", () => {
     expect(telemetryState.tracer.startSpan).not.toHaveBeenCalled();
   });
 
+  test("attempts every provider shutdown and reports every failure", async () => {
+    const logError = new Error("log provider failed");
+    const sdkError = new Error("SDK providers failed");
+    logShutdown.mockRejectedValueOnce(logError);
+    sdkShutdown.mockRejectedValueOnce(sdkError);
+    const { service, ctx } = await startOtelService({ traces: true, metrics: true, logs: true });
+
+    const stopError = await Promise.resolve(service.stop?.(ctx)).catch((error: unknown) => error);
+
+    expect(logShutdown).toHaveBeenCalledTimes(1);
+    expect(sdkShutdown).toHaveBeenCalledTimes(1);
+    expect(stopError).toBeInstanceOf(AggregateError);
+    expect(stopError).toMatchObject({
+      errors: [logError, sdkError],
+      message: expect.stringContaining("log provider failed"),
+    });
+    expect(stopError).toMatchObject({
+      message: expect.stringContaining("SDK providers failed"),
+    });
+  });
+
   test("registers and removes an OTLP exporter unhandled rejection handler", async () => {
     const { service, ctx } = await startOtelService({ traces: true, metrics: true, logs: true });
 
@@ -1058,6 +1079,38 @@ describe("diagnostics-otel service", () => {
     expect(unhandledRejectionHandlerState.getHandlers()).toHaveLength(0);
   });
 
+  test("cleans up existing providers and does not reinitialize without capability", async () => {
+    const service = createDiagnosticsOtelService();
+    const enabledCtx = createOtelContext(OTEL_TEST_ENDPOINT, {
+      traces: true,
+      metrics: true,
+      logs: true,
+    });
+    await service.start(enabledCtx);
+
+    sdkCtor.mockClear();
+    sdkStart.mockClear();
+    logExporterCtor.mockClear();
+    const deniedCtx = createOtelContext(OTEL_TEST_ENDPOINT, {
+      traces: true,
+      metrics: true,
+      logs: true,
+    });
+    delete deniedCtx.internalDiagnostics;
+
+    await service.start(deniedCtx);
+    await service.stop?.(deniedCtx);
+
+    expect(deniedCtx.logger.error).toHaveBeenCalledWith(
+      "diagnostics-otel: internal diagnostics capability unavailable",
+    );
+    expect(sdkCtor).not.toHaveBeenCalled();
+    expect(sdkStart).not.toHaveBeenCalled();
+    expect(logExporterCtor).not.toHaveBeenCalled();
+    expect(sdkShutdown).toHaveBeenCalledOnce();
+    expect(logShutdown).toHaveBeenCalledOnce();
+  });
+
   test("does not retain an OTLP exporter handler when startup setup fails", async () => {
     const startupError = new Error("trace exporter setup failed");
     traceExporterCtor.mockImplementationOnce(() => {

extensions/diagnostics-otel/src/service.ts

@@ -67,14 +67,28 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
     stopActiveTrustedSpans = null;
     unregisterUnhandledRejectionHandler = null;
 
-    currentUnregisterUnhandledRejectionHandler?.();
-    currentUnsubscribe?.();
-    currentStopActiveTrustedSpans?.();
-    if (currentLogProvider) {
-      await currentLogProvider.shutdown().catch(() => undefined);
+    const settle = async (...stops: Array<(() => void | Promise<void>) | null>) =>
+      (
+        await Promise.allSettled(stops.map((stop) => Promise.resolve().then(() => stop?.())))
+      ).flatMap((result) => (result.status === "rejected" ? [result.reason] : []));
+    // Preserve cleanup -> provider flush -> handler removal while attempting every step per phase.
+    const failures = await settle(currentUnsubscribe, currentStopActiveTrustedSpans);
+    failures.push(
+      ...(await settle(
+        currentLogProvider ? () => currentLogProvider.shutdown() : null,
+        currentSdk ? () => currentSdk.shutdown() : null,
+      )),
+      ...(await settle(currentUnregisterUnhandledRejectionHandler)),
+    );
+
+    if (failures.length === 1) {
+      throw failures[0];
     }
-    if (currentSdk) {
-      await currentSdk.shutdown().catch(() => undefined);
+    if (failures.length > 1) {
+      throw new AggregateError(
+        failures,
+        `diagnostics-otel shutdown failed: ${failures.join("; ")}`,
+      );
     }
   };
 
@@ -128,6 +142,12 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
       if (enabledSignals.length === 0) {
         return;
       }
+      // This capability is the admission gate; no exporter may outlive a denied start.
+      const subscribe = ctx.internalDiagnostics?.onEvent;
+      if (!subscribe) {
+        ctx.logger.error("diagnostics-otel: internal diagnostics capability unavailable");
+        return;
+      }
 
       const envProtocol = process.env.OTEL_EXPORTER_OTLP_PROTOCOL;
       const protocol = otel.protocol ?? (envProtocol?.trim() ? envProtocol : "http/protobuf");
@@ -308,11 +328,6 @@ export function createDiagnosticsOtelService(): OpenClawPluginService {
         ...createModelRecorders(recorderRuntime),
         ...createToolAndSystemRecorders(recorderRuntime),
       };
-      const subscribe = ctx.internalDiagnostics?.onEvent;
-      if (!subscribe) {
-        ctx.logger.error("diagnostics-otel: internal diagnostics capability unavailable");
-        return;
-      }
 
       unsubscribe = subscribe(
         createDiagnosticsEventHandler({

src/infra/diagnostic-events.test.ts

@@ -630,6 +630,46 @@ describe("diagnostic-events", () => {
     expect(events).toHaveLength(250);
   });
 
+  it("does not extend a drain barrier for events queued after it starts", async () => {
+    const callIds: string[] = [];
+    onDiagnosticEvent((event) => {
+      if (event.type === "model.call.started") {
+        callIds.push(event.callId);
+      }
+    });
+
+    emitDiagnosticEvent({
+      type: "model.call.started",
+      runId: "run-before-barrier",
+      callId: "before-barrier",
+      provider: "openai",
+      model: "gpt-5.4",
+    });
+    const drained = waitForDiagnosticEventsDrained();
+    for (let index = 0; index < 250; index += 1) {
+      emitDiagnosticEvent({
+        type: "model.call.started",
+        runId: `run-after-${index}`,
+        callId: `after-${index}`,
+        provider: "openai",
+        model: "gpt-5.4",
+      });
+    }
+
+    await drained;
+
+    expect(callIds).toHaveLength(100);
+    expect(callIds[0]).toBe("before-barrier");
+    expect(
+      hasPendingInternalDiagnosticEvent(
+        (event) => event.type === "model.call.started" && event.callId === "after-249",
+      ),
+    ).toBe(true);
+
+    await waitForDiagnosticEventsDrained();
+    expect(callIds).toHaveLength(251);
+  });
+
   it("reports pending async diagnostic events before they drain", async () => {
     emitTrustedDiagnosticEvent({
       type: "tool.execution.error",

src/infra/diagnostic-events.ts

@@ -1216,10 +1216,15 @@ function dispatchAsyncDiagnosticDropSummary(state: DiagnosticEventsGlobalState):
   dispatchDiagnosticEvent(state, event, createInternalDiagnosticMetadata(false));
 }
 
-/** Waits until queued async diagnostic events have been delivered to listeners. */
+/** Waits until async diagnostic events queued when called are no longer pending. */
 export async function waitForDiagnosticEventsDrained(): Promise<void> {
   const state = getDiagnosticEventsState();
-  while (state.asyncDrainScheduled || state.asyncQueue.length > 0) {
+  const targetSeq = state.asyncQueue.at(-1)?.event.seq;
+  if (targetSeq === undefined) {
+    return;
+  }
+  // The queue is append-ordered by seq, so a newer head means this snapshot drained or dropped.
+  while ((state.asyncQueue[0]?.event.seq ?? Number.POSITIVE_INFINITY) <= targetSeq) {
     await new Promise<void>((resolve) => {
       setImmediate(resolve);
     });

src/plugins/services.test.ts

@@ -17,6 +17,11 @@ vi.mock("../logging/subsystem.js", () => ({
 }));
 
 import { STATE_DIR } from "../config/paths.js";
+import {
+  emitTrustedDiagnosticEvent,
+  resetDiagnosticEventsForTest,
+  waitForDiagnosticEventsDrained,
+} from "../infra/diagnostic-events.js";
 import { queuePluginSessionsChanged, subscribePluginSessionsChanged } from "./gateway-events.js";
 import { registerPluginHttpRoute } from "./http-registry.js";
 import { resetPluginRuntimeStateForTest, setActivePluginRegistry } from "./runtime.js";
@@ -142,6 +147,7 @@ function createTrackingService(
 describe("startPluginServices", () => {
   beforeEach(() => {
     vi.clearAllMocks();
+    resetDiagnosticEventsForTest();
     resetPluginRuntimeStateForTest();
   });
 
@@ -165,6 +171,82 @@ describe("startPluginServices", () => {
     expectServiceLifecycleState({ starts, stops, contexts, config });
   });
 
+  it("drains producer diagnostics before exporters stop and propagates exporter failures", async () => {
+    const order: string[] = [];
+    const producerError = new Error("producer stop failed");
+    const exporterError = new Error("exporter stop failed");
+    let unsubscribe: () => void = () => undefined;
+    const registry = createRegistry(
+      [
+        {
+          id: "producer",
+          start: () => undefined,
+          stop: () => {
+            order.push("producer");
+            emitTrustedDiagnosticEvent({
+              type: "log.record",
+              level: "INFO",
+              message: "queued during producer shutdown",
+            });
+            throw producerError;
+          },
+        },
+      ],
+      "plugin:test",
+      "workspace",
+    );
+    registry.services.push(
+      ...createRegistry(
+        [
+          {
+            id: "diagnostics-prometheus",
+            start: () => undefined,
+            stop: () => {
+              order.push("prometheus");
+            },
+          },
+        ],
+        "diagnostics-prometheus",
+        "bundled",
+      ).services,
+      ...createRegistry(
+        [
+          {
+            id: "diagnostics-otel",
+            start: (ctx) => {
+              unsubscribe = ctx.internalDiagnostics!.onEvent((event) => {
+                if (event.type === "log.record") {
+                  order.push("event");
+                }
+              });
+            },
+            stop: () => {
+              order.push("otel");
+              unsubscribe();
+              throw exporterError;
+            },
+          },
+        ],
+        "diagnostics-otel",
+        "bundled",
+      ).services,
+    );
+
+    const handle = await startPluginServices({
+      registry,
+      config: createServiceConfig(),
+    });
+
+    await expect(handle.stop()).rejects.toBe(exporterError);
+    await waitForDiagnosticEventsDrained();
+
+    expect(order).toEqual(["producer", "event", "otel", "prometheus"]);
+    expect(mockedLogger.warn.mock.calls).toEqual([
+      ["plugin service stop failed (producer): Error: producer stop failed"],
+      ["plugin service stop failed (diagnostics-otel): Error: exporter stop failed"],
+    ]);
+  });
+
   it("rolls back partially started services before starting their siblings", async () => {
     const acquired = new Set<string>();
     const received = vi.fn();
@@ -474,10 +556,15 @@ describe("startPluginServices", () => {
     await handle.stop();
   });
 
-  it("logs start/stop failures and continues", async () => {
+  it("attempts every ordinary service stop and preserves warn-and-continue failures", async () => {
     const stopOk = vi.fn();
-    const stopThrows = vi.fn(() => {
-      throw new Error("stop failed");
+    const firstError = new Error("first stop failed");
+    const secondError = new Error("second stop failed");
+    const stopFirst = vi.fn(() => {
+      throw firstError;
+    });
+    const stopSecond = vi.fn(() => {
+      throw secondError;
     });
 
     const handle = await startTrackingServices({
@@ -486,12 +573,13 @@ describe("startPluginServices", () => {
           failOnStart: true,
           stopSpy: vi.fn(),
         }),
+        createTrackingService("service-stop-first", { stopSpy: stopFirst }),
         createTrackingService("service-ok", { stopSpy: stopOk }),
-        createTrackingService("service-stop-fail", { stopSpy: stopThrows }),
+        createTrackingService("service-stop-second", { stopSpy: stopSecond }),
       ],
     });
 
-    await handle.stop();
+    await expect(handle.stop()).resolves.toBeUndefined();
 
     expect(mockedLogger.error.mock.calls).toEqual([
       [
@@ -500,10 +588,12 @@ describe("startPluginServices", () => {
     ]);
     expect(requireLoggerErrorMessage()).not.toContain("\n");
     expect(mockedLogger.warn.mock.calls).toEqual([
-      ["plugin service stop failed (service-stop-fail): Error: stop failed"],
+      ["plugin service stop failed (service-stop-second): Error: second stop failed"],
+      ["plugin service stop failed (service-stop-first): Error: first stop failed"],
     ]);
     expect(stopOk).toHaveBeenCalledOnce();
-    expect(stopThrows).toHaveBeenCalledOnce();
+    expect(stopFirst).toHaveBeenCalledOnce();
+    expect(stopSecond).toHaveBeenCalledOnce();
   });
 
   it("continues starting siblings when rollback also fails", async () => {

src/plugins/services.ts

@@ -5,6 +5,7 @@ import type { GatewayPluginEventBroadcastFn } from "../gateway/server-broadcast-
 import {
   emitTrustedDiagnosticEventWithPrivateData,
   onTrustedInternalDiagnosticEvent,
+  waitForDiagnosticEventsDrained,
 } from "../infra/diagnostic-events.js";
 import { createSubsystemLogger } from "../logging/subsystem.js";
 import { subscribePluginSessionsChanged } from "./gateway-events.js";
@@ -171,16 +172,18 @@ export async function startPluginServices(params: {
 }): Promise<PluginServicesHandle> {
   const running: Array<{
     id: string;
+    diagnosticsExporter: boolean;
     stop?: () => void | Promise<void>;
     revokeGatewayEvents: () => void;
   }> = [];
-  const stopService = async (entry: (typeof running)[number]) => {
+  const stopService = async (entry: (typeof running)[number], failures?: unknown[]) => {
     try {
       if (entry.stop) {
         await withPluginHttpRouteRegistry(params.registry, () => entry.stop?.());
       }
     } catch (err) {
       log.warn(`plugin service stop failed (${entry.id}): ${String(err)}`);
+      failures?.push(err);
     } finally {
       entry.revokeGatewayEvents();
     }
@@ -202,6 +205,7 @@ export async function startPluginServices(params: {
     });
     const runningService = {
       id: service.id,
+      diagnosticsExporter: serviceContext.internalDiagnostics !== undefined,
       stop: service.stop ? () => service.stop?.(serviceContext) : undefined,
       revokeGatewayEvents: scopedGatewayEvents.revoke,
     };
@@ -235,8 +239,30 @@ export async function startPluginServices(params: {
     stop: () =>
       // Store the shared promise before plugin cleanup runs so shutdown cannot start twice.
       (stopPromise ??= Promise.resolve().then(async () => {
-        for (const entry of running.toReversed()) {
-          await stopService(entry);
+        const reversed = running.toReversed();
+        const diagnosticsExporters = reversed.filter((entry) => entry.diagnosticsExporter);
+        const exporterFailures: unknown[] = [];
+        const stopServices = async (services: typeof reversed, failures?: unknown[]) => {
+          for (const entry of services) {
+            await stopService(entry, failures);
+          }
+        };
+        await stopServices(reversed.filter((entry) => !entry.diagnosticsExporter));
+        if (diagnosticsExporters.length > 0) {
+          // Producers stop first; this barrier preserves their queued tail before exporters detach.
+          await waitForDiagnosticEventsDrained();
+        }
+        // Ordinary plugin cleanup stays warn-and-continue. Trusted diagnostics
+        // exporter failures propagate because they can mean telemetry was lost.
+        await stopServices(diagnosticsExporters, exporterFailures);
+        if (exporterFailures.length === 1) {
+          throw exporterFailures[0];
+        }
+        if (exporterFailures.length > 1) {
+          throw new AggregateError(
+            exporterFailures,
+            "multiple diagnostics exporters failed to stop",
+          );
         }
       })),
   };