fix(channels): attribute zero-payload turn warnings and stop false silent-drop alarms (#116486)

* fix: attribute zero-count channel turn warnings

* fix(replies): preserve routed delivery across abort

---------

Co-authored-by: Altay <altay@hey.com>

src/auto-reply/reply/dispatch-from-config.abort-and-dedupe.test-utils.ts

@@ -37,6 +37,7 @@ import {
   globalBeforeAll0,
   describe0BeforeEach0,
 } from "./dispatch-from-config.test-harness.js";
+import { withDispatchProcessedOutcomeSink } from "./dispatch-processed-outcome.js";
 import { buildTestCtx } from "./test-ctx.js";
 
 function setupResolvedAcpSessionNotice(params: { bound: boolean; messageThreadId?: string }) {
@@ -1451,6 +1452,32 @@ describe("dispatchReplyFromConfig", () => {
     expect(duplicateReplyResolver).not.toHaveBeenCalled();
   });
 
+  it("attributes the processed outcome on completed and duplicate returns", async () => {
+    setNoAbort();
+    const cfg = emptyConfig;
+    const ctx = buildTestCtx({
+      Provider: "whatsapp",
+      OriginatingChannel: "whatsapp",
+      OriginatingTo: "whatsapp:+15555550123",
+      AccountId: "default",
+      MessageSid: "msg-duplicate-attributed",
+    });
+    const replyResolver = vi.fn(async () => ({ text: "hi" }) as ReplyPayload);
+
+    const first = await withDispatchProcessedOutcomeSink(() =>
+      dispatchReplyFromConfig({ ctx, cfg, replyResolver, dispatcher: createDispatcher() }),
+    );
+    const duplicate = await withDispatchProcessedOutcomeSink(() =>
+      dispatchReplyFromConfig({ ctx, cfg, replyResolver, dispatcher: createDispatcher() }),
+    );
+
+    expect(replyResolver).toHaveBeenCalledTimes(1);
+    expect(first.processedOutcome).toEqual({ outcome: "completed" });
+    // The duplicate skip queues nothing; the sink must name the branch so the
+    // kernel's zero-count warning is attributable to a benign dedupe hit.
+    expect(duplicate.processedOutcome).toEqual({ outcome: "skipped", reason: "duplicate" });
+  });
+
   it("keeps message-tool-only delivery mode on duplicate inbound returns", async () => {
     setNoAbort();
     const cfg = {

src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts

@@ -658,7 +658,7 @@ describe("dispatchReplyFromConfig ACP abort", () => {
     });
     expect(diagnosticMocks.logMessageProcessed).toHaveBeenCalledWith(
       expect.objectContaining({
-        outcome: "completed",
+        outcome: "skipped",
         reason: "reply_operation_aborted",
       }),
     );
@@ -1095,7 +1095,7 @@ describe("dispatchReplyFromConfig ACP abort", () => {
     });
     expect(diagnosticMocks.logMessageProcessed).toHaveBeenCalledWith(
       expect.objectContaining({
-        outcome: "completed",
+        outcome: "skipped",
         reason: "reply_operation_aborted",
       }),
     );

src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts

@@ -11,6 +11,7 @@ import {
   runWithDiagnosticTraceContext,
 } from "../../infra/diagnostic-trace-context.js";
 import type { SessionBindingRecord } from "../../infra/outbound/session-binding-service.js";
+import type { PluginTargetedInboundClaimOutcome } from "../../plugins/hooks.test-fixtures.js";
 import { createTestRegistry } from "../../test-utils/channel-plugins.js";
 import { getReplyPayloadMetadata, setReplyPayloadMetadata } from "../reply-payload.js";
 import type { MsgContext } from "../templating.js";
@@ -40,6 +41,7 @@ import {
   globalBeforeAll0,
   describe0BeforeEach0,
 } from "./dispatch-from-config.test-harness.js";
+import { withDispatchProcessedOutcomeSink } from "./dispatch-processed-outcome.js";
 import { getPreparedReplyDispatchRuntime } from "./prepared-reply-dispatch-context.js";
 import { usesFullReplyRuntime } from "./reply-config-runtime-mode.js";
 import { createReplyDispatcher } from "./reply-dispatcher.js";
@@ -278,7 +280,11 @@ describe("dispatchReplyFromConfig", () => {
 
     const result = await dispatchReplyFromConfig({ ctx, cfg, dispatcher, replyResolver });
 
-    expect(result).toEqual({ queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } });
+    expect(result).toEqual({
+      queuedFinal: false,
+      counts: { tool: 0, block: 0, final: 0 },
+      observedReplyDelivery: true,
+    });
     expect(dispatcher.sendFinalReply).toHaveBeenCalledWith({ text: "Codex native reply" });
     expect(
       getReplyPayloadMetadata(
@@ -291,6 +297,93 @@ describe("dispatchReplyFromConfig", () => {
     expect(replyResolver).not.toHaveBeenCalled();
   });
 
+  it("aborts plugin-bound completion while reply delivery is still settling", async () => {
+    setNoAbort();
+    hookMocks.runner.hasHooks.mockImplementation(
+      ((hookName?: string) =>
+        hookName === "inbound_claim" || hookName === "message_received") as () => boolean,
+    );
+    hookMocks.registry.plugins = [{ id: "codex", status: "loaded" }];
+    hookMocks.runner.runInboundClaimForPluginOutcome.mockResolvedValue({
+      status: "handled",
+      result: { handled: true, reply: { text: "Codex native reply" } },
+    });
+    sessionBindingMocks.resolveByConversation.mockReturnValue({
+      bindingId: "binding-reply-abort-1",
+      targetSessionKey: "plugin-binding:codex:reply-abort-123",
+      targetKind: "session",
+      conversation: {
+        channel: "discord",
+        accountId: "default",
+        conversationId: "channel:1481858418548412579",
+      },
+      status: "active",
+      boundAt: 1710000000000,
+      metadata: {
+        pluginBindingOwner: "plugin",
+        pluginId: "codex",
+        pluginRoot: "/plugins/codex",
+      },
+    } satisfies SessionBindingRecord);
+    let markDeliveryStarted: (() => void) | undefined;
+    let releaseDelivery: (() => void) | undefined;
+    const deliveryStarted = new Promise<void>((resolve) => {
+      markDeliveryStarted = resolve;
+    });
+    const deliveryRelease = new Promise<void>((resolve) => {
+      releaseDelivery = resolve;
+    });
+    const dispatcher = createReplyDispatcher({
+      deliver: async () => {
+        markDeliveryStarted?.();
+        await deliveryRelease;
+        throw new Error("delivery failed after abort");
+      },
+    });
+    const abortController = new AbortController();
+    const replyResolver = vi.fn(async () => ({ text: "should not run" }) satisfies ReplyPayload);
+
+    const dispatch = withDispatchProcessedOutcomeSink(() =>
+      dispatchReplyFromConfig({
+        ctx: buildTestCtx({
+          Provider: "discord",
+          Surface: "discord",
+          OriginatingChannel: "discord",
+          OriginatingTo: "discord:channel:1481858418548412579",
+          To: "discord:channel:1481858418548412579",
+          AccountId: "default",
+          SenderId: "user-9",
+          SenderUsername: "ada",
+          CommandAuthorized: true,
+          WasMentioned: false,
+          CommandBody: "who are you",
+          RawBody: "who are you",
+          Body: "who are you",
+          MessageSid: "msg-claim-plugin-reply-abort",
+          SessionKey: "agent:main:discord:channel:1481858418548412579",
+        }),
+        cfg: emptyConfig,
+        dispatcher,
+        replyOptions: { abortSignal: abortController.signal },
+        replyResolver,
+      }),
+    );
+
+    await deliveryStarted;
+    abortController.abort();
+    try {
+      const { result, processedOutcome } = await dispatch;
+
+      expect(result).toEqual({ queuedFinal: false, counts: { tool: 0, block: 0, final: 1 } });
+      expect(processedOutcome).toEqual({ outcome: "skipped", reason: "reply_operation_aborted" });
+      expect(replyResolver).not.toHaveBeenCalled();
+    } finally {
+      releaseDelivery?.();
+      dispatcher.markComplete();
+      await dispatcher.waitForIdle();
+    }
+  });
+
   it("persists Gateway plugin-bound turns and routed replies in the binding session", async () => {
     setNoAbort();
     hookMocks.runner.hasHooks.mockImplementation(
@@ -373,7 +466,11 @@ describe("dispatchReplyFromConfig", () => {
       replyResolver,
     });
 
-    expect(result).toEqual({ queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } });
+    expect(result).toEqual({
+      queuedFinal: false,
+      counts: { tool: 0, block: 0, final: 0 },
+      observedReplyDelivery: true,
+    });
     expect(persistApproved).toHaveBeenCalledWith({
       target: expect.objectContaining({
         sessionId: "bound-session-id",
@@ -438,7 +535,11 @@ describe("dispatchReplyFromConfig", () => {
       replyResolver,
     });
 
-    expect(rotatedResult).toEqual({ queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } });
+    expect(rotatedResult).toEqual({
+      queuedFinal: false,
+      counts: { tool: 0, block: 0, final: 0 },
+      observedReplyDelivery: true,
+    });
     const rotatedRoutedCall = firstMockArg(mocks.routeReply, "rotated plugin binding route") as {
       payload: ReplyPayload;
       sessionKey: string;
@@ -500,6 +601,179 @@ describe("dispatchReplyFromConfig", () => {
     expect(blockedDispatcher.sendFinalReply).not.toHaveBeenCalled();
   });
 
+  it.each([
+    {
+      name: "handled reply route delivers",
+      claimOutcome: {
+        status: "handled",
+        result: { handled: true, reply: { text: "Codex routed reply" } },
+      },
+      routeResult: { ok: true, delivered: true, messageId: "routed-binding-1" },
+      processedReason: "plugin-bound-handled",
+      expectObservedDelivery: true,
+    },
+    {
+      name: "handled reply route delivers before abort",
+      claimOutcome: {
+        status: "handled",
+        result: { handled: true, reply: { text: "Codex routed reply" } },
+      },
+      routeResult: { ok: true, delivered: true, messageId: "routed-binding-aborted-1" },
+      processedReason: "reply_operation_aborted",
+      expectObservedDelivery: true,
+      abortAfterRoute: true,
+    },
+    {
+      name: "handled reply route is hook-suppressed",
+      claimOutcome: {
+        status: "handled",
+        result: { handled: true, reply: { text: "Codex routed reply" } },
+      },
+      routeResult: { ok: true, delivered: false, suppressed: true },
+      processedReason: "plugin-bound-handled",
+      expectObservedDelivery: false,
+    },
+    {
+      name: "handled reply route fails",
+      claimOutcome: {
+        status: "handled",
+        result: { handled: true, reply: { text: "Codex routed reply" } },
+      },
+      routeResult: { ok: false, delivered: false, error: "transport down" },
+      processedReason: "plugin-bound-handled",
+      expectObservedDelivery: false,
+    },
+    {
+      name: "declined notice route delivers",
+      claimOutcome: { status: "declined" },
+      routeResult: { ok: true, delivered: true, messageId: "routed-declined-1" },
+      processedReason: "plugin-bound-declined",
+      expectObservedDelivery: true,
+    },
+    {
+      name: "declined notice route is hook-suppressed",
+      claimOutcome: { status: "declined" },
+      routeResult: { ok: true, delivered: false, suppressed: true },
+      processedReason: "plugin-bound-declined",
+      expectObservedDelivery: false,
+    },
+    {
+      name: "declined notice route fails",
+      claimOutcome: { status: "declined" },
+      routeResult: { ok: false, delivered: false, error: "transport down" },
+      processedReason: "plugin-bound-declined",
+      expectObservedDelivery: false,
+    },
+    {
+      name: "error notice route delivers",
+      claimOutcome: { status: "error", error: "boom" },
+      routeResult: { ok: true, delivered: true, messageId: "routed-error-1" },
+      processedReason: "plugin-bound-error",
+      expectObservedDelivery: true,
+    },
+    {
+      name: "error notice route is hook-suppressed",
+      claimOutcome: { status: "error", error: "boom" },
+      routeResult: { ok: true, delivered: false, suppressed: true },
+      processedReason: "plugin-bound-error",
+      expectObservedDelivery: false,
+    },
+    {
+      name: "error notice route fails",
+      claimOutcome: { status: "error", error: "boom" },
+      routeResult: { ok: false, delivered: false, error: "transport down" },
+      processedReason: "plugin-bound-error",
+      expectObservedDelivery: false,
+    },
+  ] satisfies Array<{
+    name: string;
+    claimOutcome: PluginTargetedInboundClaimOutcome;
+    routeResult: {
+      ok: boolean;
+      delivered: boolean;
+      messageId?: string;
+      suppressed?: boolean;
+      error?: string;
+    };
+    processedReason: string;
+    expectObservedDelivery: boolean;
+    abortAfterRoute?: boolean;
+  }>)(
+    "attests observed delivery only when the routed binding turn delivered: $name",
+    async (params) => {
+      setNoAbort();
+      hookMocks.runner.hasHooks.mockImplementation(
+        ((hookName?: string) =>
+          hookName === "inbound_claim" || hookName === "message_received") as () => boolean,
+      );
+      hookMocks.registry.plugins = [{ id: "openclaw-codex-app-server", status: "loaded" }];
+      hookMocks.runner.runInboundClaimForPluginOutcome.mockResolvedValue(params.claimOutcome);
+      const abortController = new AbortController();
+      mocks.routeReply.mockImplementation(async () => {
+        if (params.abortAfterRoute) {
+          abortController.abort();
+        }
+        return params.routeResult;
+      });
+      sessionBindingMocks.resolveByConversation.mockReturnValue({
+        bindingId: "binding-routed-attest-1",
+        targetSessionKey: "plugin-binding:codex:routed-attest",
+        targetKind: "session",
+        conversation: {
+          channel: "slack",
+          accountId: "default",
+          conversationId: "user:U123",
+        },
+        status: "active",
+        boundAt: 1710000000000,
+        metadata: {
+          pluginBindingOwner: "plugin",
+          pluginId: "openclaw-codex-app-server",
+          pluginRoot: "/plugins/codex",
+        },
+      } satisfies SessionBindingRecord);
+      const dispatcher = createDispatcher();
+      const replyResolver = vi.fn(async () => ({ text: "should not run" }) satisfies ReplyPayload);
+
+      const { result, processedOutcome } = await withDispatchProcessedOutcomeSink(() =>
+        dispatchReplyFromConfig({
+          ctx: buildTestCtx({
+            Provider: "openclaw",
+            Surface: "openclaw",
+            OriginatingChannel: "slack",
+            OriginatingTo: "user:U123",
+            To: "user:U123",
+            AccountId: "default",
+            CommandAuthorized: true,
+            Body: "continue",
+            RawBody: "continue",
+            MessageSid: `msg-routed-attest-${params.name.replace(/\s+/g, "-")}`,
+            SessionKey: "agent:main:main",
+          }),
+          cfg: emptyConfig,
+          dispatcher,
+          replyOptions: { abortSignal: abortController.signal },
+          replyResolver,
+        }),
+      );
+
+      // A hook-suppressed or failed route reached no recipient, so the result
+      // must stay warning-eligible instead of reading as a visible delivery.
+      expect(result).toEqual({
+        queuedFinal: false,
+        counts: { tool: 0, block: 0, final: 0 },
+        ...(params.expectObservedDelivery ? { observedReplyDelivery: true } : {}),
+      });
+      expect(processedOutcome).toEqual({
+        outcome: params.abortAfterRoute ? "skipped" : "completed",
+        reason: params.processedReason,
+      });
+      expect(mocks.routeReply).toHaveBeenCalledTimes(1);
+      expect(dispatcher.sendFinalReply).not.toHaveBeenCalled();
+      expect(replyResolver).not.toHaveBeenCalled();
+    },
+  );
+
   it("routes plugin-owned Discord DM bindings to the owning plugin before generic inbound claim broadcast", async () => {
     setNoAbort();
     hookMocks.runner.hasHooks.mockImplementation(

src/auto-reply/reply/dispatch-from-config.gather.ts

@@ -44,6 +44,7 @@ import {
 } from "./dispatch-from-config.runtime-loaders.js";
 import { createReplyHotPathTimingTracker } from "./dispatch-from-config.timing.js";
 import type { DispatchFromConfigParams } from "./dispatch-from-config.types.js";
+import { noteDispatchProcessedOutcome } from "./dispatch-processed-outcome.js";
 import { resolveEffectiveReplyRoute } from "./effective-reply-route.js";
 import type { ReplySessionBinding } from "./get-reply.types.js";
 import { finalizeInboundContext, isFinalizedInboundContext } from "./inbound-context.js";
@@ -96,6 +97,7 @@ export async function gatherDispatchRequest(
   const replyOperationRunState: ReplyOperationRunState =
     resolveReplyOperationRunState(normalizedParams.replyOptions) ?? {};
   if (params.replyOptions?.abortSignal?.aborted) {
+    noteDispatchProcessedOutcome({ outcome: "skipped", reason: "reply_operation_aborted" });
     messageAuditTerminal?.note("skipped", { reason: "reply_operation_aborted" });
     return {
       status: "complete" as const,
@@ -158,6 +160,10 @@ export async function gatherDispatchRequest(
   let agentDispatchStartedAt = 0;
 
   const recordProcessed = (outcome: DispatchProcessedOutcome, opts?: DispatchProcessedOptions) => {
+    noteDispatchProcessedOutcome({
+      outcome,
+      ...(opts?.reason !== undefined ? { reason: opts.reason } : {}),
+    });
     messageAuditTerminal?.note(outcome, opts);
     if (diagnosticsEnabled) {
       replyHotPathTiming.logIfSlow({
@@ -248,6 +254,7 @@ export async function gatherDispatchRequest(
     dispatchOperationSessionKey &&
     initialDispatchReplyOperation
   ) {
+    noteDispatchProcessedOutcome({ outcome: "skipped", reason: "reply-operation-active" });
     messageAuditTerminal?.note("skipped", { reason: "reply-operation-active" });
     return {
       status: "complete" as const,

src/auto-reply/reply/dispatch-from-config.hooks-and-send-policy.test-utils.ts

@@ -2015,6 +2015,7 @@ describe("sendPolicy deny — suppress delivery, not processing (#53328)", () =>
         queuedFinal: false,
         counts: { tool: 0, block: 0, final: 0 },
         sourceReplyDeliveryMode: "message_tool_only",
+        ...(params.expectPluginReplyDelivered ? { observedReplyDelivery: true } : {}),
       });
       expect(sessionBindingMocks.touch).toHaveBeenCalledWith(params.bindingId);
       expect(hookMocks.runner.runInboundClaimForPluginOutcome).toHaveBeenCalledWith(

src/auto-reply/reply/dispatch-from-config.lifecycle-and-bindings.test-utils.ts

@@ -48,6 +48,7 @@ import {
   globalBeforeAll0,
   describe0BeforeEach0,
 } from "./dispatch-from-config.test-harness.js";
+import { withDispatchProcessedOutcomeSink } from "./dispatch-processed-outcome.js";
 import { finalizeInboundContextForSdk } from "./inbound-context.js";
 import { buildTestCtx } from "./test-ctx.js";
 
@@ -1881,15 +1882,20 @@ describe("dispatchReplyFromConfig", () => {
     });
     const replyResolver = vi.fn(async () => ({ text: "should not run" }) satisfies ReplyPayload);
 
-    const result = await dispatchReplyFromConfig({
-      ctx,
-      cfg,
-      dispatcher,
-      replyOptions: { abortSignal: abortController.signal },
-      replyResolver,
-    });
+    const { result, processedOutcome } = await withDispatchProcessedOutcomeSink(() =>
+      dispatchReplyFromConfig({
+        ctx,
+        cfg,
+        dispatcher,
+        replyOptions: { abortSignal: abortController.signal },
+        replyResolver,
+      }),
+    );
 
     expect(result).toEqual({ queuedFinal: false, counts: { tool: 0, block: 0, final: 0 } });
+    // The aborted skip queues nothing; the sink must name the branch so the
+    // kernel's zero-count warning is attributable to a benign abort.
+    expect(processedOutcome).toEqual({ outcome: "skipped", reason: "reply_operation_aborted" });
     expect(sessionBindingMocks.touch).not.toHaveBeenCalled();
     expect(hookMocks.runner.runInboundClaimForPluginOutcome).not.toHaveBeenCalled();
     expect(replyResolver).not.toHaveBeenCalled();

src/auto-reply/reply/dispatch-from-config.prepare-context.ts

@@ -479,12 +479,13 @@ export async function prepareDispatchOperationContext(state: PrepareDispatchDeli
     } else {
       commitInboundDedupeIfClaimed();
     }
-    recordProcessed("completed", { reason: "reply_operation_aborted" });
+    recordProcessed("skipped", { reason: "reply_operation_aborted" });
     markIdle("message_completed");
     state.completeDispatchReplyOperation();
     return attachSourceReplyDeliveryMode({
       queuedFinal,
       counts: dispatcher.getQueuedCounts(),
+      ...(state.turnLedger.hasVisibleDelivery() ? { observedReplyDelivery: true } : {}),
     });
   };
 

src/auto-reply/reply/dispatch-from-config.prepare-operation.ts

@@ -48,6 +48,7 @@ export async function prepareDispatchOperation(state: PrepareDispatchOperationCo
     sessionKey,
     sessionStoreEntry,
     suppressDelivery,
+    turnLedger,
   } = state;
   const abortRuntime = params.fastAbortResolver ? null : await loadAbortRuntime();
   const fastAbortResolver = params.fastAbortResolver ?? abortRuntime?.tryFastAbortFromMessage;
@@ -165,6 +166,17 @@ export async function prepareDispatchOperation(state: PrepareDispatchOperationCo
     };
   }
 
+  const settlePluginBindingDeliveryVisibility = async () => {
+    const settlement = await turnLedger.settleQueued(state.getPreDispatchAbortSignal());
+    if (settlement === "aborted" || isPreDispatchOperationAborted()) {
+      return { status: "aborted" as const };
+    }
+    return {
+      status: "ready" as const,
+      observedReplyDelivery: turnLedger.hasVisibleDelivery(),
+    };
+  };
+
   if (pluginOwnedBinding) {
     if (isPreDispatchOperationAborted()) {
       return { status: "complete" as const, result: finishReplyOperationAbortedDispatch() };
@@ -250,15 +262,24 @@ export async function prepareDispatchOperation(state: PrepareDispatchOperationCo
               transcriptOwner,
             );
           }
+          const deliveryVisibility = await settlePluginBindingDeliveryVisibility();
+          if (deliveryVisibility.status === "aborted") {
+            return { status: "complete" as const, result: finishReplyOperationAbortedDispatch() };
+          }
           markIdle("plugin_binding_dispatch");
           recordProcessed("completed", { reason: "plugin-bound-handled" });
           commitInboundDedupeIfClaimed();
           completeDispatchReplyOperation();
           return {
             status: "complete" as const,
+            // Routed binding deliveries bypass the dispatcher counters, so the
+            // ledger's settled visibility keeps a delivered reply from reading as
+            // a silent zero-count turn. A hook-suppressed or failed route never
+            // reached the recipient, so it must keep the warning eligible.
             result: attachSourceReplyDeliveryMode({
               queuedFinal: false,
               counts: dispatcher.getQueuedCounts(),
+              ...(deliveryVisibility.observedReplyDelivery ? { observedReplyDelivery: true } : {}),
             }),
           };
         }
@@ -305,6 +326,10 @@ export async function prepareDispatchOperation(state: PrepareDispatchOperationCo
             "terminal",
             transcriptOwner,
           );
+          const deliveryVisibility = await settlePluginBindingDeliveryVisibility();
+          if (deliveryVisibility.status === "aborted") {
+            return { status: "complete" as const, result: finishReplyOperationAbortedDispatch() };
+          }
           markIdle("plugin_binding_declined");
           recordProcessed("completed", { reason: "plugin-bound-declined" });
           commitInboundDedupeIfClaimed();
@@ -314,6 +339,7 @@ export async function prepareDispatchOperation(state: PrepareDispatchOperationCo
             result: attachSourceReplyDeliveryMode({
               queuedFinal: false,
               counts: dispatcher.getQueuedCounts(),
+              ...(deliveryVisibility.observedReplyDelivery ? { observedReplyDelivery: true } : {}),
             }),
           };
         }
@@ -327,6 +353,10 @@ export async function prepareDispatchOperation(state: PrepareDispatchOperationCo
             "terminal",
             transcriptOwner,
           );
+          const deliveryVisibility = await settlePluginBindingDeliveryVisibility();
+          if (deliveryVisibility.status === "aborted") {
+            return { status: "complete" as const, result: finishReplyOperationAbortedDispatch() };
+          }
           markIdle("plugin_binding_error");
           recordProcessed("completed", { reason: "plugin-bound-error" });
           commitInboundDedupeIfClaimed();
@@ -336,6 +366,7 @@ export async function prepareDispatchOperation(state: PrepareDispatchOperationCo
             result: attachSourceReplyDeliveryMode({
               queuedFinal: false,
               counts: dispatcher.getQueuedCounts(),
+              ...(deliveryVisibility.observedReplyDelivery ? { observedReplyDelivery: true } : {}),
             }),
           };
         }

src/auto-reply/reply/dispatch-from-config.reply-dispatch.test.ts

@@ -245,12 +245,7 @@ describe("dispatchReplyFromConfig reply_dispatch hook", () => {
 
     const result = await dispatchReplyFromConfig({
       ctx: createHookCtx(),
-      cfg: {
-        ...emptyConfig,
-        session: {
-          sendPolicy: { default: "deny" },
-        },
-      },
+      cfg: { ...emptyConfig, session: { sendPolicy: { default: "deny" } } },
       dispatcher: createDispatcher(),
       replyResolver: async () => ({ text: "model reply" }),
     });

src/auto-reply/reply/dispatch-processed-outcome.ts

@@ -0,0 +1,40 @@
+// Channel-internal seam carrying a dispatch's terminal processed outcome to the turn kernel.
+import { AsyncLocalStorage } from "node:async_hooks";
+import { resolveGlobalSingleton } from "../../shared/global-singleton.js";
+import type { DispatchProcessedOutcome } from "./dispatch-from-config.audit.js";
+
+/** Terminal outcome recorded while dispatching; names the branch that ended the turn. */
+export type DispatchProcessedNote = {
+  outcome: DispatchProcessedOutcome;
+  reason?: string;
+};
+
+type DispatchProcessedOutcomeSink = { current?: DispatchProcessedNote };
+
+const DISPATCH_PROCESSED_OUTCOME_SINK_KEY: unique symbol = Symbol.for(
+  "openclaw.dispatchProcessedOutcomeSink",
+);
+
+const dispatchProcessedOutcomeSink = resolveGlobalSingleton<
+  AsyncLocalStorage<DispatchProcessedOutcomeSink>
+>(DISPATCH_PROCESSED_OUTCOME_SINK_KEY, () => new AsyncLocalStorage<DispatchProcessedOutcomeSink>());
+
+/**
+ * Runs a channel turn's dispatch under a sink so its terminal outcome can attribute
+ * zero-count warnings without widening the plugin-visible dispatch result contract.
+ */
+export async function withDispatchProcessedOutcomeSink<T>(
+  run: () => Promise<T>,
+): Promise<{ result: T; processedOutcome?: DispatchProcessedNote }> {
+  const sink: DispatchProcessedOutcomeSink = {};
+  const result = await dispatchProcessedOutcomeSink.run(sink, run);
+  return { result, processedOutcome: sink.current };
+}
+
+/** Records the dispatch's terminal outcome for the surrounding channel turn, if any. */
+export function noteDispatchProcessedOutcome(note: DispatchProcessedNote): void {
+  const sink = dispatchProcessedOutcomeSink.getStore();
+  if (sink) {
+    sink.current = note;
+  }
+}

src/channels/turn/execution.ts

@@ -1,3 +1,7 @@
+import {
+  withDispatchProcessedOutcomeSink,
+  type DispatchProcessedNote,
+} from "../../auto-reply/reply/dispatch-processed-outcome.js";
 import { clearChannelHistoryIfEnabled } from "../../auto-reply/reply/history.js";
 import type { FinalizedMsgContext } from "../../auto-reply/templating.js";
 import {
@@ -88,6 +92,7 @@ function maybeWarnZeroCountVisibleDispatch<TDispatchResult>(
     "admission" | "channel" | "ctxPayload" | "messageId" | "routeSessionKey"
   > & {
     dispatchResult: TDispatchResult;
+    processedOutcome?: DispatchProcessedNote;
     log?: (event: ChannelTurnLogEvent) => void;
   },
 ): void {
@@ -102,11 +107,19 @@ function maybeWarnZeroCountVisibleDispatch<TDispatchResult>(
   if (hasVisibleChannelTurnDispatch(dispatchResult, NO_ADDITIONAL_DELIVERY_SIGNALS)) {
     return;
   }
+  // The processed outcome names the dispatch branch that produced the silence,
+  // so operators can tell a benign duplicate or busy skip from a lost message.
+  // It stays in this core-owned log line; the channel log event is a plugin
+  // contract and must not widen.
+  const processed = params.processedOutcome;
+  const cause = processed
+    ? `${processed.outcome}${processed.reason ? `:${processed.reason}` : ""}`
+    : undefined;
   log.warn(
     `visible channel turn dispatched with no queued reply payloads: channel=${params.channel} ` +
       `messageId=${params.messageId ?? "unknown"} sessionKey=${
         params.ctxPayload.SessionKey ?? params.routeSessionKey
-      }`,
+      } cause=${cause ?? "unknown"}`,
   );
   emit({
     ...params,
@@ -321,14 +334,21 @@ async function runPreparedChannelTurnCoreInTrace<
     } else if (admission.kind === "observeOnly") {
       await params.runDispatchLifecycle?.onDispatchSkipped("observeOnly");
     }
-    dispatchResult =
-      admission.kind === "observeOnly"
-        ? resolveObserveOnlyDispatchResult(params)
-        : await params.runDispatch();
+    let processedOutcome: DispatchProcessedNote | undefined;
+    if (admission.kind === "observeOnly") {
+      dispatchResult = resolveObserveOnlyDispatchResult(params);
+    } else {
+      // The sink carries the dispatch's terminal outcome to the warning below
+      // without widening the plugin-visible dispatch result contract.
+      ({ result: dispatchResult, processedOutcome } = await withDispatchProcessedOutcomeSink(() =>
+        params.runDispatch(),
+      ));
+    }
     maybeWarnZeroCountVisibleDispatch({
       ...params,
       admission,
       dispatchResult,
+      processedOutcome,
     });
   } catch (err) {
     emit({

src/channels/turn/run-channel-turn.pipeline.test.ts

@@ -1,6 +1,7 @@
 // Channel turn pipeline tests cover orchestration, dispatch, and completion behavior.
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import type { ReplyPayload } from "../../auto-reply/reply-payload.js";
+import { noteDispatchProcessedOutcome } from "../../auto-reply/reply/dispatch-processed-outcome.js";
 import type { DispatchReplyWithBufferedBlockDispatcher } from "../../auto-reply/reply/provider-dispatcher.types.js";
 import { createReplyDispatcher } from "../../auto-reply/reply/reply-dispatcher.js";
 import { getReplySystemEventSessionKey } from "../../auto-reply/reply/system-event-session-key.js";
@@ -25,7 +26,7 @@ import { outboundMessageIdentities } from "../message/outbound-echo-state.js";
 import type { RecordInboundSession } from "../session.types.js";
 import { runPreparedChannelTurn } from "./execution.js";
 import { dispatchAssembledChannelTurn } from "./lifecycle.js";
-import type { ChannelTurnResult } from "./types.js";
+import type { ChannelTurnResult, PreparedChannelTurn } from "./types.js";
 
 const deliverOutboundPayloads = vi.hoisted(() => vi.fn());
 const resolveOutboundDurableFinalDeliverySupport = vi.hoisted(() => vi.fn());
@@ -39,6 +40,27 @@ const createMessageSentEmitter = vi.hoisted(() =>
   vi.fn(() => ({ emitMessageSent, hasMessageSentHooks: true })),
 );
 const readRecentUserAssistantTextForSession = vi.hoisted(() => vi.fn());
+const subsystemWarn = vi.hoisted(() => vi.fn());
+
+vi.mock("../../logging/subsystem.js", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("../../logging/subsystem.js")>();
+  const makeLogger = (subsystem: string): import("../../logging/subsystem.js").SubsystemLogger => ({
+    subsystem,
+    isEnabled: () => true,
+    trace: vi.fn(),
+    debug: vi.fn(),
+    info: vi.fn(),
+    warn: subsystemWarn,
+    error: vi.fn(),
+    fatal: vi.fn(),
+    raw: vi.fn(),
+    child: (name: string) => makeLogger(`${subsystem}/${name}`),
+  });
+  return {
+    ...actual,
+    createSubsystemLogger: makeLogger,
+  };
+});
 
 vi.mock("../../auto-reply/reply/provider-dispatcher.js", async (importOriginal) => {
   const actual =
@@ -169,6 +191,20 @@ function dispatchTestAssembledTurn(
   });
 }
 
+function runTestPreparedChannelTurn<TDispatchResult>(
+  params: Pick<PreparedChannelTurn<TDispatchResult>, "runDispatch" | "log" | "messageId">,
+) {
+  return runPreparedChannelTurn({
+    channel: "test",
+    routeSessionKey: "agent:main:test:peer",
+    storePath: "/tmp/sessions.json",
+    ctxPayload: createCtx(),
+    recordInboundSession: createRecordInboundSession(),
+    record: { onRecordError: vi.fn() },
+    ...params,
+  });
+}
+
 type TurnLogEvent = {
   event?: string;
   messageId?: string;
@@ -911,26 +947,16 @@ describe("channel turn pipeline", () => {
   });
 
   it("logs a warning when a visible prepared dispatch queues no payloads", async () => {
-    const events: string[] = [];
     const log = vi.fn();
-    const recordInboundSession = createRecordInboundSession(events);
     const runDispatch = vi.fn(async () => ({
       queuedFinal: false,
       counts: { tool: 0, block: 0, final: 0 },
     }));
 
-    const result = await runPreparedChannelTurn({
-      channel: "test",
-      routeSessionKey: "agent:main:test:peer",
-      storePath: "/tmp/sessions.json",
-      ctxPayload: createCtx(),
-      recordInboundSession,
+    const result = await runTestPreparedChannelTurn({
       runDispatch,
       log,
       messageId: "msg-zero",
-      record: {
-        onRecordError: vi.fn(),
-      },
     });
 
     expectDispatched(result);
@@ -943,6 +969,38 @@ describe("channel turn pipeline", () => {
         reason: "zero-count-visible-dispatch",
       }),
     ]);
+    // A dispatch that recorded no processed outcome reads as unknown in the
+    // core-owned warn line.
+    expect(subsystemWarn).toHaveBeenCalledWith(expect.stringContaining("cause=unknown"));
+  });
+
+  it("attributes the zero-count warn line with the dispatch's processed outcome", async () => {
+    const log = vi.fn();
+    // The dispatch pipeline records its terminal branch through the kernel's
+    // sink instead of widening the plugin-visible result contract.
+    const runDispatch = vi.fn(async () => {
+      noteDispatchProcessedOutcome({ outcome: "skipped", reason: "duplicate" });
+      return {
+        queuedFinal: false,
+        counts: { tool: 0, block: 0, final: 0 },
+      };
+    });
+
+    const result = await runTestPreparedChannelTurn({
+      runDispatch,
+      log,
+      messageId: "msg-zero-cause",
+    });
+
+    expectDispatched(result);
+    expect(subsystemWarn).toHaveBeenCalledWith(expect.stringContaining("messageId=msg-zero-cause"));
+    expect(subsystemWarn).toHaveBeenCalledWith(expect.stringContaining("cause=skipped:duplicate"));
+    // The channel log event is a plugin contract; the attribution must stay out of it.
+    const warning = log.mock.calls
+      .map(([event]) => event as Record<string, unknown>)
+      .find((event) => event.reason === "zero-count-visible-dispatch");
+    expect(warning).toBeDefined();
+    expect(warning).not.toHaveProperty("cause");
   });
 
   it.each([
@@ -966,12 +1024,7 @@ describe("channel turn pipeline", () => {
   ])("$name", async ({ dispatchResult, warns }) => {
     const log = vi.fn();
 
-    await runPreparedChannelTurn({
-      channel: "test",
-      routeSessionKey: "agent:main:test:peer",
-      storePath: "/tmp/sessions.json",
-      ctxPayload: createCtx(),
-      recordInboundSession: createRecordInboundSession(),
+    await runTestPreparedChannelTurn({
       runDispatch: vi.fn(async () => dispatchResult),
       log,
       messageId: "msg-compat",
@@ -983,9 +1036,7 @@ describe("channel turn pipeline", () => {
   });
 
   it("does not warn for observed-path deliveries with zero queued counts", async () => {
-    const events: string[] = [];
     const log = vi.fn();
-    const recordInboundSession = createRecordInboundSession(events);
     // Observed-delivery path: queuedFinal false and all counts zero, but the reply was
     // delivered via observedReplyDelivery and must not trip the silent-drop sentinel.
     const runDispatch = vi.fn(async () => ({
@@ -994,18 +1045,10 @@ describe("channel turn pipeline", () => {
       observedReplyDelivery: true,
     }));
 
-    const result = await runPreparedChannelTurn({
-      channel: "test",
-      routeSessionKey: "agent:main:test:peer",
-      storePath: "/tmp/sessions.json",
-      ctxPayload: createCtx(),
-      recordInboundSession,
+    const result = await runTestPreparedChannelTurn({
       runDispatch,
       log,
       messageId: "msg-observed",
-      record: {
-        onRecordError: vi.fn(),
-      },
     });
 
     expectDispatched(result);
@@ -1016,27 +1059,17 @@ describe("channel turn pipeline", () => {
   });
 
   it("does not warn when an active run accepts deferred steer ownership", async () => {
-    const events: string[] = [];
     const log = vi.fn();
-    const recordInboundSession = createRecordInboundSession(events);
     const runDispatch = vi.fn(async () => ({
       queuedFinal: false,
       counts: { tool: 0, block: 0, final: 0 },
       deferredToActiveRun: "steer" as const,
     }));
 
-    const result = await runPreparedChannelTurn({
-      channel: "test",
-      routeSessionKey: "agent:main:test:peer",
-      storePath: "/tmp/sessions.json",
-      ctxPayload: createCtx(),
-      recordInboundSession,
+    const result = await runTestPreparedChannelTurn({
       runDispatch,
       log,
       messageId: "msg-deferred-steer",
-      record: {
-        onRecordError: vi.fn(),
-      },
     });
 
     expectDispatched(result);