diff --git a/src/auto-reply/reply/dispatch-from-config.abort-and-dedupe.test-utils.ts b/src/auto-reply/reply/dispatch-from-config.abort-and-dedupe.test-utils.ts index 3d4e4c140667..304ed0e33539 100644 --- a/src/auto-reply/reply/dispatch-from-config.abort-and-dedupe.test-utils.ts +++ b/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 = { diff --git a/src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts b/src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts index 43603277a022..47d41eed117d 100644 --- a/src/auto-reply/reply/dispatch-from-config.acp-abort.test.ts +++ b/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", }), ); diff --git a/src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts b/src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts index 57eda182f00a..e8c4bc7df790 100644 --- a/src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts +++ b/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((resolve) => { + markDeliveryStarted = resolve; + }); + const deliveryRelease = new Promise((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( diff --git a/src/auto-reply/reply/dispatch-from-config.gather.ts b/src/auto-reply/reply/dispatch-from-config.gather.ts index 0ccf468ce298..055a243265d7 100644 --- a/src/auto-reply/reply/dispatch-from-config.gather.ts +++ b/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, diff --git a/src/auto-reply/reply/dispatch-from-config.hooks-and-send-policy.test-utils.ts b/src/auto-reply/reply/dispatch-from-config.hooks-and-send-policy.test-utils.ts index f2e7b6aa3567..242c57015351 100644 --- a/src/auto-reply/reply/dispatch-from-config.hooks-and-send-policy.test-utils.ts +++ b/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( diff --git a/src/auto-reply/reply/dispatch-from-config.lifecycle-and-bindings.test-utils.ts b/src/auto-reply/reply/dispatch-from-config.lifecycle-and-bindings.test-utils.ts index 832538b9bbad..460512f733de 100644 --- a/src/auto-reply/reply/dispatch-from-config.lifecycle-and-bindings.test-utils.ts +++ b/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(); diff --git a/src/auto-reply/reply/dispatch-from-config.prepare-context.ts b/src/auto-reply/reply/dispatch-from-config.prepare-context.ts index 8f6f75103efa..a73bbb7ee192 100644 --- a/src/auto-reply/reply/dispatch-from-config.prepare-context.ts +++ b/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 } : {}), }); }; diff --git a/src/auto-reply/reply/dispatch-from-config.prepare-operation.ts b/src/auto-reply/reply/dispatch-from-config.prepare-operation.ts index d5bf5b039af9..e7a0a8643dbc 100644 --- a/src/auto-reply/reply/dispatch-from-config.prepare-operation.ts +++ b/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 } : {}), }), }; } diff --git a/src/auto-reply/reply/dispatch-from-config.reply-dispatch.test.ts b/src/auto-reply/reply/dispatch-from-config.reply-dispatch.test.ts index 4ea911dc54ea..89808778a1e0 100644 --- a/src/auto-reply/reply/dispatch-from-config.reply-dispatch.test.ts +++ b/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" }), }); diff --git a/src/auto-reply/reply/dispatch-processed-outcome.ts b/src/auto-reply/reply/dispatch-processed-outcome.ts new file mode 100644 index 000000000000..de3829854904 --- /dev/null +++ b/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 +>(DISPATCH_PROCESSED_OUTCOME_SINK_KEY, () => new AsyncLocalStorage()); + +/** + * 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( + run: () => Promise, +): 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; + } +} diff --git a/src/channels/turn/execution.ts b/src/channels/turn/execution.ts index 7e15b6b31f84..69ca370f0e91 100644 --- a/src/channels/turn/execution.ts +++ b/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( "admission" | "channel" | "ctxPayload" | "messageId" | "routeSessionKey" > & { dispatchResult: TDispatchResult; + processedOutcome?: DispatchProcessedNote; log?: (event: ChannelTurnLogEvent) => void; }, ): void { @@ -102,11 +107,19 @@ function maybeWarnZeroCountVisibleDispatch( 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({ diff --git a/src/channels/turn/run-channel-turn.pipeline.test.ts b/src/channels/turn/run-channel-turn.pipeline.test.ts index 1cc8a3c872eb..9b09002a994d 100644 --- a/src/channels/turn/run-channel-turn.pipeline.test.ts +++ b/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(); + 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( + params: Pick, "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) + .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);