From fb8e1aa59d90bf749658abc75abd20b364510e8a Mon Sep 17 00:00:00 2001 From: Vito Cappello Date: Tue, 28 Jul 2026 20:50:31 -0400 Subject: [PATCH] fix(auto-reply): prevent false no-reply fallbacks from routed and accepted turns (#115016) * fix(auto-reply): observe routed reset hook replies * fix(auto-reply): track settled route delivery * fix(auto-reply): preserve reasoning route suppression * fix(auto-reply): suppress fallback for accepted busy turns * fix(auto-reply): preserve editable partial delivery ids * test(auto-reply): align route delivery mock * fix(auto-reply): preserve routed delivery semantics * docs(auto-reply): clarify lost adoption ownership * test(auto-reply): split routed delivery evidence cases * fix(auto-reply): preserve ambiguous route sends * fix(auto-reply): omit non-delivery sentinel ids --------- Co-authored-by: Ayaan Zaidi --- ...agent-runner-direct-runtime-config.test.ts | 8 + src/auto-reply/reply/agent-runner-run.ts | 15 +- .../agent-runner.runreplyagent.e2e.test.ts | 3 + .../reply/commands-private-route.ts | 5 +- .../reply/commands-reset-hooks.test.ts | 89 ++++++- src/auto-reply/reply/commands-reset-hooks.ts | 8 +- src/auto-reply/reply/commands-reset.ts | 2 + .../reply/dispatch-acp-delivery.test.ts | 42 +++- src/auto-reply/reply/dispatch-acp-delivery.ts | 9 +- src/auto-reply/reply/dispatch-acp.test.ts | 35 ++- .../reply/dispatch-from-config.execute.ts | 3 + .../reply/dispatch-from-config.finalize.ts | 4 + .../reply/dispatch-from-config.gather.ts | 7 + ...config.hooks-and-send-policy.test-utils.ts | 52 +++- ...onfig.lifecycle-and-bindings.test-utils.ts | 8 +- .../dispatch-from-config.prepare-delivery.ts | 5 +- ...ispatch-from-config.reply-dispatch.test.ts | 6 +- ...ispatch-from-config.shared.test-harness.ts | 9 +- ...ispatch-from-config.stale-recovery.test.ts | 2 +- ...atch-from-config.terminal-recovery.test.ts | 2 +- .../dispatch-from-config.test-harness.ts | 6 +- .../reply/followup-delivery.test.ts | 63 ++++- src/auto-reply/reply/followup-delivery.ts | 16 +- .../get-reply.reset-hooks-fallback.test.ts | 14 +- src/auto-reply/reply/get-reply.ts | 1 + src/auto-reply/reply/get-reply.types.ts | 4 +- .../reply/reply-operation-run-state.ts | 3 +- .../reply/route-reply.delivery-result.test.ts | 226 ++++++++++++++++++ src/auto-reply/reply/route-reply.test.ts | 11 +- src/auto-reply/reply/route-reply.ts | 99 +++++++- 30 files changed, 692 insertions(+), 65 deletions(-) create mode 100644 src/auto-reply/reply/route-reply.delivery-result.test.ts diff --git a/src/auto-reply/reply/agent-runner-direct-runtime-config.test.ts b/src/auto-reply/reply/agent-runner-direct-runtime-config.test.ts index 86ddf841c75d..7b1f7ea2427c 100644 --- a/src/auto-reply/reply/agent-runner-direct-runtime-config.test.ts +++ b/src/auto-reply/reply/agent-runner-direct-runtime-config.test.ts @@ -9,6 +9,10 @@ import type { TemplateContext } from "../templating.js"; import { SILENT_REPLY_TOKEN } from "../tokens.js"; import { createTestFollowupRun } from "./agent-runner.test-fixtures.js"; import type { QueueSettings } from "./queue.js"; +import { + REPLY_OPERATION_RUN_STATE, + type ReplyOperationRunState, +} from "./reply-operation-run-state.js"; import type { ReplyOperation } from "./reply-run-registry.js"; import { createMockTypingController } from "./test-helpers.js"; @@ -604,9 +608,13 @@ describe("runReplyAgent runtime config", () => { shouldFollowup: true, isActive: true, }); + const runState: ReplyOperationRunState = {}; + replyParams.opts = { [REPLY_OPERATION_RUN_STATE]: runState }; + enqueueFollowupRunMock.mockReturnValueOnce(true); await expect(runReplyAgent(replyParams)).resolves.toBeUndefined(); + expect(runState.admission).toEqual({ status: "accepted", mode: "followup" }); expect(resolveQueuedReplyExecutionConfigMock).not.toHaveBeenCalled(); expect(enqueueFollowupRunMock).toHaveBeenCalledTimes(1); const enqueueCall = enqueueFollowupRunMock.mock.calls.at(0); diff --git a/src/auto-reply/reply/agent-runner-run.ts b/src/auto-reply/reply/agent-runner-run.ts index 63ae56ce5e87..0c54b16ced95 100644 --- a/src/auto-reply/reply/agent-runner-run.ts +++ b/src/auto-reply/reply/agent-runner-run.ts @@ -294,14 +294,22 @@ export async function runReplyAgent( }, ); if (steerOutcome.queued) { + if (replyOperationRunState) { + // Transcript commit has already transferred this turn to the active + // session. Keep that acceptance even if ingress adoption is later lost: + // the losing dispatch must neither replay nor emit its own fallback. + replyOperationRunState.admission = { status: "accepted", mode: "steer" }; + } activeReplyOperation?.recordActivity(); try { await turnAdoptionLifecycle?.onAdopted(); } catch (error) { if (isIngressAdoptionLostError(error)) { // Claim was tombstoned/superseded/guillotined after transcript commit. - // Cancel the active run so steered tools do not keep executing; do not - // rethrow — replaying ingress would duplicate the injected user turn. + // Cancel the active run so steered tools do not keep executing. Keep + // admission accepted and do not rethrow: ingress ownership is gone, + // so replay or a local no-visible-reply fallback would duplicate or + // misreport the already-injected user turn. const abortKey = sessionKey ?? queueKey; if (abortKey) { replyRunRegistry.abort(abortKey); @@ -387,6 +395,9 @@ export async function runReplyAgent( typing.cleanup(); return undefined; } + if (replyOperationRunState) { + replyOperationRunState.admission = { status: "accepted", mode: "followup" }; + } // The queue must stay dormant while the active owner can still collect // messages. Registering after enqueue closes the owner-clear race. const activeReplyOperation = replyRunRegistry.get(queueKey); diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts index e2400b6245ba..a0e69b6481e1 100644 --- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts +++ b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts @@ -437,12 +437,14 @@ function requireBuiltChannelSourceTurnId( describe("runReplyAgent active steering", () => { it("dispatches a declined steer once with its source-turn identity", async () => { + const runState: ReplyOperationRunState = {}; state.beforeAgentReplyHasHooksMock.mockImplementation( (hookName) => hookName === "before_agent_reply", ); state.beforeAgentReplyRunMock.mockResolvedValue(undefined); state.queueEmbeddedAgentMessageMock.mockReturnValueOnce(true); const { run, sourceTurnId } = createMinimalRun({ + opts: { [REPLY_OPERATION_RUN_STATE]: runState }, isActive: true, isStreaming: true, shouldSteer: true, @@ -464,6 +466,7 @@ describe("runReplyAgent active steering", () => { await expect(run()).resolves.toBeUndefined(); + expect(runState.admission).toEqual({ status: "accepted", mode: "steer" }); expect(state.beforeAgentReplyRunMock).toHaveBeenCalledOnce(); expect(state.beforeAgentReplyRunMock).toHaveBeenCalledWith( { cleanedBody: "hello" }, diff --git a/src/auto-reply/reply/commands-private-route.ts b/src/auto-reply/reply/commands-private-route.ts index 066b1c8c6b40..6365e656042f 100644 --- a/src/auto-reply/reply/commands-private-route.ts +++ b/src/auto-reply/reply/commands-private-route.ts @@ -107,7 +107,10 @@ export async function deliverPrivateCommandReply(params: { }), ), ); - return results.some((result) => result.status === "fulfilled" && result.value.ok); + return results.some( + (result) => + result.status === "fulfilled" && (result.value.delivered || result.value.suppressed === true), + ); } /** Reads the command message thread id from command context. */ diff --git a/src/auto-reply/reply/commands-reset-hooks.test.ts b/src/auto-reply/reply/commands-reset-hooks.test.ts index b90bc2547abf..0192fe173216 100644 --- a/src/auto-reply/reply/commands-reset-hooks.test.ts +++ b/src/auto-reply/reply/commands-reset-hooks.test.ts @@ -9,7 +9,14 @@ import { parseInlineDirectives } from "./directive-handling.parse.js"; const triggerInternalHookMock = vi.hoisted(() => vi.fn().mockResolvedValue(undefined)); const routeReplyMock = vi.hoisted(() => - vi.fn<(params: unknown) => Promise<{ ok: boolean }>>(async () => ({ ok: true })), + vi.fn< + (params: unknown) => Promise<{ + ok: boolean; + delivered: boolean; + messageId?: string; + suppressed?: boolean; + }> + >(async () => ({ ok: true, delivered: true, messageId: "reset-hook-1" })), ); const resetMocks = vi.hoisted(() => ({ resetConfiguredBindingTargetInPlace: vi.fn().mockResolvedValue({ ok: true as const }), @@ -146,6 +153,11 @@ describe("handleCommands reset hooks", () => { resetMocks.resetConfiguredBindingTargetInPlace.mockResolvedValue({ ok: true }); resetMocks.resolveBoundAcpThreadSessionKey.mockReturnValue(undefined); triggerInternalHookMock.mockResolvedValue(undefined); + routeReplyMock.mockResolvedValue({ + ok: true, + delivered: true, + messageId: "reset-hook-1", + }); }); afterEach(() => { @@ -289,6 +301,7 @@ describe("handleCommands reset hooks", () => { triggerInternalHookMock.mockImplementationOnce(async (event: { messages: string[] }) => { event.messages.push("Reset hook says hi"); }); + const onObservedReplyDelivery = vi.fn(); const params = buildResetParams( "/new", { @@ -305,6 +318,7 @@ describe("handleCommands reset hooks", () => { MessageThreadId: "thread-1", }, ); + params.opts = { onObservedReplyDelivery }; const result = await maybeHandleResetCommand(params); @@ -315,9 +329,82 @@ describe("handleCommands reset hooks", () => { requesterSenderE164: "+15551234567", threadId: "thread-1", }); + expect(onObservedReplyDelivery).toHaveBeenCalledOnce(); expect(result).toEqual({ shouldContinue: false }); }); + it.each([ + ["failed", { ok: false, delivered: false }], + ["dropped", { ok: true, delivered: false }], + ] as const)( + "falls back to the standard reset acknowledgement when the hook route is %s", + async (_name, routeResult) => { + triggerInternalHookMock.mockImplementationOnce(async (event: { messages: string[] }) => { + event.messages.push("Reset hook says hi"); + }); + routeReplyMock.mockResolvedValueOnce(routeResult); + const onObservedReplyDelivery = vi.fn(); + const params = buildResetParams("/new", { + commands: { text: true }, + channels: { whatsapp: { allowFrom: ["*"] } }, + } as OpenClawConfig); + params.opts = { onObservedReplyDelivery }; + + const result = await maybeHandleResetCommand(params); + + expect(onObservedReplyDelivery).not.toHaveBeenCalled(); + expect(result).toEqual({ + shouldContinue: false, + reply: { text: "✅ New session started." }, + }); + }, + ); + + it("keeps an intentionally suppressed reset hook route silent", async () => { + triggerInternalHookMock.mockImplementationOnce(async (event: { messages: string[] }) => { + event.messages.push("Reset hook says hi"); + }); + routeReplyMock.mockResolvedValueOnce({ + ok: true, + delivered: false, + suppressed: true, + }); + const onObservedReplyDelivery = vi.fn(); + const params = buildResetParams("/new", { + commands: { text: true }, + channels: { whatsapp: { allowFrom: ["*"] } }, + } as OpenClawConfig); + params.opts = { onObservedReplyDelivery }; + + const result = await maybeHandleResetCommand(params); + + expect(onObservedReplyDelivery).not.toHaveBeenCalled(); + expect(result).toEqual({ shouldContinue: false }); + }); + + it.each([ + ["without a provider message id", { ok: true, delivered: true }], + ["before a later partial failure", { ok: false, delivered: true, messageId: "reset-hook-1" }], + ] as const)( + "marks a reset hook route as observed when delivered %s", + async (_name, routeResult) => { + triggerInternalHookMock.mockImplementationOnce(async (event: { messages: string[] }) => { + event.messages.push("Reset hook says hi"); + }); + routeReplyMock.mockResolvedValueOnce(routeResult); + const onObservedReplyDelivery = vi.fn(); + const params = buildResetParams("/new", { + commands: { text: true }, + channels: { whatsapp: { allowFrom: ["*"] } }, + } as OpenClawConfig); + params.opts = { onObservedReplyDelivery }; + + await maybeHandleResetCommand(params); + + expect(onObservedReplyDelivery).toHaveBeenCalledOnce(); + }, + ); + it("prefers the target session entry when emitting reset hooks", async () => { const params = buildResetParams("/reset", { commands: { text: true }, diff --git a/src/auto-reply/reply/commands-reset-hooks.ts b/src/auto-reply/reply/commands-reset-hooks.ts index e27fe7bbbd6f..58e6e23f9152 100644 --- a/src/auto-reply/reply/commands-reset-hooks.ts +++ b/src/auto-reply/reply/commands-reset-hooks.ts @@ -81,6 +81,7 @@ export async function emitResetCommandHooks(params: { storePath?: string; sessionEntry?: HandleCommandsParams["sessionEntry"]; previousSessionEntry?: HandleCommandsParams["previousSessionEntry"]; + onObservedReplyDelivery?: () => Promise | void; workspaceDir: string; }): Promise<{ routedReply: boolean }> { const hookAgentId = @@ -114,7 +115,7 @@ export async function emitResetCommandHooks(params: { const to = params.ctx.OriginatingTo || params.command.from || params.command.to; if (channel && to) { const { routeReply } = await loadRouteReplyRuntime(); - await routeReply({ + const result = await routeReply({ payload: { text: hookEvent.messages.join("\n\n") }, channel, to, @@ -128,7 +129,10 @@ export async function emitResetCommandHooks(params: { cfg: params.cfg, replyKind: "final", }); - routedReply = true; + if (result.delivered) { + await params.onObservedReplyDelivery?.(); + } + routedReply = result.delivered || result.suppressed === true; } } diff --git a/src/auto-reply/reply/commands-reset.ts b/src/auto-reply/reply/commands-reset.ts index 05c0d16b5a7b..244379d9934e 100644 --- a/src/auto-reply/reply/commands-reset.ts +++ b/src/auto-reply/reply/commands-reset.ts @@ -106,6 +106,7 @@ export async function maybeHandleResetCommand( storePath: params.storePath, sessionEntry: targetSessionEntry, previousSessionEntry, + onObservedReplyDelivery: params.opts?.onObservedReplyDelivery, workspaceDir: params.workspaceDir, }); params.command.softResetTriggered = true; @@ -181,6 +182,7 @@ export async function maybeHandleResetCommand( storePath: params.storePath, sessionEntry: targetSessionEntry, previousSessionEntry: params.previousSessionEntry, + onObservedReplyDelivery: params.opts?.onObservedReplyDelivery, workspaceDir: params.workspaceDir, }); if (!resetTail) { diff --git a/src/auto-reply/reply/dispatch-acp-delivery.test.ts b/src/auto-reply/reply/dispatch-acp-delivery.test.ts index 8b795d2dd7c3..1486079d42f7 100644 --- a/src/auto-reply/reply/dispatch-acp-delivery.test.ts +++ b/src/auto-reply/reply/dispatch-acp-delivery.test.ts @@ -21,10 +21,12 @@ const deliveryMocks = vi.hoisted(() => ({ _params: unknown, ): Promise<{ ok: boolean; + delivered: boolean; messageId?: string; suppressed?: boolean; reason?: string; - }> => ({ ok: true, messageId: "mock-message" }), + error?: string; + }> => ({ ok: true, delivered: true, messageId: "mock-message" }), ), runMessageAction: vi.fn(async (_params: unknown) => ({ ok: true as const })), })); @@ -180,7 +182,11 @@ async function expectVisibleChatBlockRoutesToAccount( describe("createAcpDispatchDeliveryCoordinator", () => { beforeEach(() => { deliveryMocks.routeReply.mockClear(); - deliveryMocks.routeReply.mockResolvedValue({ ok: true, messageId: "mock-message" }); + deliveryMocks.routeReply.mockResolvedValue({ + ok: true, + delivered: true, + messageId: "mock-message", + }); deliveryMocks.runMessageAction.mockClear(); deliveryMocks.runMessageAction.mockResolvedValue({ ok: true as const }); channelPluginMocks.getChannelPlugin.mockClear(); @@ -1042,8 +1048,8 @@ describe("createAcpDispatchDeliveryCoordinator", () => { deliveryMocks.routeReply.mockImplementationOnce(async (paramsUnknown: unknown) => { const params = paramsUnknown as { abortSignal?: AbortSignal }; return params.abortSignal?.aborted - ? { ok: false, error: "Reply routing aborted" } - : { ok: true, messageId: "unexpected" }; + ? { ok: false, delivered: false, error: "Reply routing aborted" } + : { ok: true, delivered: true, messageId: "unexpected" }; }); const coordinator = createAcpDispatchDeliveryCoordinator({ cfg: createAcpTestConfig(), @@ -1070,9 +1076,37 @@ describe("createAcpDispatchDeliveryCoordinator", () => { await expect(coordinator.resolveAccumulatedDeliveredTranscriptText()).resolves.toBe(""); }); + it("does not retry routed ACP text after a partial delivery failure", async () => { + deliveryMocks.routeReply.mockResolvedValueOnce({ + ok: false, + delivered: true, + messageId: "visible-1", + error: "later chunk failed", + }); + const coordinator = createAcpDispatchDeliveryCoordinator({ + cfg: createAcpTestConfig(), + ctx: buildTestCtx({ + Provider: "visiblechat", + Surface: "visiblechat", + SessionKey: "agent:codex-acp:session-1", + }), + dispatcher: createDispatcher(), + inboundAudio: false, + shouldRouteToOriginating: true, + originatingChannel: "visiblechat", + originatingTo: "channel:thread-1", + }); + + const delivered = await coordinator.deliver("final", { text: "hello" }, { skipTts: true }); + + expect(delivered).toBe(true); + expect(coordinator.getRoutedCounts().final).toBe(1); + }); + it("treats hook-suppressed routed ACP block text as handled", async () => { deliveryMocks.routeReply.mockResolvedValueOnce({ ok: true, + delivered: false, suppressed: true, reason: "cancelled_by_reply_payload_sending_hook", }); diff --git a/src/auto-reply/reply/dispatch-acp-delivery.ts b/src/auto-reply/reply/dispatch-acp-delivery.ts index 63d4c7e6c4d3..7fcec94ce921 100644 --- a/src/auto-reply/reply/dispatch-acp-delivery.ts +++ b/src/auto-reply/reply/dispatch-acp-delivery.ts @@ -484,7 +484,7 @@ export function createAcpDispatchDeliveryCoordinator(params: { replyKind: kind, runId: params.runId, }); - if (!result.ok) { + if (!result.delivered && !result.suppressed) { if (tracksVisibleText) { state.failedVisibleTextDelivery = true; } @@ -502,6 +502,13 @@ export function createAcpDispatchDeliveryCoordinator(params: { } return true; } + if (!result.ok) { + logVerbose( + `dispatch-acp: route-reply (acp/${kind}) partially failed after delivery: ${ + result.error ?? "unknown error" + }`, + ); + } if (kind === "tool" && meta?.toolCallId && result.messageId) { state.toolMessageByCallId.set(meta.toolCallId, { channel: params.originatingChannel, diff --git a/src/auto-reply/reply/dispatch-acp.test.ts b/src/auto-reply/reply/dispatch-acp.test.ts index 9414ced3a0c2..9fd2362e3e3d 100644 --- a/src/auto-reply/reply/dispatch-acp.test.ts +++ b/src/auto-reply/reply/dispatch-acp.test.ts @@ -50,8 +50,13 @@ const policyMocks = vi.hoisted(() => ({ const routeMocks = vi.hoisted(() => ({ routeReply: vi.fn< - (_params: unknown) => Promise<{ ok: true; messageId: string } | { ok: false; error: string }> - >(async () => ({ ok: true, messageId: "mock" })), + ( + _params: unknown, + ) => Promise< + | { ok: true; delivered: boolean; messageId?: string } + | { ok: false; delivered: boolean; error: string } + > + >(async () => ({ ok: true, delivered: true, messageId: "mock" })), })); const channelPluginMocks = vi.hoisted(() => ({ @@ -485,7 +490,11 @@ describe("tryDispatchAcpReply", () => { policyMocks.resolveAcpAgentPolicyError.mockReset(); policyMocks.resolveAcpAgentPolicyError.mockReturnValue(null); routeMocks.routeReply.mockReset(); - routeMocks.routeReply.mockResolvedValue({ ok: true, messageId: "mock" }); + routeMocks.routeReply.mockResolvedValue({ + ok: true, + delivered: true, + messageId: "mock", + }); channelPluginMocks.getChannelPlugin.mockClear(); messageActionMocks.runMessageAction.mockReset(); messageActionMocks.runMessageAction.mockResolvedValue({ ok: true as const }); @@ -601,7 +610,11 @@ describe("tryDispatchAcpReply", () => { it("persists ACP transcript when routed delivery fails", async () => { setReadyAcpResolution(); mockRoutedTextTurn("hello"); - routeMocks.routeReply.mockResolvedValue({ ok: false, error: "missing channel adapter" }); + routeMocks.routeReply.mockResolvedValue({ + ok: false, + delivered: false, + error: "missing channel adapter", + }); await runDispatch({ bodyForAgent: "reply", @@ -733,7 +746,11 @@ describe("tryDispatchAcpReply", () => { it("edits ACP tool lifecycle updates in place when supported", async () => { setReadyAcpResolution(); mockToolLifecycleTurn("call-1"); - routeMocks.routeReply.mockResolvedValueOnce({ ok: true, messageId: "tool-msg-1" }); + routeMocks.routeReply.mockResolvedValueOnce({ + ok: true, + delivered: true, + messageId: "tool-msg-1", + }); const { dispatcher } = createDispatcher(); await runDispatch({ @@ -754,8 +771,12 @@ describe("tryDispatchAcpReply", () => { setReadyAcpResolution(); mockToolLifecycleTurn("call-2"); routeMocks.routeReply - .mockResolvedValueOnce({ ok: true, messageId: "tool-msg-2" }) - .mockResolvedValueOnce({ ok: true, messageId: "tool-msg-2-fallback" }); + .mockResolvedValueOnce({ ok: true, delivered: true, messageId: "tool-msg-2" }) + .mockResolvedValueOnce({ + ok: true, + delivered: true, + messageId: "tool-msg-2-fallback", + }); messageActionMocks.runMessageAction.mockRejectedValueOnce(new Error("edit unsupported")); const { dispatcher } = createDispatcher(); diff --git a/src/auto-reply/reply/dispatch-from-config.execute.ts b/src/auto-reply/reply/dispatch-from-config.execute.ts index 4566d609e2ee..27c1769e2f0e 100644 --- a/src/auto-reply/reply/dispatch-from-config.execute.ts +++ b/src/auto-reply/reply/dispatch-from-config.execute.ts @@ -21,6 +21,7 @@ import { import { extendPreparedDispatchState } from "./dispatch-from-config.phase-state.js"; import type { PrepareDispatchExecutionReadyState } from "./dispatch-from-config.prepare-execution.js"; import { waitForReplyDispatcherIdle } from "./reply-dispatcher.js"; +import { REPLY_OPERATION_RUN_STATE } from "./reply-operation-run-state.js"; export async function executeDispatch(state: PrepareDispatchExecutionReadyState) { const { @@ -74,6 +75,7 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState) recordRoutedBlockReplyDelivery, replyConfig, replyContextAccountId, + replyOperationRunState, replyResolver, replyRoute, resolveToolDeliveryPayload, @@ -125,6 +127,7 @@ export async function executeDispatch(state: PrepareDispatchExecutionReadyState) ctx, { ...getReplyOptions(), + [REPLY_OPERATION_RUN_STATE]: replyOperationRunState, sourceReplyDeliveryMode, sessionPromptSourceReplyDeliveryMode: sessionStableSourceReplyDeliveryMode, ...({ diff --git a/src/auto-reply/reply/dispatch-from-config.finalize.ts b/src/auto-reply/reply/dispatch-from-config.finalize.ts index c3fbe34fe17f..e6ed18fd21c7 100644 --- a/src/auto-reply/reply/dispatch-from-config.finalize.ts +++ b/src/auto-reply/reply/dispatch-from-config.finalize.ts @@ -51,6 +51,7 @@ export async function finalizeDispatchAndAudit(state: ExecuteDispatchReadyState) recordAgentDispatchCompleted, recordProcessed, replyResult, + replyOperationRunState, replyRoute, routeReplyToOriginating, sendFinalPayload, @@ -283,6 +284,7 @@ export async function finalizeDispatchAndAudit(state: ExecuteDispatchReadyState) // ledger intentionally does not own. Directedness gates both the fallback and // eligibility: only a turn that positively addressed the bot may surface a // visible failure notice. + const replyAcceptedByActiveRun = replyOperationRunState.admission?.status === "accepted"; const noVisibleReplyFallbackAllowed = () => noVisibleReplyFallbackDirected && !suppressDelivery && @@ -290,6 +292,7 @@ export async function finalizeDispatchAndAudit(state: ExecuteDispatchReadyState) sourceReplyDeliveryMode !== "message_tool_only" && !emptyFinalAllowedAsSilent && !getObservedReplyDelivery() && + !replyAcceptedByActiveRun && !turnLedger.hasVisibleDelivery() && !turnLedger.hasForeignQueuedAdmissions(); let queuedSettleResult: Awaited> = "settled"; @@ -386,6 +389,7 @@ export async function finalizeDispatchAndAudit(state: ExecuteDispatchReadyState) !turnLedger.hasVisibleDelivery() && !noVisibleReplyFallbackDelivered && !getObservedReplyDelivery() && + !replyAcceptedByActiveRun && !emptyFinalAllowedAsSilent ? { noVisibleReplyFallbackEligible: true } : {}), diff --git a/src/auto-reply/reply/dispatch-from-config.gather.ts b/src/auto-reply/reply/dispatch-from-config.gather.ts index 0f9c98c982aa..fb63d6a21bd2 100644 --- a/src/auto-reply/reply/dispatch-from-config.gather.ts +++ b/src/auto-reply/reply/dispatch-from-config.gather.ts @@ -46,6 +46,10 @@ import { resolveEffectiveReplyRoute } from "./effective-reply-route.js"; import type { ReplySessionBinding } from "./get-reply.types.js"; import { finalizeInboundContext, isFinalizedInboundContext } from "./inbound-context.js"; import { hasInboundAudio } from "./inbound-media.js"; +import { + resolveReplyOperationRunState, + type ReplyOperationRunState, +} from "./reply-operation-run-state.js"; import { replyRunRegistry } from "./reply-run-registry.js"; import { isReplyProfilerEnabled } from "./reply-timing-tracker.js"; import { resolveRoutedDeliveryThreadId } from "./routed-delivery-thread.js"; @@ -61,6 +65,8 @@ export async function gatherDispatchRequest( const normalizedParams = ctx === params.ctx ? params : { ...params, ctx }; const state = { params: normalizedParams, messageAuditTerminal }; const { cfg, dispatcher } = normalizedParams; + const replyOperationRunState: ReplyOperationRunState = + resolveReplyOperationRunState(normalizedParams.replyOptions) ?? {}; if (params.replyOptions?.abortSignal?.aborted) { messageAuditTerminal?.note("skipped", { reason: "reply_operation_aborted" }); return { @@ -454,6 +460,7 @@ export async function gatherDispatchRequest( inboundAudio, sessionTtsAuto, workspaceDir, + replyOperationRunState, completeDispatchReplyOperation, dispatchHookDispatcher, ensureDispatchReplyOperation, 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 3d9dceb45aa9..854f5e1af04b 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 @@ -32,6 +32,7 @@ import { } from "./dispatch-from-config.test-harness.js"; import { PROVIDER_CONVERSATION_STATE_ERROR_USER_MESSAGE } from "./provider-request-error-classifier.js"; import { createReplyDispatcher } from "./reply-dispatcher.js"; +import { resolveReplyOperationRunState } from "./reply-operation-run-state.js"; import { buildTestCtx } from "./test-ctx.js"; beforeAll(globalBeforeAll0); @@ -461,6 +462,37 @@ describe("sendPolicy deny — suppress delivery, not processing (#53328)", () => expect(result.noVisibleReplyFallbackEligible).toBeUndefined(); }); + it("does not treat an active-run accepted turn as an empty completion", async () => { + setNoAbort(); + const dispatcher = createDispatcher(); + const replyResolver = vi.fn(async (_ctx: MsgContext, opts?: GetReplyOptions) => { + const runState = resolveReplyOperationRunState(opts); + if (!runState) { + throw new Error("expected reply operation run state"); + } + runState.admission = { status: "accepted", mode: "followup" }; + return undefined; + }); + + const result = await dispatchReplyFromConfig({ + ctx: buildTestCtx({ + Surface: "telegram", + Provider: "telegram", + SessionKey: "agent:main:telegram:direct:test", + }), + cfg: emptyConfig, + dispatcher, + replyResolver, + }); + + expect(replyResolver).toHaveBeenCalledOnce(); + expect(dispatcher.sendFinalReply).not.toHaveBeenCalled(); + expect(result).toEqual({ + queuedFinal: false, + counts: { tool: 0, block: 0, final: 0 }, + }); + }); + it("keeps room_event turns silent even when silence policy is disallow", async () => { setNoAbort(); const dispatcher = createDispatcher(); @@ -734,7 +766,11 @@ describe("sendPolicy deny — suppress delivery, not processing (#53328)", () => it("delivers routed fallback when routing drops an empty final without sending", async () => { setNoAbort(); - mocks.routeReply.mockResolvedValue({ ok: true, messageId: "fallback-1" }); + mocks.routeReply.mockResolvedValueOnce({ ok: true, delivered: false }).mockResolvedValueOnce({ + ok: true, + delivered: true, + messageId: "fallback-1", + }); const dispatcher = createDispatcher(); const replyResolver = vi.fn(async () => ({ text: "" })); const ctx = buildTestCtx({ @@ -776,7 +812,7 @@ describe("sendPolicy deny — suppress delivery, not processing (#53328)", () => it("keeps eligibility when an empty routed final precedes a suppressed fallback", async () => { setNoAbort(); - mocks.routeReply.mockResolvedValue({ ok: true, suppressed: true }); + mocks.routeReply.mockResolvedValue({ ok: true, delivered: false, suppressed: true }); const dispatcher = createDispatcher(); const replyResolver = vi.fn(async () => ({ text: "" })); const ctx = buildTestCtx({ @@ -812,7 +848,7 @@ describe("sendPolicy deny — suppress delivery, not processing (#53328)", () => it("does not report a hook-suppressed routed fallback as delivered", async () => { setNoAbort(); - mocks.routeReply.mockResolvedValue({ ok: true, suppressed: true }); + mocks.routeReply.mockResolvedValue({ ok: true, delivered: false, suppressed: true }); const dispatcher = createDispatcher(); const replyResolver = vi.fn(async () => undefined); const ctx = buildTestCtx({ @@ -847,7 +883,11 @@ describe("sendPolicy deny — suppress delivery, not processing (#53328)", () => it("does not deliver no-visible fallback after a routed media-only block", async () => { setNoAbort(); - mocks.routeReply.mockResolvedValue({ ok: true, messageId: "media-block-1" }); + mocks.routeReply.mockResolvedValue({ + ok: true, + delivered: true, + messageId: "media-block-1", + }); const dispatcher = createDispatcher(); const replyResolver = vi.fn(async (_ctx: MsgContext, opts?: GetReplyOptions) => { await opts?.onBlockReply?.({ mediaUrl: "https://example.com/seatmap.png" }); @@ -1008,8 +1048,8 @@ describe("sendPolicy deny — suppress delivery, not processing (#53328)", () => mocks.routeReply.mockImplementation(async (paramsUnknown: unknown) => { const params = paramsUnknown as { payload?: { text?: string } }; return params.payload?.text === NO_VISIBLE_REPLY_FALLBACK_TEXT - ? { ok: true, messageId: "fallback-1" } - : { ok: true, suppressed: true }; + ? { ok: true, delivered: true, messageId: "fallback-1" } + : { ok: true, delivered: false, suppressed: true }; }); const dispatcher = createDispatcher(); const replyResolver = vi.fn(async () => ({ text: "real answer" })); 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 19f86b2198cc..c5cd058fb9c0 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 @@ -1276,10 +1276,12 @@ describe("dispatchReplyFromConfig", () => { const sessionKey = "agent:main:discord:channel:interrupted-fallback"; const sessionId = "interrupted-fallback-session"; sessionStoreMocks.currentEntry = { sessionId, updatedAt: Date.now() }; - let resolveNotice: ((result: { ok: true; messageId: string }) => void) | undefined; + let resolveNotice: + | ((result: { ok: true; delivered: true; messageId: string }) => void) + | undefined; mocks.routeReply.mockImplementationOnce( async () => - await new Promise<{ ok: true; messageId: string }>((resolve) => { + await new Promise<{ ok: true; delivered: true; messageId: string }>((resolve) => { resolveNotice = resolve; }), ); @@ -1329,7 +1331,7 @@ describe("dispatchReplyFromConfig", () => { }); expect(mutationRan).toBe(false); - resolveNotice?.({ ok: true, messageId: "fallback-notice" }); + resolveNotice?.({ ok: true, delivered: true, messageId: "fallback-notice" }); const result = await dispatch; await mutation; diff --git a/src/auto-reply/reply/dispatch-from-config.prepare-delivery.ts b/src/auto-reply/reply/dispatch-from-config.prepare-delivery.ts index 1bae51a2789f..4b8c10130231 100644 --- a/src/auto-reply/reply/dispatch-from-config.prepare-delivery.ts +++ b/src/auto-reply/reply/dispatch-from-config.prepare-delivery.ts @@ -187,8 +187,7 @@ export async function prepareDispatchDelivery(state: GatherDispatchRequestReadyS return result; }; - const isRoutedReplyDelivered = (result: { ok: boolean; suppressed?: boolean }) => - result.ok && result.suppressed !== true; + const isRoutedReplyDelivered = (result: { delivered: boolean }) => result.delivered; /** * Helper to send a payload via route-reply (async). @@ -260,7 +259,7 @@ export async function prepareDispatchDelivery(state: GatherDispatchRequestReadyS `dispatch-from-config: route-reply (plugin binding notice) failed: ${result.error ?? "unknown error"}`, ); } - return result.ok; + return result.delivered || result.suppressed === true; } markInboundDedupeReplayUnsafe(); return mode === "additive" 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 8eda9ee68f63..b2ca7558b545 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 @@ -89,7 +89,9 @@ describe("dispatchReplyFromConfig reply_dispatch hook", () => { resetReplyRunRegistry(); setDiscordTestRegistry(); resetInboundDedupe(); - mocks.routeReply.mockReset().mockResolvedValue({ ok: true, messageId: "mock" }); + mocks.routeReply + .mockReset() + .mockResolvedValue({ ok: true, delivered: true, messageId: "mock" }); mocks.tryFastAbortFromMessage.mockReset().mockResolvedValue({ handled: false, aborted: false, @@ -229,7 +231,7 @@ describe("dispatchReplyFromConfig reply_dispatch hook", () => { }), }; sessionStoreMocks.loadSessionStore.mockClear(); - mocks.routeReply.mockResolvedValue({ ok: true, messageId: "mock" }); + mocks.routeReply.mockResolvedValue({ ok: true, delivered: true, messageId: "mock" }); const deliver = vi.fn().mockResolvedValue(undefined); const dispatcher = createReplyDispatcher({ deliver }); diff --git a/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts b/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts index 973582c24056..7f613036d8ed 100644 --- a/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts +++ b/src/auto-reply/reply/dispatch-from-config.shared.test-harness.ts @@ -35,8 +35,15 @@ const mocks = vi.hoisted(() => ({ routeReply: vi.fn( async ( _params: unknown, - ): Promise<{ ok: boolean; messageId?: string; suppressed?: boolean; error?: string }> => ({ + ): Promise<{ + ok: boolean; + delivered: boolean; + messageId?: string; + suppressed?: boolean; + error?: string; + }> => ({ ok: true, + delivered: true, messageId: "mock", }), ), diff --git a/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts b/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts index 17e2e46ffc4e..a8db5158789a 100644 --- a/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.stale-recovery.test.ts @@ -61,7 +61,7 @@ describe("dispatchReplyFromConfig stale visible admission recovery", () => { resetPluginTtsAndThreadMocks(); runtimePluginMocks.ensureRuntimePluginsLoaded.mockReset(); mocks.routeReply.mockReset(); - mocks.routeReply.mockResolvedValue({ ok: true, messageId: "mock" }); + mocks.routeReply.mockResolvedValue({ ok: true, delivered: true, messageId: "mock" }); mocks.tryFastAbortFromMessage.mockReset(); setNoAbort(); diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockReset(); diff --git a/src/auto-reply/reply/dispatch-from-config.terminal-recovery.test.ts b/src/auto-reply/reply/dispatch-from-config.terminal-recovery.test.ts index 8e0423412674..0519e406e08d 100644 --- a/src/auto-reply/reply/dispatch-from-config.terminal-recovery.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.terminal-recovery.test.ts @@ -54,7 +54,7 @@ describe("dispatchReplyFromConfig terminal visible admission recovery", () => { resetPluginTtsAndThreadMocks(); runtimePluginMocks.ensureRuntimePluginsLoaded.mockReset(); mocks.routeReply.mockReset(); - mocks.routeReply.mockResolvedValue({ ok: true, messageId: "mock" }); + mocks.routeReply.mockResolvedValue({ ok: true, delivered: true, messageId: "mock" }); mocks.tryFastAbortFromMessage.mockReset(); mocks.tryFastAbortFromMessage.mockResolvedValue(noAbortResult); diagnosticMocks.requestStuckDiagnosticSessionRecovery.mockReset(); diff --git a/src/auto-reply/reply/dispatch-from-config.test-harness.ts b/src/auto-reply/reply/dispatch-from-config.test-harness.ts index 7798b7a8c101..f109519da81d 100644 --- a/src/auto-reply/reply/dispatch-from-config.test-harness.ts +++ b/src/auto-reply/reply/dispatch-from-config.test-harness.ts @@ -476,7 +476,7 @@ export const describe0BeforeEach0 = () => { ), ); mocks.routeReply.mockReset(); - mocks.routeReply.mockResolvedValue({ ok: true, messageId: "mock" }); + mocks.routeReply.mockResolvedValue({ ok: true, delivered: true, messageId: "mock" }); mocks.tryFastApproveFromMessage.mockReset(); mocks.tryFastApproveFromMessage.mockResolvedValue({ handled: false }); acpMocks.listAcpSessionEntries.mockReset().mockResolvedValue([]); @@ -579,7 +579,7 @@ export const createHookCtx = (overrides: Partial = {}) => export const describe1BeforeEach0 = () => { resetInboundDedupe(); mocks.routeReply.mockReset(); - mocks.routeReply.mockResolvedValue({ ok: true, messageId: "mock" }); + mocks.routeReply.mockResolvedValue({ ok: true, delivered: true, messageId: "mock" }); threadInfoMocks.parseSessionThreadInfo.mockReset(); threadInfoMocks.parseSessionThreadInfo.mockImplementation(parseGenericThreadSessionInfo); ttsMocks.state.synthesizeFinalAudio = false; @@ -599,7 +599,7 @@ export const describe2BeforeEach0 = () => { // Same routeReply reset as the sibling suite setups: queued once-values and // persistent overrides must not leak between tests. mocks.routeReply.mockReset(); - mocks.routeReply.mockResolvedValue({ ok: true, messageId: "mock" }); + mocks.routeReply.mockResolvedValue({ ok: true, delivered: true, messageId: "mock" }); sessionStoreMocks.currentEntry = undefined; sessionBindingMocks.resolveByConversation.mockReset(); sessionBindingMocks.resolveByConversation.mockReturnValue(null); diff --git a/src/auto-reply/reply/followup-delivery.test.ts b/src/auto-reply/reply/followup-delivery.test.ts index bfe4f23d55fc..8f949dc612c6 100644 --- a/src/auto-reply/reply/followup-delivery.test.ts +++ b/src/auto-reply/reply/followup-delivery.test.ts @@ -796,7 +796,11 @@ describe("deliverFollowupDecision", () => { it("never forwards cross-channel reply content to the live dispatcher on route failure", async () => { const onBlockReply = vi.fn(async (_payload: ReplyPayload) => {}); deliveryState.routeReply.mockReset(); - deliveryState.routeReply.mockResolvedValue({ ok: false, error: "offline" }); + deliveryState.routeReply.mockResolvedValue({ + ok: false, + delivered: false, + error: "offline", + }); const turn = createTurn(); turn.queued.run.messageProvider = "slack"; @@ -817,7 +821,11 @@ describe("deliverFollowupDecision", () => { it("allows the latest same-channel dispatcher to recover a route failure", async () => { const onBlockReply = vi.fn(async (_payload: ReplyPayload) => {}); deliveryState.routeReply.mockReset(); - deliveryState.routeReply.mockResolvedValue({ ok: false, error: "offline" }); + deliveryState.routeReply.mockResolvedValue({ + ok: false, + delivered: false, + error: "offline", + }); const turn = createTurn(); turn.queued.run.messageProvider = "discord"; @@ -836,7 +844,7 @@ describe("deliverFollowupDecision", () => { it("keeps block-status delivery out of the assistant transcript", async () => { deliveryState.routeReply.mockReset(); - deliveryState.routeReply.mockResolvedValue({ ok: true }); + deliveryState.routeReply.mockResolvedValue({ ok: true, delivered: true }); await deliverFollowupDecision({ decision: { kind: "deliver", payloads: [{ text: "compacting" }] }, @@ -855,7 +863,11 @@ describe("deliverFollowupDecision", () => { it("reports an origin delivery failure when no dispatcher can recover it", async () => { deliveryState.routeReply.mockReset(); deliveryState.runtimeError.mockReset(); - deliveryState.routeReply.mockResolvedValue({ ok: false, error: "offline" }); + deliveryState.routeReply.mockResolvedValue({ + ok: false, + delivered: false, + error: "offline", + }); await deliverFollowupDecision({ decision: { kind: "deliver", payloads: [{ text: "undelivered" }] }, @@ -873,4 +885,47 @@ describe("deliverFollowupDecision", () => { expect.stringContaining("route-reply failed: offline"), ); }); + + it("does not duplicate a follow-up after a partial route failure delivered it", async () => { + const onBlockReply = vi.fn(async (_payload: ReplyPayload) => {}); + deliveryState.routeReply.mockReset(); + deliveryState.routeReply.mockResolvedValue({ + ok: false, + delivered: true, + error: "later chunk failed", + }); + const turn = createTurn(); + turn.queued.run.messageProvider = "discord"; + + await deliverFollowupDecision({ + decision: { kind: "deliver", payloads: [{ text: "already delivered" }] }, + turn, + defaults: createDefaults(onBlockReply), + runId: "run-1", + runFollowup: vi.fn(async () => {}), + }); + + expect(onBlockReply).not.toHaveBeenCalled(); + }); + + it("does not retry an intentionally suppressed routed follow-up", async () => { + const onBlockReply = vi.fn(async (_payload: ReplyPayload) => {}); + deliveryState.routeReply.mockReset(); + deliveryState.routeReply.mockResolvedValue({ + ok: true, + delivered: false, + suppressed: true, + reason: "reasoning_payload_not_external", + }); + + await deliverFollowupDecision({ + decision: { kind: "deliver", payloads: [{ text: "internal reasoning", isReasoning: true }] }, + turn: createTurn(), + defaults: createDefaults(onBlockReply), + runId: "run-1", + runFollowup: vi.fn(async () => {}), + }); + + expect(onBlockReply).not.toHaveBeenCalled(); + }); }); diff --git a/src/auto-reply/reply/followup-delivery.ts b/src/auto-reply/reply/followup-delivery.ts index 5e15d1af4411..608ad634ed4e 100644 --- a/src/auto-reply/reply/followup-delivery.ts +++ b/src/auto-reply/reply/followup-delivery.ts @@ -388,8 +388,9 @@ async function sendFollowupPayloads(params: { replyKind: params.kind, runId: params.runId, }); - if (!result.ok) { - logVerbose(`followup queue: route-reply failed: ${result.error ?? "unknown error"}`); + if (!result.delivered && !result.suppressed) { + const routeError = result.error ?? "no visible delivery"; + logVerbose(`followup queue: route-reply failed: ${routeError}`); const provider = resolveOriginMessageProvider({ provider: turn.queued.run.messageProvider, }); @@ -399,11 +400,16 @@ async function sendFollowupPayloads(params: { } else if (defaults.opts?.onBlockReply) { crossChannelFailure = true; } else { - defaultRuntime.error?.( - `followup queue: route-reply failed: ${result.error ?? "unknown error"}`, + defaultRuntime.error?.(`followup queue: route-reply failed: ${routeError}`); + } + } else if (result.delivered) { + if (!result.ok) { + logVerbose( + `followup queue: route-reply partially failed after delivery: ${ + result.error ?? "unknown error" + }`, ); } - } else if (!result.suppressed) { const provider = resolveOriginMessageProvider({ provider: turn.queued.run.messageProvider, }); diff --git a/src/auto-reply/reply/get-reply.reset-hooks-fallback.test.ts b/src/auto-reply/reply/get-reply.reset-hooks-fallback.test.ts index 7c2f18465763..a3372110dd4b 100644 --- a/src/auto-reply/reply/get-reply.reset-hooks-fallback.test.ts +++ b/src/auto-reply/reply/get-reply.reset-hooks-fallback.test.ts @@ -76,19 +76,27 @@ describe("getReplyFromConfig reset-hook fallback", () => { it("emits reset hooks when inline actions return early without marking resetHookTriggered", async () => { mocks.handleInlineActions.mockResolvedValue({ kind: "reply", reply: undefined }); + const onObservedReplyDelivery = vi.fn(); - await getReplyFromConfig(buildNativeResetContext(), undefined, {}); + await getReplyFromConfig(buildNativeResetContext(), { onObservedReplyDelivery }, {}); expect(mocks.emitResetCommandHooks).toHaveBeenCalledTimes(1); const [hookParams] = expectDefined( ( mocks.emitResetCommandHooks.mock.calls as unknown as Array< - [{ action?: string; sessionKey?: string }] + [ + { + action?: string; + onObservedReplyDelivery?: () => Promise | void; + sessionKey?: string; + }, + ] > )[0], - "(mocks.emitResetCommandHooks.mock.calls as unknown as Array<\n [{ action?: string; sessionKey?: string }]\n >)[0] test invariant", + "reset hook params", ); expect(hookParams.action).toBe("new"); + expect(hookParams.onObservedReplyDelivery).toBe(onObservedReplyDelivery); expect(hookParams.sessionKey).toBe("agent:main:telegram:direct:123"); }); diff --git a/src/auto-reply/reply/get-reply.ts b/src/auto-reply/reply/get-reply.ts index ff2069e91f46..e5a3f9f0456b 100644 --- a/src/auto-reply/reply/get-reply.ts +++ b/src/auto-reply/reply/get-reply.ts @@ -879,6 +879,7 @@ export async function getReplyFromConfig( storePath, sessionEntry, previousSessionEntry, + onObservedReplyDelivery: resolvedOpts?.onObservedReplyDelivery, workspaceDir, }); }; diff --git a/src/auto-reply/reply/get-reply.types.ts b/src/auto-reply/reply/get-reply.types.ts index e6b2bb660219..199d9e9cc856 100644 --- a/src/auto-reply/reply/get-reply.types.ts +++ b/src/auto-reply/reply/get-reply.types.ts @@ -5,6 +5,7 @@ import type { GetReplyOptions } from "../get-reply-options.types.js"; import type { ReplyPayload } from "../reply-payload.js"; import type { MsgContext } from "../templating.js"; import type { QueueMode } from "./queue/types.js"; +import type { ReplyOptionsWithOperationRunState } from "./reply-operation-run-state.js"; import type { ReplyOperation } from "./reply-run-registry.js"; export type ReplySessionBinding = { @@ -31,7 +32,8 @@ type InternalReplySessionOptions = { export type InternalGetReplyOptions = GetReplyOptions & InternalReplySessionOptions & - ReplyOptionsWithHeartbeatRunScope; + ReplyOptionsWithHeartbeatRunScope & + ReplyOptionsWithOperationRunState; export function shouldBridgeCliPreambleEvents(opts: InternalGetReplyOptions | undefined): boolean { return opts?.commentaryProgressEnabled === true || opts?.progressPreambleEnabled === true; diff --git a/src/auto-reply/reply/reply-operation-run-state.ts b/src/auto-reply/reply/reply-operation-run-state.ts index e91e88e6fc6a..933761028161 100644 --- a/src/auto-reply/reply/reply-operation-run-state.ts +++ b/src/auto-reply/reply/reply-operation-run-state.ts @@ -1,5 +1,6 @@ type ReplyOperationAdmissionSnapshot = | { status: "owned" } + | { status: "accepted"; mode: "steer" | "followup" } | { status: "skipped"; reason: "active-run" | "aborted" | "lifecycle-invalidated" }; export type ReplyOperationRunState = { @@ -10,7 +11,7 @@ export type ReplyOperationRunState = { // heartbeat cleanup never infers it from whichever operation is active later. export const REPLY_OPERATION_RUN_STATE = Symbol("openclaw.replyOperationRunState"); -type ReplyOptionsWithOperationRunState = { +export type ReplyOptionsWithOperationRunState = { [REPLY_OPERATION_RUN_STATE]?: ReplyOperationRunState; }; diff --git a/src/auto-reply/reply/route-reply.delivery-result.test.ts b/src/auto-reply/reply/route-reply.delivery-result.test.ts new file mode 100644 index 000000000000..1327363ece81 --- /dev/null +++ b/src/auto-reply/reply/route-reply.delivery-result.test.ts @@ -0,0 +1,226 @@ +// Tests routeReply delivery evidence and editable message identity. +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { ChannelPlugin } from "../../channels/plugins/types.public.js"; +import { OutboundDeliveryError } from "../../infra/outbound/deliver-types.js"; +import { setActivePluginRegistry } from "../../plugins/runtime.js"; +import { + createChannelTestPluginBase, + createTestRegistry, +} from "../../test-utils/channel-plugins.js"; + +const mocks = vi.hoisted(() => ({ + deliverOutboundPayloads: vi.fn(), +})); + +vi.mock("../../infra/outbound/deliver-runtime.js", () => ({ + deliverOutboundPayloads: mocks.deliverOutboundPayloads, + deliverOutboundPayloadsInternal: mocks.deliverOutboundPayloads, +})); + +vi.mock("../../infra/outbound/deliver.js", () => ({ + deliverOutboundPayloads: mocks.deliverOutboundPayloads, + deliverOutboundPayloadsInternal: mocks.deliverOutboundPayloads, +})); + +const { routeReply: routeReplyRuntime } = await import("./route-reply.js"); +type RouteReplyParams = Parameters[0]; +const routeReply = ( + params: Omit & { replyKind?: RouteReplyParams["replyKind"] }, +) => routeReplyRuntime({ replyKind: "final", ...params }); + +function createChannelPlugin(id: ChannelPlugin["id"], label: string): ChannelPlugin { + return createChannelTestPluginBase({ + id, + label, + config: { listAccountIds: () => [], resolveAccount: () => ({}) }, + }); +} + +describe("routeReply delivery result", () => { + beforeEach(() => { + setActivePluginRegistry( + createTestRegistry([ + { + pluginId: "telegram", + plugin: createChannelPlugin("telegram", "Telegram"), + source: "test", + }, + { + pluginId: "whatsapp", + plugin: createChannelPlugin("whatsapp", "WhatsApp"), + source: "test", + }, + ]), + ); + mocks.deliverOutboundPayloads.mockReset(); + mocks.deliverOutboundPayloads.mockResolvedValue([]); + }); + + afterEach(() => { + setActivePluginRegistry(createTestRegistry()); + }); + + it.each(["cancelled_by_message_sending_hook", "empty_after_message_sending_hook"] as const)( + "returns routed message hook suppression reason %s", + async (reason) => { + mocks.deliverOutboundPayloads.mockImplementationOnce( + async ({ + onPayloadDeliveryOutcome, + }: { + onPayloadDeliveryOutcome?: (outcome: unknown) => void; + }) => { + onPayloadDeliveryOutcome?.({ + index: 0, + status: "suppressed", + reason, + }); + return []; + }, + ); + + const res = await routeReply({ + payload: { text: "hello" }, + channel: "telegram", + to: "chat-1", + cfg: {} as never, + }); + + expect(res).toEqual({ + ok: true, + delivered: false, + suppressed: true, + reason, + }); + }, + ); + + it("treats a send without adapter identity as ambiguous and non-retryable", async () => { + mocks.deliverOutboundPayloads.mockImplementationOnce( + async ({ + onPayloadDeliveryOutcome, + }: { + onPayloadDeliveryOutcome?: (outcome: unknown) => void; + }) => { + onPayloadDeliveryOutcome?.({ + index: 0, + status: "suppressed", + reason: "adapter_returned_no_identity", + }); + return []; + }, + ); + + const res = await routeReply({ + payload: { text: "hello" }, + channel: "telegram", + to: "chat-1", + cfg: {} as never, + }); + + expect(res).toEqual({ + ok: true, + delivered: true, + ambiguous: true, + reason: "adapter_returned_no_identity", + }); + }); + + it("preserves the last delivered message id when a later send fails", async () => { + const cause = new Error("network reset"); + mocks.deliverOutboundPayloads.mockRejectedValueOnce( + new OutboundDeliveryError("network reset", { + cause, + results: [{ channel: "telegram", messageId: "msg-1" }], + stage: "platform_send", + }), + ); + + const res = await routeReply({ + payload: { text: "hello" }, + channel: "telegram", + to: "chat-1", + cfg: {} as never, + }); + + expect(res).toEqual({ + ok: false, + delivered: true, + error: "Failed to route reply to telegram: network reset", + messageId: "msg-1", + }); + }); + + it.each([ + ["a trailing suppression sentinel", { channel: "telegram", messageId: "suppressed" }], + ["a trailing unknown sentinel", { channel: "telegram", messageId: "unknown" }], + ["a trailing ok sentinel", { channel: "telegram", messageId: "ok" }], + ["a trailing no-id receipt", { channel: "telegram", messageId: "" }], + ])("preserves an earlier editable message id after %s", async (_label, trailingResult) => { + const cause = new Error("network reset"); + mocks.deliverOutboundPayloads.mockRejectedValueOnce( + new OutboundDeliveryError("network reset", { + cause, + results: [{ channel: "telegram", messageId: "msg-1" }, trailingResult], + stage: "platform_send", + }), + ); + + const res = await routeReply({ + payload: { text: "hello" }, + channel: "telegram", + to: "chat-1", + cfg: {} as never, + }); + + expect(res).toEqual({ + ok: false, + delivered: true, + error: "Failed to route reply to telegram: network reset", + messageId: "msg-1", + }); + }); + + it("reports delivery when the provider returns a non-id delivery identity", async () => { + mocks.deliverOutboundPayloads.mockResolvedValueOnce([ + { channel: "whatsapp", messageId: "", toJid: "group:ops" }, + ]); + + const res = await routeReply({ + payload: { text: "hello" }, + channel: "whatsapp", + to: "group:ops", + cfg: {} as never, + }); + + expect(res).toEqual({ + ok: true, + delivered: true, + messageId: "", + }); + }); + + it.each([ + ["skipped", false, undefined], + ["suppressed", false, undefined], + ["unknown", true, undefined], + ["ok", true, undefined], + ] as const)( + "reports message id %s visibility as %s", + async (messageId, delivered, returnedId) => { + mocks.deliverOutboundPayloads.mockResolvedValueOnce([{ channel: "telegram", messageId }]); + + const res = await routeReply({ + payload: { text: "hello" }, + channel: "telegram", + to: "chat-1", + cfg: {} as never, + }); + + expect(res).toEqual({ + ok: true, + delivered, + ...(returnedId === undefined ? {} : { messageId: returnedId }), + }); + }, + ); +}); diff --git a/src/auto-reply/reply/route-reply.test.ts b/src/auto-reply/reply/route-reply.test.ts index 8fe1a9f7e2f4..72ebc63f9ec3 100644 --- a/src/auto-reply/reply/route-reply.test.ts +++ b/src/auto-reply/reply/route-reply.test.ts @@ -241,7 +241,13 @@ describe("routeReply", () => { }); it("suppresses reasoning payloads", async () => { - await expectSlackNoDelivery({ text: "step", isReasoning: true }); + await expect(expectSlackNoDelivery({ text: "step", isReasoning: true })).resolves.toMatchObject( + { + delivered: false, + suppressed: true, + reason: "reasoning_payload_not_external", + }, + ); }); it("drops silent token payloads", async () => { @@ -546,6 +552,7 @@ describe("routeReply", () => { expect(res).toEqual({ ok: true, + delivered: false, suppressed: true, reason: "cancelled_by_reply_payload_sending_hook", }); @@ -585,6 +592,7 @@ describe("routeReply", () => { expect(res).toEqual({ ok: true, + delivered: false, suppressed: true, reason: "cancelled_by_reply_payload_sending_hook", }); @@ -616,6 +624,7 @@ describe("routeReply", () => { expect(res).toEqual({ ok: true, + delivered: false, suppressed: true, reason: "empty_after_reply_payload_sending_hook", }); diff --git a/src/auto-reply/reply/route-reply.ts b/src/auto-reply/reply/route-reply.ts index f9bdf19470ec..360be10dda5f 100644 --- a/src/auto-reply/reply/route-reply.ts +++ b/src/auto-reply/reply/route-reply.ts @@ -111,16 +111,59 @@ type RouteReplyParams = { type RouteReplyResult = { /** Whether the reply was sent successfully. */ ok: boolean; + /** Whether a recipient-visible send completed or may already have completed. */ + delivered: boolean; + /** True when the adapter may have sent but returned no delivery identity. */ + ambiguous?: boolean; /** True when a hook intentionally suppressed provider delivery. */ suppressed?: boolean; - /** Suppression reason when delivery was intentionally skipped. */ - reason?: "cancelled_by_reply_payload_sending_hook" | "empty_after_reply_payload_sending_hook"; + /** Delivery disposition reason when additional caller context is useful. */ + reason?: + | "reasoning_payload_not_external" + | "adapter_returned_no_identity" + | "cancelled_by_message_sending_hook" + | "cancelled_by_reply_payload_sending_hook" + | "empty_after_message_sending_hook" + | "empty_after_reply_payload_sending_hook"; /** Optional message ID from the provider. */ messageId?: string; /** Error message if the send failed. */ error?: string; }; +function summarizeVisibleRouteReplyDelivery( + results: readonly { messageId?: string }[], +): Pick { + // Durable results may prove delivery through a receipt or alternate identity + // when messageId is empty. Provider success sentinels prove delivery but are + // not editable IDs; explicit suppression sentinels prove neither. + let delivered = false; + let lastVisibleMessageId: string | undefined; + for (let index = results.length - 1; index >= 0; index -= 1) { + const result = results[index]; + if (!result) { + continue; + } + const messageId = result.messageId?.trim().toLowerCase(); + if (messageId === "skipped" || messageId === "suppressed") { + continue; + } + if (!delivered) { + delivered = true; + if (!messageId) { + lastVisibleMessageId = result.messageId; + } + } + if (messageId && messageId !== "unknown" && messageId !== "ok") { + return { delivered: true, messageId: result.messageId }; + } + } + return { + delivered, + messageId: delivered ? lastVisibleMessageId : undefined, + }; +} + /** * Routes a reply payload to the specified channel. * @@ -132,7 +175,12 @@ type RouteReplyResult = { export async function routeReply(params: RouteReplyParams): Promise { const { payload, channel, to, accountId, threadId, cfg, abortSignal } = params; if (shouldSuppressReasoningPayload(payload)) { - return { ok: true }; + return { + ok: true, + delivered: false, + suppressed: true, + reason: "reasoning_payload_not_external", + }; } const normalizedChannel = normalizeMessageChannel(channel); const channelId = @@ -167,7 +215,7 @@ export async function routeReply(params: RouteReplyParams): Promise