From da6a24d7dc41d2e19d2c9bfd095ae8b6c850c4fb Mon Sep 17 00:00:00 2001 From: NianJiu <3235467914@qq.com> Date: Sun, 12 Jul 2026 06:47:15 +0800 Subject: [PATCH] fix(reply): prevent hung pre-delivery hooks from blocking lanes (#104256) * fix(reply): bound pre-delivery hook settlement * test(plugins): preserve hook timeout fixtures * fix(reply): preserve declared pre-delivery budgets * chore(reply): finalize pre-delivery recovery * fix(reply): preserve per-final recovery semantics * fix(reply): guard pending-final settlement identity * test(reply): align pending-final store fixture * fix(reply): bind settlement to originating intent * test(reply): satisfy timeout fixture lint * chore(plugin-sdk): refresh rebased baseline * fix(reply): preserve normalized retry ownership * fix(reply): narrow pending retry metadata * fix: align reply dispatch state access * chore(plugin-sdk): refresh rebased surface baseline * test(reply): align rebased accessor assertion * chore(plugin-sdk): regenerate drifted API baseline --------- Co-authored-by: NianJiuZst <180004567+NianJiuZst@users.noreply.github.com> Co-authored-by: Peter Steinberger --- CHANGELOG.md | 1 + .../.generated/plugin-sdk-api-baseline.sha256 | 4 +- docs/plugins/hooks.md | 12 + .../mattermost/src/mattermost/monitor.ts | 24 +- .../lib/session-accessor-debt-baseline.json | 1 - scripts/plugin-sdk-surface-report.mjs | 2 +- src/auto-reply/dispatch.freshness.test.ts | 84 ++++ src/auto-reply/dispatch.ts | 115 +++-- src/auto-reply/reply-payload.ts | 4 + .../agent-runner.runreplyagent.e2e.test.ts | 18 +- src/auto-reply/reply/agent-runner.ts | 37 +- ...ispatch-from-config.reply-dispatch.test.ts | 462 +++++++++++++++++- ...ispatch-from-config.shared.test-harness.ts | 21 + src/auto-reply/reply/dispatch-from-config.ts | 288 ++++++++++- .../reply/pending-final-delivery.ts | 13 +- src/auto-reply/reply/reply-dispatcher.ts | 205 +++++++- .../reply/reply-dispatcher.types.ts | 11 +- src/auto-reply/reply/reply-flow.test.ts | 252 +++++++++- src/plugin-sdk/reply-runtime.ts | 1 + .../contracts/plugin-sdk-subpaths.test.ts | 12 +- src/plugins/hooks.test-helpers.ts | 7 +- src/plugins/hooks.ts | 4 + src/plugins/wired-hooks-message.test.ts | 75 +++ .../wired-hooks-reply-payload-sending.test.ts | 48 ++ 24 files changed, 1568 insertions(+), 133 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 08c1ca3c90c7..2645d972d2ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -204,6 +204,7 @@ Docs: https://docs.openclaw.ai - **Bedrock Mantle discovery:** bound model-catalog fetch time and response size, and release rejected response bodies so stalled, oversized, or failed provider responses fall back safely. (#99961) Thanks @zhangguiping-xydt. - **Discord thread-title prompts:** truncate generated-title message and channel context on UTF-16 boundaries so emoji cannot leave malformed model prompt text. (#101551) Thanks @Alix-007. - **Task state migration:** canonicalize legacy `not-requested` delivery statuses during sidecar import and existing shared-database open so upgraded task registries and linked TaskFlows recover without manual SQL, and surface rejected persisted values in compact console diagnostics. (#103946) Thanks @bek91. +- **Reply pre-delivery recovery:** bound each pre-delivery callback with an owner-overridable deadline, release serialized reply lanes after hung plugin work, and preserve durable final-delivery retry state only when transport never started. (#104256) Thanks @NianJiuZst. ## 2026.7.1 diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index 8f0c19f98905..43585b4131e5 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -1,2 +1,2 @@ -1a655db74b630c67a5549e4254fd4dbb2965dbb8f1420ba425eafde5c6e41ad9 plugin-sdk-api-baseline.json -da435187b9793395e7555949d1e2ca483e9ed4d8478d666bbb20586b33a62c32 plugin-sdk-api-baseline.jsonl +d8383ea9819598d92e1d1052eeb9adb61f79cd5b16e0991fc60e17a99612848d plugin-sdk-api-baseline.json +b9c7eed49500d9f5d371a25c0bc7b968d55e6272081a6e9be2113f81c2415689 plugin-sdk-api-baseline.jsonl diff --git a/docs/plugins/hooks.md b/docs/plugins/hooks.md index 9386718ab65e..5bb9c74a207c 100644 --- a/docs/plugins/hooks.md +++ b/docs/plugins/hooks.md @@ -93,6 +93,18 @@ receive a cancellation signal. The hook dispatch can release its Gateway admission while that plugin work is still in progress. Plugins that own long-running work must provide their own cancellation and shutdown lifecycle. +Outbound modifying hooks `message_sending` and `reply_payload_sending` use a +15-second default per handler. If one times out, OpenClaw logs the plugin error +and continues with the latest payload so the serialized delivery lane can +settle. Set a larger per-hook budget for plugins that intentionally do slower +work before delivery. + +Channel plugins that use `createReplyDispatcher` can likewise declare a larger +positive per-stage budget with `beforeDeliverOptions: { timeoutMs }`, or when +appending work with `dispatcher.appendBeforeDeliver(handler, { timeoutMs })`. +Without an owner-declared budget, those callbacks use the same 15-second +default so a hung callback cannot retain the serialized delivery lane. + Each hook receives `event.context.pluginConfig`, the resolved config for the plugin that registered that handler. OpenClaw injects it per handler without mutating the shared event object other plugins see. diff --git a/extensions/mattermost/src/mattermost/monitor.ts b/extensions/mattermost/src/mattermost/monitor.ts index 425acf2904bc..3d5560d00ebd 100644 --- a/extensions/mattermost/src/mattermost/monitor.ts +++ b/extensions/mattermost/src/mattermost/monitor.ts @@ -856,18 +856,24 @@ export async function monitorMattermostProvider(opts: MonitorMattermostOpts = {} onReplyStart: typingCallbacks?.onReplyStart, }); - await core.channel.reply.dispatchReplyFromConfig({ - ctx: ctxPayload, - cfg, + await core.channel.reply.withReplyDispatcher({ dispatcher, - replyOptions: { - ...replyOptions, - disableBlockStreaming: - typeof account.blockStreaming === "boolean" ? !account.blockStreaming : undefined, - onModelSelected, + onSettled: () => { + markDispatchIdle(); }, + run: () => + core.channel.reply.dispatchReplyFromConfig({ + ctx: ctxPayload, + cfg, + dispatcher, + replyOptions: { + ...replyOptions, + disableBlockStreaming: + typeof account.blockStreaming === "boolean" ? !account.blockStreaming : undefined, + onModelSelected, + }, + }), }); - markDispatchIdle(); }, log: (msg) => runtime.log?.(msg), }), diff --git a/scripts/lib/session-accessor-debt-baseline.json b/scripts/lib/session-accessor-debt-baseline.json index cbee09387ced..175d021c4d06 100644 --- a/scripts/lib/session-accessor-debt-baseline.json +++ b/scripts/lib/session-accessor-debt-baseline.json @@ -22,7 +22,6 @@ "sessionAccessorWrite": { "src/agents/session-suspension.ts": 2, "src/agents/subagent-spawn.test-helpers.ts": 1, - "src/auto-reply/reply/dispatch-from-config.ts": 2, "src/commands/doctor-heartbeat-main-session-repair.ts": 2, "src/commands/doctor-session-snapshots.ts": 2, "src/commands/doctor-session-state-providers.ts": 2, diff --git a/scripts/plugin-sdk-surface-report.mjs b/scripts/plugin-sdk-surface-report.mjs index 8391735e0fc0..00e6a81ff2b4 100644 --- a/scripts/plugin-sdk-surface-report.mjs +++ b/scripts/plugin-sdk-surface-report.mjs @@ -195,7 +195,7 @@ export function readPluginSdkSurfaceBudgets(env = process.env) { ), publicExports: readPluginSdkSurfaceBudgetEnv( "OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS", - 10553, + 10554, env, ), publicFunctionExports: readPluginSdkSurfaceBudgetEnv( diff --git a/src/auto-reply/dispatch.freshness.test.ts b/src/auto-reply/dispatch.freshness.test.ts index adfd7315ec1e..188996975b08 100644 --- a/src/auto-reply/dispatch.freshness.test.ts +++ b/src/auto-reply/dispatch.freshness.test.ts @@ -5,6 +5,7 @@ import { OutboundDeliveryError } from "../infra/outbound/deliver-types.js"; import { resetGlobalHookRunner } from "../plugins/hook-runner-global.js"; import { getReplyPayloadMetadata } from "./reply-payload.js"; import type { ReplyDispatchBeforeDeliver } from "./reply/reply-dispatcher.js"; +import type { ReplyDispatchBeforeDeliverOptions } from "./reply/reply-dispatcher.types.js"; import { buildTestCtx } from "./reply/test-ctx.js"; import type { FinalizedMsgContext, MsgContext } from "./templating.js"; import type { ReplyPayload } from "./types.js"; @@ -64,6 +65,7 @@ function dispatchWithDeliveries( deliveries: Delivery[], dispatcherOptions: { beforeDeliver?: ReplyDispatchBeforeDeliver; + beforeDeliverOptions?: ReplyDispatchBeforeDeliverOptions; deliver?: (payload: ReplyPayload, info: { kind: Delivery["kind"] }) => Promise; onBeforeDeliverCancelled?: (payload: ReplyPayload, info: { kind: Delivery["kind"] }) => void; onSettled?: () => object | void | Promise; @@ -177,6 +179,88 @@ describe("foreground reply freshness", () => { expect(cancellationReasons).toEqual([undefined]); }); + it("releases a WhatsApp-shaped lane after beforeDeliver times out", async () => { + vi.useFakeTimers(); + try { + const deliveries: Delivery[] = []; + const hookStarted = createDeferred(); + const onSettled = vi.fn(); + let hookCalls = 0; + const beforeDeliver = vi.fn((payload: ReplyPayload) => { + hookCalls += 1; + if (hookCalls === 1) { + hookStarted.resolve(); + return new Promise(() => {}); + } + return payload; + }); + hoisted.dispatchReplyFromConfigMock.mockImplementation( + async (params: DispatchReplyFromConfigParams) => { + params.dispatcher.sendFinalReply({ text: "stuck final" }); + params.dispatcher.sendFinalReply({ text: "follow-up final" }); + return { + queuedFinal: true, + counts: { tool: 0, block: 0, final: 2 }, + }; + }, + ); + + const dispatch = dispatchWithDeliveries(buildForegroundCtx(), deliveries, { + beforeDeliver, + onSettled, + }); + await hookStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + + await expect(dispatch).resolves.toEqual({ + queuedFinal: true, + counts: { tool: 0, block: 0, final: 1 }, + failedCounts: { tool: 0, block: 0, final: 1 }, + }); + expect(beforeDeliver).toHaveBeenCalledTimes(2); + expect(deliveries).toEqual([{ kind: "final", text: "follow-up final" }]); + expect(onSettled).toHaveBeenCalledOnce(); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("honors a configured beforeDeliver budget inside the foreground fence", async () => { + vi.useFakeTimers(); + try { + const deliveries: Delivery[] = []; + const hookStarted = createDeferred(); + hoisted.dispatchReplyFromConfigMock.mockImplementation( + async (params: DispatchReplyFromConfigParams) => { + params.dispatcher.sendFinalReply({ text: "budgeted final" }); + return queuedFinalResult(); + }, + ); + + const dispatch = dispatchWithDeliveries(buildForegroundCtx(), deliveries, { + beforeDeliver: async (payload) => { + hookStarted.resolve(); + await new Promise((resolve) => { + setTimeout(resolve, 16_000); + }); + return payload; + }, + beforeDeliverOptions: { timeoutMs: 20_000 }, + }); + await hookStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + expect(deliveries).toEqual([]); + await vi.advanceTimersByTimeAsync(1_000); + + await expect(dispatch).resolves.toEqual(queuedFinalResult()); + expect(deliveries).toEqual([{ kind: "final", text: "budgeted final" }]); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + it("keeps an older foreground final when a newer inbound has no visible delivery while beforeDeliver is pending", async () => { const deliveries: Delivery[] = []; const beforeDeliverStarted = createDeferred(); diff --git a/src/auto-reply/dispatch.ts b/src/auto-reply/dispatch.ts index 2aa824a19755..52eef112bd4a 100644 --- a/src/auto-reply/dispatch.ts +++ b/src/auto-reply/dispatch.ts @@ -30,8 +30,10 @@ import type { } from "./reply/get-reply.types.js"; import { finalizeInboundContext } from "./reply/inbound-context.js"; import { + composeReplyDispatchBeforeDeliver, createReplyDispatcher, createReplyDispatcherWithTyping, + markReplyDispatchBeforeDeliverDeadlineOwned, type ReplyDispatchBeforeDeliver, type ReplyDispatcherOptions, type ReplyDispatcherWithTypingOptions, @@ -362,24 +364,26 @@ function buildMessageSendingBeforeDeliver( const hookCtx = deriveInboundMessageHookContext(finalized); const replyTarget = resolveInboundReplyHookTarget(finalized, hookCtx); - return async (payload: ReplyPayload): Promise => { - if (!payload.text) { + return markReplyDispatchBeforeDeliverDeadlineOwned( + async (payload: ReplyPayload): Promise => { + if (!payload.text) { + return payload; + } + + const result = await hookRunner.runMessageSending( + { content: payload.text, to: replyTarget }, + toPluginMessageContext(hookCtx), + ); + + if (result?.cancel) { + return null; + } + if (result?.content != null) { + return copyReplyPayloadMetadata(payload, { ...payload, text: result.content }); + } return payload; - } - - const result = await hookRunner.runMessageSending( - { content: payload.text, to: replyTarget }, - toPluginMessageContext(hookCtx), - ); - - if (result?.cancel) { - return null; - } - if (result?.content != null) { - return copyReplyPayloadMetadata(payload, { ...payload, text: result.content }); - } - return payload; - }; + }, + ); } function buildReplyPayloadSendingBeforeDeliver( @@ -389,22 +393,24 @@ function buildReplyPayloadSendingBeforeDeliver( const finalized = finalizeInboundContext(ctx); const hookCtx = deriveInboundMessageHookContext(finalized); - return async (payload: ReplyPayload, info): Promise => { - const runId = runState.runId; - const hookedPayload = await runReplyPayloadSendingHook({ - payload, - kind: info.kind, - channel: finalized.Surface ?? finalized.Provider, - sessionKey: finalized.SessionKey, - runId, - usageState: consumeReplyUsageState(runId), - context: { - ...toPluginMessageContext(hookCtx), + return markReplyDispatchBeforeDeliverDeadlineOwned( + async (payload: ReplyPayload, info): Promise => { + const runId = runState.runId; + const hookedPayload = await runReplyPayloadSendingHook({ + payload, + kind: info.kind, + channel: finalized.Surface ?? finalized.Provider, + sessionKey: finalized.SessionKey, runId, - }, - }); - return hookedPayload && hasOutboundReplyContent(hookedPayload) ? hookedPayload : null; - }; + usageState: consumeReplyUsageState(runId), + context: { + ...toPluginMessageContext(hookCtx), + runId, + }, + }); + return hookedPayload && hasOutboundReplyContent(hookedPayload) ? hookedPayload : null; + }, + ); } function bindReplyPayloadRunState( @@ -446,27 +452,6 @@ function markReplyPayloadSendingBeforeDeliverInstalled( } } -function combineBeforeDeliverHooks( - ...hooks: Array -): ReplyDispatchBeforeDeliver | undefined { - const activeHooks = hooks.filter((hook): hook is ReplyDispatchBeforeDeliver => Boolean(hook)); - if (activeHooks.length === 0) { - return undefined; - } - - return async (payload, info) => { - let current: ReplyPayload | null = payload; - for (const hook of activeHooks) { - if (!current) { - return null; - } - const next = await hook(current, info); - current = next ? copyReplyPayloadMetadata(current, next) : null; - } - return current; - }; -} - function buildDispatchTimelineAttributes(ctx: MsgContext | FinalizedMsgContext) { const commandTurn = resolveCommandTurnContext(ctx); return { @@ -603,16 +588,22 @@ export async function dispatchInboundMessageWithBufferedDispatcher(params: { finalized, replyPayloadRunState, ); - const globalBeforeDeliver = combineBeforeDeliverHooks( + const globalBeforeDeliver = composeReplyDispatchBeforeDeliver( replyPayloadBeforeDeliver, buildMessageSendingBeforeDeliver(finalized), ); const configuredBeforeDeliver = params.dispatcherOptions.beforeDeliver - ? combineBeforeDeliverHooks(params.dispatcherOptions.beforeDeliver, replyPayloadBeforeDeliver) + ? composeReplyDispatchBeforeDeliver( + { + hook: params.dispatcherOptions.beforeDeliver, + options: params.dispatcherOptions.beforeDeliverOptions, + }, + replyPayloadBeforeDeliver, + ) : globalBeforeDeliver; const beforeDeliver: ReplyDispatchBeforeDeliver | undefined = foregroundReplyFence || configuredBeforeDeliver - ? async (payload, info) => { + ? markReplyDispatchBeforeDeliverDeadlineOwned(async (payload, info) => { // Check both before and after hooks because hooks can await while newer replies finish. if (await shouldCancelForegroundReplyDelivery(foregroundReplyFence)) { // Only the foreground fence proves "not shown because stale"; hook @@ -635,7 +626,7 @@ export async function dispatchInboundMessageWithBufferedDispatcher(params: { return null; } return deliverPayload; - } + }) : undefined; const deliver: ReplyDispatcherWithTypingOptions["deliver"] = async (payload, info) => { try { @@ -715,12 +706,18 @@ export async function dispatchInboundMessageWithDispatcher(params: { params.ctx, replyPayloadRunState, ); - const globalBeforeDeliver = combineBeforeDeliverHooks( + const globalBeforeDeliver = composeReplyDispatchBeforeDeliver( replyPayloadBeforeDeliver, buildMessageSendingBeforeDeliver(params.ctx), ); const composedBeforeDeliver = params.dispatcherOptions.beforeDeliver - ? combineBeforeDeliverHooks(params.dispatcherOptions.beforeDeliver, replyPayloadBeforeDeliver) + ? composeReplyDispatchBeforeDeliver( + { + hook: params.dispatcherOptions.beforeDeliver, + options: params.dispatcherOptions.beforeDeliverOptions, + }, + replyPayloadBeforeDeliver, + ) : globalBeforeDeliver; const dispatcher = createReplyDispatcher({ ...params.dispatcherOptions, diff --git a/src/auto-reply/reply-payload.ts b/src/auto-reply/reply-payload.ts index 76a41df2e8e9..b57ffd56d03d 100644 --- a/src/auto-reply/reply-payload.ts +++ b/src/auto-reply/reply-payload.ts @@ -221,6 +221,10 @@ export type ReplyPayloadMetadata = { }; /** Opaque owner for one final-delivery transcript capture on a shared dispatcher. */ finalDeliveryCapture?: object; + /** Durable pending-final intent represented by this runtime payload. */ + pendingFinalDeliveryIntentId?: string; + /** Restart-safe text this payload contributes to its pending-final intent. */ + pendingFinalDeliveryRetryText?: string; /** replyToId existed before reply threading could inject an implicit target. */ replyToIdExplicit?: boolean; /** Canonical reply policy used by both message-tool dedupe and final delivery routing. */ 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 37e17e068cd0..d892cae35e09 100644 --- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts +++ b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts @@ -748,11 +748,19 @@ describe("runReplyAgent pending final delivery capture", () => { storePath, }); - await run(); + const result = await run(); const stored = await readStoredMainSession(storePath); expect(stored.pendingFinalDelivery).toBe(true); expect(stored.pendingFinalDeliveryText).toBe("visible final"); + expect(stored.pendingFinalDeliveryIntentId).toEqual(expect.any(String)); + const visiblePayload = (Array.isArray(result) ? result : [result]).find( + (payload) => payload?.text === "visible final", + ); + expect(getReplyPayloadMetadata(visiblePayload ?? {})).toMatchObject({ + pendingFinalDeliveryIntentId: stored.pendingFinalDeliveryIntentId, + pendingFinalDeliveryRetryText: "visible final", + }); }); it("persists auto-reply delivery context for restart recovery", async () => { @@ -958,11 +966,16 @@ describe("runReplyAgent pending final delivery capture", () => { storePath, }); - await run(); + const result = await run(); const stored = await readStoredMainSession(storePath); expect(stored.pendingFinalDelivery).toBe(true); expect(stored.pendingFinalDeliveryText).toBe(longRemainder); + const payload = Array.isArray(result) ? result[0] : result; + expect(getReplyPayloadMetadata(payload ?? {})).toMatchObject({ + pendingFinalDeliveryIntentId: stored.pendingFinalDeliveryIntentId, + pendingFinalDeliveryRetryText: longRemainder, + }); }); }); @@ -3114,4 +3127,5 @@ describe("runReplyAgent typing (heartbeat)", () => { }); }); +import { getReplyPayloadMetadata } from "../reply-payload.js"; import type { ReplyPayload } from "../types.js"; diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts index b70eb54a8956..1126ffb3e7c9 100644 --- a/src/auto-reply/reply/agent-runner.ts +++ b/src/auto-reply/reply/agent-runner.ts @@ -108,7 +108,10 @@ import { createFollowupRunner } from "./followup-runner.js"; import { REPLY_RUN_STILL_SHUTTING_DOWN_TEXT } from "./get-reply-run-queue.js"; import { normalizeReplyPayload } from "./normalize-reply.js"; import { resolveOriginMessageProvider, resolveOriginMessageTo } from "./origin-routing.js"; -import { sanitizePendingFinalDeliveryText } from "./pending-final-delivery.js"; +import { + buildPendingFinalDeliveryText, + sanitizePendingFinalDeliveryText, +} from "./pending-final-delivery.js"; import { drainPendingToolTasks } from "./pending-tool-task-drain.js"; import { readPostCompactionContext } from "./post-compaction-context.js"; import { @@ -174,6 +177,18 @@ function markBeforeAgentRunBlockedPayloads(payloads: ReplyPayload[]): ReplyPaylo ); } +function resolvePendingFinalDeliveryRetryText(params: { + isHeartbeat: boolean; + payload: ReplyPayload; +}): string { + const pendingText = buildPendingFinalDeliveryText([params.payload]); + if (!params.isHeartbeat) { + return pendingText; + } + const stripped = stripHeartbeatToken(pendingText, { mode: "message" }); + return stripped.shouldSkip ? "" : stripped.text || pendingText; +} + function buildSilentFallbackFailurePayload(params: { fallbackTransition: ReturnType; fallbackFailureKnown: boolean; @@ -1000,15 +1015,6 @@ function joinCommitmentAssistantText(payloads: ReplyPayload[]): string { .trim(); } -function buildPendingFinalDeliveryText(payloads: ReplyPayload[]): string { - const text = payloads - .filter((payload) => payload.isReasoning !== true) - .map((payload) => payload.text) - .filter((textLocal): textLocal is string => Boolean(textLocal)) - .join("\n\n"); - return sanitizePendingFinalDeliveryText(text); -} - function normalizeAssistantFinalDeliveryText(text: string): string { const parsed = normalizeReplyPayloadDirectives({ payload: { text }, @@ -2708,6 +2714,16 @@ export async function runReplyAgent(params: { })() : pendingText; if (resolvedPendingText) { + const pendingFinalDeliveryIntentId = crypto.randomUUID(); + for (const payload of finalPayloads) { + setReplyPayloadMetadata(payload, { + pendingFinalDeliveryIntentId, + pendingFinalDeliveryRetryText: resolvePendingFinalDeliveryRetryText({ + isHeartbeat, + payload, + }), + }); + } const pendingFinalDeliveryContext = resolveReplyRunDeliveryContext({ cfg, sessionCtx, @@ -2721,6 +2737,7 @@ export async function runReplyAgent(params: { () => ({ pendingFinalDelivery: true, pendingFinalDeliveryText: resolvedPendingText, + pendingFinalDeliveryIntentId, pendingFinalDeliveryContext, pendingFinalDeliveryCreatedAt: Date.now(), updatedAt: Date.now(), 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 87c63bf18fdb..d19f7d0b1336 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 @@ -3,6 +3,7 @@ import { beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { clearAgentHarnesses } from "../../agents/harness/registry.js"; import type { PluginHookReplyDispatchResult } from "../../plugins/hooks.js"; import { createInternalHookEventPayload } from "../../test-utils/internal-hook-event-payload.js"; +import { withReplyDispatcher } from "../dispatch-dispatcher.js"; import { setReplyPayloadMetadata, type ReplyPayload } from "../types.js"; import { acpManagerRuntimeMocks, @@ -52,6 +53,14 @@ function firstReplyDispatchCall() { | undefined; } +function createDeferred() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + describe("dispatchReplyFromConfig reply_dispatch hook", () => { beforeAll(async () => { ({ dispatchReplyFromConfig } = await import("./dispatch-from-config.js")); @@ -101,10 +110,12 @@ describe("dispatchReplyFromConfig reply_dispatch hook", () => { () => sessionStoreMocks.currentEntry, ); sessionStoreMocks.loadSessionStore.mockReset().mockReturnValue({}); - sessionStoreMocks.readSessionEntry.mockReset().mockReturnValue(undefined); + sessionStoreMocks.readSessionEntry + .mockReset() + .mockImplementation(() => sessionStoreMocks.currentEntry); sessionStoreMocks.resolveStorePath.mockReset().mockReturnValue("/tmp/mock-sessions.json"); sessionStoreMocks.resolveSessionStoreEntry.mockReset().mockReturnValue({ existing: undefined }); - sessionStoreMocks.updateSessionStoreEntry.mockClear(); + sessionStoreMocks.updateSessionEntry.mockClear(); acpManagerRuntimeMocks.getAcpSessionManager.mockReset(); acpManagerRuntimeMocks.getAcpSessionManager.mockImplementation(() => ({ resolveSession: () => ({ kind: "none" as const }), @@ -211,12 +222,18 @@ describe("dispatchReplyFromConfig reply_dispatch hook", () => { sessionStoreMocks.loadSessionStore.mockClear(); mocks.routeReply.mockResolvedValue({ ok: true, messageId: "mock" }); + const deliver = vi.fn().mockResolvedValue(undefined); + const dispatcher = createReplyDispatcher({ deliver }); const result = await dispatchReplyFromConfig({ ctx: createHookCtx(), cfg: emptyConfig, - dispatcher: createDispatcher(), + dispatcher, replyResolver: async () => ({ text: "durable reply" }), }); + await dispatcher.waitForIdle(); + await vi.waitFor(() => { + expect(sessionStoreMocks.updateSessionEntry).toHaveBeenCalledOnce(); + }); expect(result.queuedFinal).toBe(true); expect(sessionStoreMocks.loadSessionStoreEntry).toHaveBeenCalledWith({ @@ -227,7 +244,7 @@ describe("dispatchReplyFromConfig reply_dispatch hook", () => { clone: false, }); expect(sessionStoreMocks.loadSessionStore).not.toHaveBeenCalled(); - expect(sessionStoreMocks.updateSessionStoreEntry).toHaveBeenCalledOnce(); + expect(deliver).toHaveBeenCalledOnce(); expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBeUndefined(); expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBeUndefined(); expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryCreatedAt).toBeUndefined(); @@ -258,25 +275,33 @@ describe("dispatchReplyFromConfig reply_dispatch hook", () => { existing: sessionStoreMocks.currentEntry, }); const abortController = new AbortController(); - const dispatcher = createDispatcher(); - vi.mocked(dispatcher.sendFinalReply).mockImplementation(() => { + const deliver = vi.fn().mockResolvedValue(undefined); + const dispatcher = createReplyDispatcher({ deliver }); + const sendFinalReply = dispatcher.sendFinalReply.bind(dispatcher); + vi.spyOn(dispatcher, "sendFinalReply").mockImplementation((payload) => { + const queued = sendFinalReply(payload); abortController.abort(); - return true; + return queued; }); - const result = await dispatchReplyFromConfig({ - ctx: createHookCtx(), - cfg: emptyConfig, + const result = await withReplyDispatcher({ dispatcher, - replyOptions: { abortSignal: abortController.signal }, - replyResolver: async () => ({ text: "durable reply" }), + run: () => + dispatchReplyFromConfig({ + ctx: createHookCtx(), + cfg: emptyConfig, + dispatcher, + replyOptions: { abortSignal: abortController.signal }, + replyResolver: async () => ({ text: "durable reply" }), + }), }); // Abort landed after delivery: the run is still surfaced as aborted // (queuedFinal:false), but the pending-final state is fully cleared. expect(dispatcher.sendFinalReply).toHaveBeenCalledOnce(); + expect(deliver).toHaveBeenCalledOnce(); expect(result.queuedFinal).toBe(false); - expect(sessionStoreMocks.updateSessionStoreEntry).toHaveBeenCalledOnce(); + expect(sessionStoreMocks.updateSessionEntry).toHaveBeenCalledOnce(); expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBeUndefined(); expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBeUndefined(); expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryCreatedAt).toBeUndefined(); @@ -309,12 +334,421 @@ describe("dispatchReplyFromConfig reply_dispatch hook", () => { }); expect(result.queuedFinal).toBe(false); - expect(sessionStoreMocks.updateSessionStoreEntry).not.toHaveBeenCalled(); + expect(sessionStoreMocks.updateSessionEntry).not.toHaveBeenCalled(); expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBe(true); expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBe("durable reply"); expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryCreatedAt).toBe(1); }); + it("preserves pending final delivery when beforeDeliver times out", async () => { + vi.useFakeTimers(); + try { + hookMocks.runner.hasHooks.mockReturnValue(false); + sessionStoreMocks.currentEntry = { + sessionKey: "agent:test:session", + pendingFinalDelivery: true, + pendingFinalDeliveryText: "durable reply", + pendingFinalDeliveryCreatedAt: 1, + pendingFinalDeliveryContext: { channel: "whatsapp", to: "+1000" }, + }; + sessionStoreMocks.resolveSessionStoreEntry.mockReturnValue({ + existing: sessionStoreMocks.currentEntry, + }); + const hookStarted = createDeferred(); + const deliver = vi.fn().mockResolvedValue(undefined); + const dispatcher = createReplyDispatcher({ + deliver, + beforeDeliver: () => { + hookStarted.resolve(); + return new Promise(() => {}); + }, + }); + + const resultPromise = withReplyDispatcher({ + dispatcher, + run: () => + dispatchReplyFromConfig({ + ctx: createHookCtx(), + cfg: emptyConfig, + dispatcher, + replyResolver: async () => ({ text: "durable reply" }), + }), + }); + await hookStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + const result = await resultPromise; + + expect(result.queuedFinal).toBe(true); + expect(deliver).not.toHaveBeenCalled(); + expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 0, final: 1 }); + expect(sessionStoreMocks.updateSessionEntry).toHaveBeenCalledOnce(); + expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBe(true); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBe("durable reply"); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryContext).toEqual({ + channel: "whatsapp", + to: "+1000", + }); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("clears pending final delivery when a later queued final succeeds", async () => { + vi.useFakeTimers(); + try { + hookMocks.runner.hasHooks.mockReturnValue(false); + sessionStoreMocks.currentEntry = { + sessionKey: "agent:test:session", + pendingFinalDelivery: true, + pendingFinalDeliveryText: "durable reply", + pendingFinalDeliveryCreatedAt: 1, + }; + sessionStoreMocks.resolveSessionStoreEntry.mockReturnValue({ + existing: sessionStoreMocks.currentEntry, + }); + const hookStarted = createDeferred(); + const deliver = vi.fn().mockResolvedValue(undefined); + let hookCalls = 0; + const dispatcher = createReplyDispatcher({ + deliver, + beforeDeliver: (payload) => { + hookCalls += 1; + if (hookCalls === 1) { + hookStarted.resolve(); + return new Promise(() => {}); + } + return payload; + }, + }); + + const resultPromise = withReplyDispatcher({ + dispatcher, + run: () => + dispatchReplyFromConfig({ + ctx: createHookCtx(), + cfg: emptyConfig, + dispatcher, + replyResolver: async () => [{ text: "first" }, { text: "durable reply" }], + }), + }); + await hookStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + await resultPromise; + + expect(deliver).toHaveBeenCalledOnce(); + expect(deliver).toHaveBeenCalledWith( + expect.objectContaining({ text: "durable reply" }), + expect.objectContaining({ kind: "final" }), + ); + expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 0, final: 1 }); + expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBeUndefined(); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBeUndefined(); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("preserves the durable final when an earlier auxiliary final succeeds", async () => { + vi.useFakeTimers(); + try { + hookMocks.runner.hasHooks.mockReturnValue(false); + sessionStoreMocks.currentEntry = { + sessionKey: "agent:test:session", + pendingFinalDelivery: true, + pendingFinalDeliveryText: "durable reply", + pendingFinalDeliveryCreatedAt: 1, + }; + sessionStoreMocks.resolveSessionStoreEntry.mockReturnValue({ + existing: sessionStoreMocks.currentEntry, + }); + const hookStarted = createDeferred(); + const deliver = vi.fn().mockResolvedValue(undefined); + let hookCalls = 0; + const dispatcher = createReplyDispatcher({ + deliver, + beforeDeliver: (payload) => { + hookCalls += 1; + if (hookCalls === 2) { + hookStarted.resolve(); + return new Promise(() => {}); + } + return payload; + }, + }); + + const resultPromise = withReplyDispatcher({ + dispatcher, + run: () => + dispatchReplyFromConfig({ + ctx: createHookCtx(), + cfg: emptyConfig, + dispatcher, + replyResolver: async () => [{ text: "auxiliary" }, { text: "durable reply" }], + }), + }); + await hookStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + await resultPromise; + + expect(deliver).toHaveBeenCalledOnce(); + expect(deliver).toHaveBeenCalledWith( + expect.objectContaining({ text: "auxiliary" }), + expect.objectContaining({ kind: "final" }), + ); + expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBe(true); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBe("durable reply"); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("narrows combined retry text to finals that failed before transport", async () => { + vi.useFakeTimers(); + try { + hookMocks.runner.hasHooks.mockReturnValue(false); + sessionStoreMocks.currentEntry = { + sessionKey: "agent:test:session", + pendingFinalDelivery: true, + pendingFinalDeliveryText: "auxiliary\n\ndurable reply", + pendingFinalDeliveryCreatedAt: 1, + }; + sessionStoreMocks.resolveSessionStoreEntry.mockReturnValue({ + existing: sessionStoreMocks.currentEntry, + }); + const hookStarted = createDeferred(); + let hookCalls = 0; + const dispatcher = createReplyDispatcher({ + deliver: vi.fn().mockResolvedValue(undefined), + beforeDeliver: (payload) => { + hookCalls += 1; + if (hookCalls === 2) { + hookStarted.resolve(); + return new Promise(() => {}); + } + return payload; + }, + }); + + const resultPromise = withReplyDispatcher({ + dispatcher, + run: () => + dispatchReplyFromConfig({ + ctx: createHookCtx(), + cfg: emptyConfig, + dispatcher, + replyResolver: async () => [{ text: "auxiliary" }, { text: "durable reply" }], + }), + }); + await hookStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + await resultPromise; + + expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBe(true); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBe("durable reply"); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryCreatedAt).toBe(1); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("narrows heartbeat-normalized retry text using its originating payloads", async () => { + vi.useFakeTimers(); + try { + hookMocks.runner.hasHooks.mockReturnValue(false); + sessionStoreMocks.currentEntry = { + sessionKey: "agent:test:session", + pendingFinalDelivery: true, + pendingFinalDeliveryText: "auxiliary durable reply", + pendingFinalDeliveryCreatedAt: 1, + pendingFinalDeliveryIntentId: "heartbeat-intent", + }; + sessionStoreMocks.resolveSessionStoreEntry.mockReturnValue({ + existing: sessionStoreMocks.currentEntry, + }); + const hookStarted = createDeferred(); + let hookCalls = 0; + const dispatcher = createReplyDispatcher({ + deliver: vi.fn().mockResolvedValue(undefined), + beforeDeliver: (payload) => { + hookCalls += 1; + if (hookCalls === 2) { + hookStarted.resolve(); + return new Promise(() => {}); + } + return payload; + }, + }); + + const resultPromise = withReplyDispatcher({ + dispatcher, + run: () => + dispatchReplyFromConfig({ + ctx: createHookCtx(), + cfg: emptyConfig, + dispatcher, + replyResolver: async () => [ + setReplyPayloadMetadata( + { text: "auxiliary" }, + { + pendingFinalDeliveryIntentId: "heartbeat-intent", + pendingFinalDeliveryRetryText: "auxiliary", + }, + ), + setReplyPayloadMetadata( + { text: "durable reply" }, + { + pendingFinalDeliveryIntentId: "heartbeat-intent", + pendingFinalDeliveryRetryText: "durable reply", + }, + ), + ], + }), + }); + await hookStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + await resultPromise; + + expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBe(true); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBe("durable reply"); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryCreatedAt).toBe(1); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryIntentId).toBe("heartbeat-intent"); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("does not let an older settlement rewrite a newer pending-final intent", async () => { + vi.useFakeTimers(); + try { + hookMocks.runner.hasHooks.mockReturnValue(false); + sessionStoreMocks.currentEntry = { + sessionKey: "agent:test:session", + pendingFinalDelivery: true, + pendingFinalDeliveryText: "older reply", + pendingFinalDeliveryCreatedAt: 1, + pendingFinalDeliveryIntentId: "older-intent", + }; + sessionStoreMocks.resolveSessionStoreEntry.mockReturnValue({ + existing: sessionStoreMocks.currentEntry, + }); + const hookStarted = createDeferred(); + const dispatcher = createReplyDispatcher({ + deliver: vi.fn().mockResolvedValue(undefined), + beforeDeliver: () => { + hookStarted.resolve(); + return new Promise(() => {}); + }, + }); + + const resultPromise = withReplyDispatcher({ + dispatcher, + run: () => + dispatchReplyFromConfig({ + ctx: createHookCtx(), + cfg: emptyConfig, + dispatcher, + replyResolver: async () => + setReplyPayloadMetadata( + { text: "older reply" }, + { pendingFinalDeliveryIntentId: "older-intent" }, + ), + }), + }); + await hookStarted.promise; + sessionStoreMocks.currentEntry = { + ...sessionStoreMocks.currentEntry, + pendingFinalDeliveryText: "newer reply", + pendingFinalDeliveryCreatedAt: 2, + pendingFinalDeliveryIntentId: "newer-intent", + }; + await vi.advanceTimersByTimeAsync(15_000); + await resultPromise; + + expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBe(true); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBe("newer reply"); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryCreatedAt).toBe(2); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryIntentId).toBe("newer-intent"); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("clears pending final delivery after transport delivery has started", async () => { + hookMocks.runner.hasHooks.mockReturnValue(false); + sessionStoreMocks.currentEntry = { + sessionKey: "agent:test:session", + pendingFinalDelivery: true, + pendingFinalDeliveryText: "possibly visible reply", + pendingFinalDeliveryCreatedAt: 1, + }; + sessionStoreMocks.resolveSessionStoreEntry.mockReturnValue({ + existing: sessionStoreMocks.currentEntry, + }); + const dispatcher = createReplyDispatcher({ + deliver: async () => { + throw new Error("transport failed after send started"); + }, + }); + + await withReplyDispatcher({ + dispatcher, + run: () => + dispatchReplyFromConfig({ + ctx: createHookCtx(), + cfg: emptyConfig, + dispatcher, + replyResolver: async () => ({ text: "possibly visible reply" }), + }), + }); + + expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 0, final: 1 }); + expect(sessionStoreMocks.updateSessionEntry).toHaveBeenCalledOnce(); + expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBeUndefined(); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBeUndefined(); + }); + + it("clears pending final delivery after intentional pre-delivery cancellation", async () => { + hookMocks.runner.hasHooks.mockReturnValue(false); + sessionStoreMocks.currentEntry = { + sessionKey: "agent:test:session", + pendingFinalDelivery: true, + pendingFinalDeliveryText: "policy-suppressed reply", + pendingFinalDeliveryCreatedAt: 1, + }; + sessionStoreMocks.resolveSessionStoreEntry.mockReturnValue({ + existing: sessionStoreMocks.currentEntry, + }); + const deliver = vi.fn().mockResolvedValue(undefined); + const dispatcher = createReplyDispatcher({ + deliver, + beforeDeliver: () => null, + }); + + const result = await dispatchReplyFromConfig({ + ctx: createHookCtx(), + cfg: emptyConfig, + dispatcher, + replyResolver: async () => ({ text: "policy-suppressed reply" }), + }); + await dispatcher.waitForIdle(); + await vi.waitFor(() => { + expect(sessionStoreMocks.updateSessionEntry).toHaveBeenCalledOnce(); + }); + + expect(result.queuedFinal).toBe(true); + expect(deliver).not.toHaveBeenCalled(); + expect(dispatcher.getCancelledCounts?.()).toEqual({ tool: 0, block: 0, final: 1 }); + expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 0, final: 0 }); + expect(sessionStoreMocks.currentEntry?.pendingFinalDelivery).toBeUndefined(); + expect(sessionStoreMocks.currentEntry?.pendingFinalDeliveryText).toBeUndefined(); + }); + it("delivers a generated final reply before queued follow-up admission", async () => { hookMocks.runner.hasHooks.mockReturnValue(false); const dispatcher = createDispatcher(); 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 c5d76cb04241..25233de0cdb9 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 @@ -125,6 +125,22 @@ const sessionStoreMocks = vi.hoisted(() => ({ return sessionStoreMocks.currentEntry; }, ), + updateSessionEntry: vi.fn( + async ( + _scope: unknown, + update: (entry: Record) => Promise | null>, + ) => { + if (!sessionStoreMocks.currentEntry) { + return null; + } + const patch = await update(sessionStoreMocks.currentEntry); + if (!patch) { + return sessionStoreMocks.currentEntry; + } + sessionStoreMocks.currentEntry = { ...sessionStoreMocks.currentEntry, ...patch }; + return sessionStoreMocks.currentEntry; + }, + ), })); const acpManagerRuntimeMocks = vi.hoisted(() => ({ getAcpSessionManager: vi.fn(), @@ -241,6 +257,11 @@ vi.mock("../../config/sessions/session-accessor.js", async (importOriginal) => { return { ...actual, loadSessionEntry: (...args: unknown[]) => sessionStoreMocks.loadSessionEntry(...args), + updateSessionEntry: (scope: unknown, update: unknown) => + sessionStoreMocks.updateSessionEntry( + scope, + update as Parameters[1], + ), }; }); vi.mock("./dispatch-from-config.runtime.js", () => ({ diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts index ef234649cc7b..02f0356ebbf5 100644 --- a/src/auto-reply/reply/dispatch-from-config.ts +++ b/src/auto-reply/reply/dispatch-from-config.ts @@ -57,6 +57,7 @@ import { shouldSuppressLocalExecApprovalPrompt } from "../../channels/plugins/ex import { applyMergePatch } from "../../config/merge-patch.js"; import { normalizeExplicitSessionKey } from "../../config/sessions/explicit-session-key-normalization.js"; import { resolveGroupSessionKey } from "../../config/sessions/group.js"; +import { loadSessionEntry, updateSessionEntry } from "../../config/sessions/session-accessor.js"; import { isRecoverableTerminalSessionStatus } from "../../config/sessions/terminal-status.js"; import { appendAssistantMessageToSessionTranscript, @@ -144,7 +145,6 @@ import { loadSessionStoreEntry, resolveStorePath, triggerInternalHook, - updateSessionStoreEntry, } from "./dispatch-from-config.runtime.js"; import type { DispatchFromConfigParams, @@ -156,8 +156,14 @@ import type { ReplySessionBinding } from "./get-reply.types.js"; import { claimInboundDedupe, commitInboundDedupe, releaseInboundDedupe } from "./inbound-dedupe.js"; import { hasInboundAudio } from "./inbound-media.js"; import { resolveOriginMessageProvider } from "./origin-routing.js"; +import { + buildPendingFinalDeliveryText, + sanitizePendingFinalDeliveryText, +} from "./pending-final-delivery.js"; import { appendReplyDispatcherBeforeDeliverCancelled, + captureReplyDispatchDeliveryOutcome, + type ReplyDispatchDeliveryOutcome, waitForReplyDispatcherIdle, } from "./reply-dispatcher.js"; import type { @@ -1051,18 +1057,20 @@ function shouldBypassPluginOwnedBindingForCommand( } async function clearPendingFinalDeliveryAfterSuccess(params: { + identity?: PendingFinalDeliveryIdentity; storePath?: string; sessionKey?: string; }): Promise { - if (!params.storePath || !params.sessionKey) { + const identity = params.identity; + if (!params.storePath || !params.sessionKey || !identity?.present) { return; } - await updateSessionStoreEntry({ - storePath: params.storePath, - sessionKey: params.sessionKey, - skipMaintenance: true, - takeCacheOwnership: true, - update: async (entry) => { + await updateSessionEntry( + { storePath: params.storePath, sessionKey: params.sessionKey }, + async (entry) => { + if (!matchesPendingFinalDeliveryIdentity(entry, identity)) { + return null; + } if (!entry.pendingFinalDelivery && !entry.pendingFinalDeliveryText) { return null; } @@ -1078,7 +1086,193 @@ async function clearPendingFinalDeliveryAfterSuccess(params: { updatedAt: Date.now(), }; }, - }); + { skipMaintenance: true, takeCacheOwnership: true }, + ); +} + +type SettledFinalDelivery = { + outcome: ReplyDispatchDeliveryOutcome; + payload: ReplyPayload; +}; + +type PendingFinalDeliveryIdentity = { + createdAt?: number; + intentId?: string; + present: boolean; + text?: string; +}; + +function capturePendingFinalDeliveryIdentity(params: { + intentId?: string; + storePath?: string; + sessionKey?: string; +}): PendingFinalDeliveryIdentity | undefined { + if (!params.storePath || !params.sessionKey) { + return undefined; + } + try { + const entry = loadSessionEntry({ + storePath: params.storePath, + sessionKey: params.sessionKey, + hydrateSkillPromptRefs: false, + readConsistency: "latest", + }); + if ( + params.intentId && + normalizeOptionalString(entry?.pendingFinalDeliveryIntentId) !== params.intentId + ) { + return { present: false }; + } + return { + present: Boolean(entry?.pendingFinalDelivery || entry?.pendingFinalDeliveryText), + intentId: params.intentId ?? normalizeOptionalString(entry?.pendingFinalDeliveryIntentId), + createdAt: + typeof entry?.pendingFinalDeliveryCreatedAt === "number" + ? entry.pendingFinalDeliveryCreatedAt + : undefined, + text: normalizeOptionalString(entry?.pendingFinalDeliveryText), + }; + } catch { + return params.intentId ? { present: true, intentId: params.intentId } : undefined; + } +} + +function matchesPendingFinalDeliveryIdentity( + entry: SessionEntry, + expected: PendingFinalDeliveryIdentity, +): boolean { + const currentPresent = Boolean(entry.pendingFinalDelivery || entry.pendingFinalDeliveryText); + if (currentPresent !== expected.present) { + return false; + } + if (expected.intentId) { + return normalizeOptionalString(entry.pendingFinalDeliveryIntentId) === expected.intentId; + } + return ( + entry.pendingFinalDeliveryCreatedAt === expected.createdAt && + normalizeOptionalString(entry.pendingFinalDeliveryText) === expected.text + ); +} + +function resolvePendingFinalDeliveryPayloads(params: { + intentId?: string; + pendingText: string; + replies: ReplyPayload[]; +}): ReplyPayload[] | undefined { + const intentReplies = params.intentId + ? params.replies.filter((reply) => { + const metadata = getReplyPayloadMetadata(reply); + return ( + metadata?.pendingFinalDeliveryIntentId === params.intentId && + metadata?.pendingFinalDeliveryRetryText !== undefined + ); + }) + : []; + const intentContributors = intentReplies.filter( + (reply) => getReplyPayloadMetadata(reply)?.pendingFinalDeliveryRetryText, + ); + const intentText = buildPendingFinalDeliveryRetryText(intentContributors); + if ( + intentReplies.length > 0 && + intentText.replace(/\s+/g, " ").trim() === params.pendingText.replace(/\s+/g, " ").trim() + ) { + return intentContributors; + } + const contributingReplies = params.replies.filter( + (reply) => buildPendingFinalDeliveryText([reply]) !== "", + ); + if (buildPendingFinalDeliveryText(contributingReplies) === params.pendingText) { + return contributingReplies; + } + const exactMatches = contributingReplies.filter( + (reply) => buildPendingFinalDeliveryText([reply]) === params.pendingText, + ); + return exactMatches.length === 1 ? exactMatches : undefined; +} + +function buildPendingFinalDeliveryRetryText(payloads: ReplyPayload[]): string { + return sanitizePendingFinalDeliveryText( + payloads + .map( + (payload) => + getReplyPayloadMetadata(payload)?.pendingFinalDeliveryRetryText ?? + buildPendingFinalDeliveryText([payload]), + ) + .filter(Boolean) + .join("\n\n"), + ); +} + +async function reconcilePendingFinalDeliveryAfterSettlement(params: { + deliveries: SettledFinalDelivery[]; + identity?: PendingFinalDeliveryIdentity; + replies: ReplyPayload[]; + storePath?: string; + sessionKey?: string; +}): Promise { + const identity = params.identity; + if (!params.storePath || !params.sessionKey || !identity?.present) { + return; + } + await updateSessionEntry( + { storePath: params.storePath, sessionKey: params.sessionKey }, + async (entry) => { + if (!matchesPendingFinalDeliveryIdentity(entry, identity)) { + return null; + } + const pendingText = normalizeOptionalString(entry.pendingFinalDeliveryText); + if (!entry.pendingFinalDelivery && !pendingText) { + return null; + } + const pendingPayloads = pendingText + ? resolvePendingFinalDeliveryPayloads({ + intentId: identity.intentId, + pendingText, + replies: params.replies, + }) + : undefined; + const pendingPayloadSet = pendingPayloads ? new Set(pendingPayloads) : undefined; + const relevantDeliveries = pendingPayloadSet + ? params.deliveries.filter((delivery) => pendingPayloadSet.has(delivery.payload)) + : params.deliveries; + const ownsEveryPendingPayload = + !pendingPayloadSet || relevantDeliveries.length === pendingPayloadSet.size; + const failedBeforeDeliver = relevantDeliveries.filter( + (delivery) => delivery.outcome === "failed-before-deliver", + ); + + if ( + relevantDeliveries.length > 0 && + failedBeforeDeliver.length === relevantDeliveries.length + ) { + return null; + } + if (pendingPayloadSet && ownsEveryPendingPayload && failedBeforeDeliver.length > 0) { + const retryText = buildPendingFinalDeliveryRetryText( + failedBeforeDeliver.map((delivery) => delivery.payload), + ); + if (retryText) { + return { + pendingFinalDelivery: true, + pendingFinalDeliveryText: retryText, + updatedAt: Date.now(), + }; + } + } + return { + pendingFinalDelivery: undefined, + pendingFinalDeliveryText: undefined, + pendingFinalDeliveryCreatedAt: undefined, + pendingFinalDeliveryLastAttemptAt: undefined, + pendingFinalDeliveryAttemptCount: undefined, + pendingFinalDeliveryLastError: undefined, + pendingFinalDeliveryContext: undefined, + pendingFinalDeliveryIntentId: undefined, + updatedAt: Date.now(), + }; + }, + { skipMaintenance: true, takeCacheOwnership: true }, + ); } async function mirrorDeliveredReplyToTranscript(params: { @@ -2958,7 +3152,11 @@ async function dispatchReplyFromConfigInner( const sendFinalPayload = async ( payload: ReplyPayload, options: { abortSignal?: AbortSignal; deliveryId?: string } = {}, - ): Promise<{ queuedFinal: boolean; routedFinalCount: number }> => { + ): Promise<{ + queuedFinal: boolean; + routedFinalCount: number; + dispatcherOutcome?: Promise; + }> => { const abortSignal = options.abortSignal ?? getDispatchAbortSignal(); const throwIfFinalDeliveryAborted = () => { if (abortSignal?.aborted) { @@ -3081,7 +3279,10 @@ async function dispatchReplyFromConfigInner( if (finalDeliveryCapture) { setReplyPayloadMetadata(normalizedPayload, { finalDeliveryCapture }); } + const deliveryOutcome = captureReplyDispatchDeliveryOutcome(normalizedPayload); const queuedFinal = dispatcher.sendFinalReply(normalizedPayload); + const dispatcherOutcome = + queuedFinal && deliveryOutcome.isTracked() ? deliveryOutcome.promise : undefined; if (queuedFinal && deliveredTranscriptMirror && finalOutcomeBefore) { // The common settle owner runs this after successful delivery or // cancellation. Keeping reconciliation out of the reply operation lets a @@ -3098,6 +3299,7 @@ async function dispatchReplyFromConfigInner( return { queuedFinal, routedFinalCount: 0, + ...(queuedFinal && dispatcherOutcome ? { dispatcherOutcome } : {}), }; }; @@ -4138,6 +4340,19 @@ async function dispatchReplyFromConfigInner( } const replies = replyResult ? (Array.isArray(replyResult) ? replyResult : [replyResult]) : []; + const pendingFinalDelivery = { + storePath: sessionStoreEntry.storePath, + sessionKey: sessionStoreEntry.sessionKey ?? sessionKey, + }; + const replyPendingIntentIds = new Set( + replies + .map((reply) => getReplyPayloadMetadata(reply)?.pendingFinalDeliveryIntentId) + .filter((intentId): intentId is string => Boolean(intentId)), + ); + const pendingFinalDeliveryIdentity = capturePendingFinalDeliveryIdentity({ + ...pendingFinalDelivery, + intentId: replyPendingIntentIds.size === 1 ? [...replyPendingIntentIds][0] : undefined, + }); // Final delivery is outside the progress wrappers. Wait until every source-ordered callback // has at least started so a delayed tool/reasoning transition cannot appear after the final. if (preserveProgressCallbackStartOrder) { @@ -4154,6 +4369,11 @@ async function dispatchReplyFromConfigInner( let routedFinalCount = 0; let attemptedFinalDelivery = false; let finalDeliveryFailed = false; + const finalDeliveries: Array<{ + outcome: Promise; + payload: ReplyPayload; + }> = []; + let allQueuedFinalsObserved = true; // Explicit command turns (native or authorized text-slash like /compact) are // user-initiated, so a marked terminal reply for the command bypasses // room_event suppression. Ambient marked notices (no CommandTurn) stay @@ -4202,20 +4422,52 @@ async function dispatchReplyFromConfigInner( const finalReply = await sendFinalPayload(reply, { deliveryId: String(replyIndex) }); queuedFinal = finalReply.queuedFinal || queuedFinal; routedFinalCount += finalReply.routedFinalCount; + if (finalReply.queuedFinal) { + if (finalReply.dispatcherOutcome) { + finalDeliveries.push({ outcome: finalReply.dispatcherOutcome, payload: reply }); + } else { + allQueuedFinalsObserved = false; + } + } if (!finalReply.queuedFinal && finalReply.routedFinalCount === 0) { finalDeliveryFailed = true; } } if (attemptedFinalDelivery && !finalDeliveryFailed) { - // The final reply already shipped, so clear the durable pending-final - // bookkeeping before honoring a late abort. A stuck-session recovery abort - // racing this window (#89115) otherwise strands pendingFinalDelivery=true, - // and the get-reply redelivery short-circuit then silently blocks all inbound. - await clearPendingFinalDeliveryAfterSuccess({ - storePath: sessionStoreEntry.storePath, - sessionKey: sessionStoreEntry.sessionKey ?? sessionKey, - }); + if (queuedFinal && allQueuedFinalsObserved) { + // Delivery observers run from the queue itself, so direct low-level callers + // reconcile too; the settle task only makes lifecycle owners await it. + const reconcilePendingFinal = Promise.all( + finalDeliveries.map(async (delivery) => ({ + outcome: await delivery.outcome, + payload: delivery.payload, + })), + ) + .then(async (deliveries) => { + await reconcilePendingFinalDeliveryAfterSettlement({ + ...pendingFinalDelivery, + deliveries, + identity: pendingFinalDeliveryIdentity, + replies, + }); + }) + .catch((error: unknown) => { + logVerbose( + `dispatch-from-config: pending final reconciliation failed: ${formatErrorMessage(error)}`, + ); + }); + registerReplyDispatcherSettledTask(dispatcher, () => reconcilePendingFinal); + } else { + // Routed delivery has a transport result already. Custom dispatchers that + // do not expose the core observer retain the legacy queue-admission behavior. + await clearPendingFinalDeliveryAfterSuccess({ + ...pendingFinalDelivery, + identity: pendingFinalDeliveryIdentity, + }); + } + // Register successful queued cleanup before honoring a late abort. The + // outer settle owner still runs it from finally (#89115). throwIfDispatchOperationAborted(); } diff --git a/src/auto-reply/reply/pending-final-delivery.ts b/src/auto-reply/reply/pending-final-delivery.ts index 6a5026979860..447b3294f36f 100644 --- a/src/auto-reply/reply/pending-final-delivery.ts +++ b/src/auto-reply/reply/pending-final-delivery.ts @@ -1,4 +1,3 @@ -/** Sanitizes pending final delivery text before channel-visible output. */ import { isSilentReplyPayloadText, isSilentReplyText, @@ -7,8 +6,20 @@ import { stripLeadingSilentToken, stripSilentToken, } from "../tokens.js"; +/** Sanitizes pending final delivery text before channel-visible output. */ +import type { ReplyPayload } from "../types.js"; import { stripInternalMetadataForDisplay } from "./display-text-sanitize.js"; +/** Build the restart-recovery text represented by one or more final payloads. */ +export function buildPendingFinalDeliveryText(payloads: ReplyPayload[]): string { + const text = payloads + .filter((payload) => payload.isReasoning !== true) + .map((payload) => payload.text) + .filter((textLocal): textLocal is string => Boolean(textLocal)) + .join("\n\n"); + return sanitizePendingFinalDeliveryText(text); +} + /** Sanitizes final pending-delivery text and removes silent control tokens. */ export function sanitizePendingFinalDeliveryText(text: string): string { let stripped = stripInternalMetadataForDisplay(text).trim(); diff --git a/src/auto-reply/reply/reply-dispatcher.ts b/src/auto-reply/reply/reply-dispatcher.ts index 5f5901772bc6..e873779a02bc 100644 --- a/src/auto-reply/reply/reply-dispatcher.ts +++ b/src/auto-reply/reply/reply-dispatcher.ts @@ -13,6 +13,7 @@ import { registerDispatcher } from "./dispatcher-registry.js"; import { normalizeReplyPayload, type NormalizeReplySkipReason } from "./normalize-reply.js"; import type { ReplyDispatchBeforeDeliver, + ReplyDispatchBeforeDeliverOptions, ReplyDispatchKind, ReplyDispatchRuntimeInfo, ReplyDispatcher, @@ -38,6 +39,18 @@ type ReplyDispatchCancelHandler = ( info: ReplyDispatchRuntimeInfo, ) => Promise | void; +export type ReplyDispatchDeliveryOutcome = + | "delivered" + | "cancelled" + | "failed-before-deliver" + | "failed-deliver"; + +type ReplyDispatchDeliveryOutcomeTracker = { + promise: Promise; + resolve: (outcome: ReplyDispatchDeliveryOutcome) => void; + tracked: boolean; +}; + type ReplyDispatchDeliverer = ( payload: ReplyPayload, info: ReplyDispatchRuntimeInfo, @@ -47,8 +60,136 @@ export type { ReplyDispatchBeforeDeliver }; const DEFAULT_HUMAN_DELAY_MIN_MS = 800; const DEFAULT_HUMAN_DELAY_MAX_MS = 2500; +const DEFAULT_BEFORE_DELIVER_TIMEOUT_MS = 15_000; const silentReplyLogger = createSubsystemLogger("silent-reply/dispatcher"); const beforeDeliverCancelledHooks = new WeakMap(); +const deliveryOutcomeTrackers = new WeakMap(); + +type ReplyDispatchBeforeDeliverStage = { + hook: ReplyDispatchBeforeDeliver; + timeoutMs?: number; +}; + +type ReplyDispatchBeforeDeliverStageInput = + | ReplyDispatchBeforeDeliver + | { + hook: ReplyDispatchBeforeDeliver; + options?: ReplyDispatchBeforeDeliverOptions; + } + | undefined; + +const beforeDeliverStagesByHook = new WeakMap< + ReplyDispatchBeforeDeliver, + readonly ReplyDispatchBeforeDeliverStage[] +>(); + +class ReplyDispatchBeforeDeliverTimeoutError extends Error { + constructor(timeoutMs: number) { + super(`beforeDeliver timed out after ${timeoutMs}ms`); + this.name = "ReplyDispatchBeforeDeliverTimeoutError"; + } +} + +function resolveReplyDispatchBeforeDeliverTimeoutMs( + options: ReplyDispatchBeforeDeliverOptions | undefined, +): number { + const timeoutMs = options?.timeoutMs ?? DEFAULT_BEFORE_DELIVER_TIMEOUT_MS; + if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) { + throw new RangeError("beforeDeliver timeoutMs must be a positive finite number"); + } + return timeoutMs; +} + +async function runReplyDispatchBeforeDeliverStage( + stage: ReplyDispatchBeforeDeliverStage, + payload: ReplyPayload, + info: ReplyDispatchRuntimeInfo, +): Promise { + const timeoutMs = stage.timeoutMs; + if (!timeoutMs) { + return await stage.hook(payload, info); + } + let timer: ReturnType | undefined; + // The hook promise cannot be cancelled. The deadline releases the serialized + // delivery owner; Promise.race still observes any late rejection. + const timeout = new Promise((_, reject) => { + timer = setTimeout( + () => reject(new ReplyDispatchBeforeDeliverTimeoutError(timeoutMs)), + timeoutMs, + ); + timer.unref?.(); + }); + try { + return await Promise.race([Promise.resolve(stage.hook(payload, info)), timeout]); + } finally { + if (timer) { + clearTimeout(timer); + } + } +} + +function resolveReplyDispatchBeforeDeliverStages( + input: ReplyDispatchBeforeDeliverStageInput, +): readonly ReplyDispatchBeforeDeliverStage[] { + if (!input) { + return []; + } + if (typeof input === "function") { + return ( + beforeDeliverStagesByHook.get(input) ?? [ + { hook: input, timeoutMs: DEFAULT_BEFORE_DELIVER_TIMEOUT_MS }, + ] + ); + } + const existingStages = beforeDeliverStagesByHook.get(input.hook); + // Internal composition already assigned each real stage its owner budget. + // Wrapping that chain again would turn one stage budget into an aggregate deadline. + if (existingStages) { + return existingStages; + } + return [ + { + hook: input.hook, + timeoutMs: resolveReplyDispatchBeforeDeliverTimeoutMs(input.options), + }, + ]; +} + +/** Compose core delivery stages while retaining a separate deadline for each actual hook. */ +export function composeReplyDispatchBeforeDeliver( + ...hooks: ReplyDispatchBeforeDeliverStageInput[] +): ReplyDispatchBeforeDeliver | undefined { + const stages: ReplyDispatchBeforeDeliverStage[] = []; + for (const hook of hooks) { + if (hook) { + stages.push(...resolveReplyDispatchBeforeDeliverStages(hook)); + } + } + if (stages.length === 0) { + return undefined; + } + const composed: ReplyDispatchBeforeDeliver = async (payload, info) => { + let current: ReplyPayload | null = payload; + for (const stage of stages) { + if (!current) { + return null; + } + const next = await runReplyDispatchBeforeDeliverStage(stage, current, info); + current = next ? copyReplyPayloadMetadata(current, next) : null; + } + return current; + }; + beforeDeliverStagesByHook.set(composed, stages); + return composed; +} + +/** Mark a core hook whose lifecycle owner controls settlement and any deadline. */ +export function markReplyDispatchBeforeDeliverDeadlineOwned( + hook: ReplyDispatchBeforeDeliver, +): ReplyDispatchBeforeDeliver { + beforeDeliverStagesByHook.set(hook, [{ hook }]); + return hook; +} /** Adds a core-internal cancellation observer without expanding the plugin-facing dispatcher. */ export function appendReplyDispatcherBeforeDeliverCancelled( @@ -63,6 +204,23 @@ export function appendReplyDispatcherBeforeDeliverCancelled( return true; } +/** Capture one core-dispatcher delivery outcome without changing send* return types. */ +export function captureReplyDispatchDeliveryOutcome(payload: ReplyPayload): { + promise: Promise; + isTracked: () => boolean; +} { + let resolveOutcome!: (outcome: ReplyDispatchDeliveryOutcome) => void; + const tracker: ReplyDispatchDeliveryOutcomeTracker = { + promise: new Promise((resolve) => { + resolveOutcome = resolve; + }), + resolve: (outcome) => resolveOutcome(outcome), + tracked: false, + }; + deliveryOutcomeTrackers.set(payload, tracker); + return { promise: tracker.promise, isTracked: () => tracker.tracked }; +} + function buildReplyDispatchRuntimeInfo( payload: ReplyPayload, kind: ReplyDispatchKind, @@ -125,6 +283,8 @@ export type ReplyDispatcherOptions = { /** Human-like delay between block replies for natural rhythm. */ humanDelay?: HumanDelayConfig; beforeDeliver?: ReplyDispatchBeforeDeliver; + /** Owner-declared deadline for the constructor before-delivery callback. */ + beforeDeliverOptions?: ReplyDispatchBeforeDeliverOptions; onBeforeDeliverCancelled?: ReplyDispatchCancelHandler; /** Observe each queued payload settling, including cancellation and delivery failure. */ onDeliverySettled?: (info: ReplyDispatchRuntimeInfo) => void; @@ -181,7 +341,11 @@ function normalizeReplyPayloadInternal( } export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDispatcher { - let beforeDeliver = options.beforeDeliver; + let beforeDeliver = composeReplyDispatchBeforeDeliver( + options.beforeDeliver + ? { hook: options.beforeDeliver, options: options.beforeDeliverOptions } + : undefined, + ); const appendedBeforeDeliverCancelledHooks: ReplyDispatchCancelHandler[] = []; let sendChain: Promise = Promise.resolve(); // Track in-flight deliveries so we can emit a reliable "idle" signal. @@ -228,7 +392,17 @@ export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDis ]; for (const observer of observers) { try { - await observer(payload, info); + await runReplyDispatchBeforeDeliverStage( + { + hook: async (current, currentInfo) => { + await observer(current, currentInfo); + return current; + }, + timeoutMs: DEFAULT_BEFORE_DELIVER_TIMEOUT_MS, + }, + payload, + info, + ); } catch (err: unknown) { reportObserverError(err, info); } @@ -261,12 +435,18 @@ export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDis } queuedCounts[kind] += 1; pending += 1; + const deliveryOutcomeTracker = deliveryOutcomeTrackers.get(payload); + if (deliveryOutcomeTracker) { + deliveryOutcomeTracker.tracked = true; + } // Determine if we should add human-like delay (only for block replies after the first). const shouldDelay = kind === "block" && sentFirstBlock; if (kind === "block") { sentFirstBlock = true; } + let deliveryStarted = false; + let deliveryOutcome: ReplyDispatchDeliveryOutcome = "failed-before-deliver"; sendChain = sendChain .then(async () => { @@ -287,20 +467,26 @@ export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDis throw err; } if (!deliverPayload) { + deliveryOutcome = "cancelled"; cancelledCounts[kind] += 1; await notifyBeforeDeliverCancelled(normalized, dispatchInfo); return; } deliverPayload = copyReplyPayloadMetadata(normalized, deliverPayload); } + deliveryStarted = true; await options.deliver(deliverPayload, dispatchInfo); + deliveryOutcome = "delivered"; }) .catch((err: unknown) => { + deliveryOutcome = deliveryStarted ? "failed-deliver" : "failed-before-deliver"; failedCounts[kind] += 1; void options.onError?.(err, buildReplyDispatchRuntimeInfo(normalized, kind)); }) .finally(() => { const dispatchInfo = buildReplyDispatchRuntimeInfo(normalized, kind); + deliveryOutcomeTracker?.resolve(deliveryOutcome); + deliveryOutcomeTrackers.delete(payload); try { options.onDeliverySettled?.(dispatchInfo); } catch (err: unknown) { @@ -347,16 +533,11 @@ export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDis sendToolResult: (payload) => enqueue("tool", payload), sendBlockReply: (payload) => enqueue("block", payload), sendFinalReply: (payload) => enqueue("final", payload), - appendBeforeDeliver: (hook) => { - const previousBeforeDeliver = beforeDeliver; - beforeDeliver = previousBeforeDeliver - ? async (payload, info) => { - const previousPayload = await previousBeforeDeliver(payload, info); - return previousPayload - ? hook(copyReplyPayloadMetadata(payload, previousPayload), info) - : null; - } - : hook; + appendBeforeDeliver: (hook, stageOptions) => { + beforeDeliver = composeReplyDispatchBeforeDeliver(beforeDeliver, { + hook, + options: stageOptions, + }); }, waitForIdle: () => sendChain, getQueuedCounts: () => ({ ...queuedCounts }), diff --git a/src/auto-reply/reply/reply-dispatcher.types.ts b/src/auto-reply/reply/reply-dispatcher.types.ts index f71429a11e72..6251ae47abe1 100644 --- a/src/auto-reply/reply/reply-dispatcher.types.ts +++ b/src/auto-reply/reply/reply-dispatcher.types.ts @@ -20,11 +20,20 @@ export type ReplyDispatchBeforeDeliver = ( info: ReplyDispatchRuntimeInfo, ) => Promise | ReplyPayload | null; +/** An owner-declared settlement budget for one before-delivery callback. */ +export type ReplyDispatchBeforeDeliverOptions = { + /** Positive finite per-callback deadline in milliseconds; omit for the dispatcher default. */ + timeoutMs?: number; +}; + export type ReplyDispatcher = { sendToolResult: (payload: ReplyPayload) => boolean; sendBlockReply: (payload: ReplyPayload) => boolean; sendFinalReply: (payload: ReplyPayload) => boolean; - appendBeforeDeliver?: (hook: ReplyDispatchBeforeDeliver) => void; + appendBeforeDeliver?: ( + hook: ReplyDispatchBeforeDeliver, + options?: ReplyDispatchBeforeDeliverOptions, + ) => void; waitForIdle: () => Promise; getQueuedCounts: () => Record; getCancelledCounts?: () => Record; diff --git a/src/auto-reply/reply/reply-flow.test.ts b/src/auto-reply/reply/reply-flow.test.ts index f75f210c2567..1c957205e08c 100644 --- a/src/auto-reply/reply/reply-flow.test.ts +++ b/src/auto-reply/reply/reply-flow.test.ts @@ -2,7 +2,11 @@ import { describe, expect, it, vi } from "vitest"; import type { OpenClawConfig } from "../../config/types.openclaw.js"; import { HEARTBEAT_TOKEN, SILENT_REPLY_TOKEN } from "../tokens.js"; -import { createReplyDispatcher, waitForReplyDispatcherIdle } from "./reply-dispatcher.js"; +import { + composeReplyDispatchBeforeDeliver, + createReplyDispatcher, + waitForReplyDispatcherIdle, +} from "./reply-dispatcher.js"; import { createReplyToModeFilter } from "./reply-threading.js"; type DeliverPayload = Parameters[0]["deliver"]>[0]; @@ -13,6 +17,14 @@ function deliveredText(deliver: DeliverMock, index = 0) { return payload?.text; } +function createDeferred() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + describe("createReplyDispatcher", () => { it("drops empty payloads and exact silent tokens without media", async () => { const deliver = vi.fn().mockResolvedValue(undefined); @@ -154,6 +166,244 @@ describe("createReplyDispatcher", () => { expect(delivered).toEqual(["tool", "block", "final"]); }); + it("releases the same dispatcher after a beforeDeliver timeout", async () => { + vi.useFakeTimers(); + try { + const hookStarted = createDeferred(); + const delivered: string[] = []; + const errors: string[] = []; + let hookCalls = 0; + const dispatcher = createReplyDispatcher({ + deliver: async (payload) => { + delivered.push(payload.text ?? ""); + }, + beforeDeliver: (payload) => { + hookCalls += 1; + if (hookCalls === 1) { + hookStarted.resolve(); + return new Promise(() => {}); + } + return payload; + }, + onError: (error) => { + errors.push(error instanceof Error ? error.message : String(error)); + }, + }); + + dispatcher.sendFinalReply({ text: "stuck final" }); + dispatcher.sendFinalReply({ text: "follow-up final" }); + dispatcher.markComplete(); + await hookStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + await dispatcher.waitForIdle(); + + expect(delivered).toEqual(["follow-up final"]); + expect(errors).toEqual(["beforeDeliver timed out after 15000ms"]); + expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 0, final: 1 }); + expect(dispatcher.getCancelledCounts?.()).toEqual({ tool: 0, block: 0, final: 0 }); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("bounds hooks appended after dispatcher construction", async () => { + vi.useFakeTimers(); + try { + const hookStarted = createDeferred(); + const delivered: string[] = []; + let hookCalls = 0; + const dispatcher = createReplyDispatcher({ + deliver: async (payload) => { + delivered.push(payload.text ?? ""); + }, + }); + dispatcher.appendBeforeDeliver?.((payload) => { + hookCalls += 1; + if (hookCalls === 1) { + hookStarted.resolve(); + return new Promise(() => {}); + } + return payload; + }); + + dispatcher.sendFinalReply({ text: "stuck final" }); + dispatcher.sendFinalReply({ text: "follow-up final" }); + dispatcher.markComplete(); + await hookStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + await dispatcher.waitForIdle(); + + expect(delivered).toEqual(["follow-up final"]); + expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 0, final: 1 }); + expect(dispatcher.getCancelledCounts?.()).toEqual({ tool: 0, block: 0, final: 0 }); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("rejects non-positive and non-finite beforeDeliver budgets", () => { + for (const timeoutMs of [0, -1, Number.NaN, Number.POSITIVE_INFINITY]) { + expect(() => + createReplyDispatcher({ + deliver: async () => {}, + beforeDeliver: (payload) => payload, + beforeDeliverOptions: { timeoutMs }, + }), + ).toThrow("beforeDeliver timeoutMs must be a positive finite number"); + + const dispatcher = createReplyDispatcher({ deliver: async () => {} }); + expect(() => dispatcher.appendBeforeDeliver?.((payload) => payload, { timeoutMs })).toThrow( + "beforeDeliver timeoutMs must be a positive finite number", + ); + } + }); + + it("honors owner-declared budgets for constructor and appended callbacks", async () => { + vi.useFakeTimers(); + try { + const delivered: string[] = []; + const errors: string[] = []; + const dispatcher = createReplyDispatcher({ + deliver: async (payload) => { + delivered.push(payload.text ?? ""); + }, + beforeDeliver: async (payload) => { + await new Promise((resolve) => { + setTimeout(resolve, 16_000); + }); + return { ...payload, text: `${payload.text}:constructor` }; + }, + beforeDeliverOptions: { timeoutMs: 20_000 }, + onError: (error) => { + errors.push(error instanceof Error ? error.message : String(error)); + }, + }); + dispatcher.appendBeforeDeliver?.( + async (payload) => { + await new Promise((resolve) => { + setTimeout(resolve, 16_000); + }); + return { ...payload, text: `${payload.text}:appended` }; + }, + { timeoutMs: 20_000 }, + ); + + dispatcher.sendFinalReply({ text: "final" }); + dispatcher.markComplete(); + await vi.advanceTimersByTimeAsync(15_000); + expect(delivered).toEqual([]); + await vi.advanceTimersByTimeAsync(1_000); + await vi.advanceTimersByTimeAsync(15_000); + expect(delivered).toEqual([]); + await vi.advanceTimersByTimeAsync(1_000); + await dispatcher.waitForIdle(); + + expect(delivered).toEqual(["final:constructor:appended"]); + expect(errors).toEqual([]); + expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 0, final: 0 }); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("does not turn a composed stage budget into a whole-chain deadline", async () => { + vi.useFakeTimers(); + try { + const delivered: string[] = []; + const beforeDeliver = composeReplyDispatchBeforeDeliver( + { + hook: async (payload) => { + await new Promise((resolve) => { + setTimeout(resolve, 16_000); + }); + return { ...payload, text: `${payload.text}:owner` }; + }, + options: { timeoutMs: 20_000 }, + }, + async (payload) => { + await new Promise((resolve) => { + setTimeout(resolve, 10_000); + }); + return { ...payload, text: `${payload.text}:plugin` }; + }, + ); + const dispatcher = createReplyDispatcher({ + deliver: async (payload) => { + delivered.push(payload.text ?? ""); + }, + beforeDeliver, + beforeDeliverOptions: { timeoutMs: 20_000 }, + }); + + dispatcher.sendFinalReply({ text: "final" }); + dispatcher.markComplete(); + await vi.advanceTimersByTimeAsync(20_000); + expect(delivered).toEqual([]); + await vi.advanceTimersByTimeAsync(6_000); + await dispatcher.waitForIdle(); + + expect(delivered).toEqual(["final:owner:plugin"]); + expect(dispatcher.getFailedCounts()).toEqual({ tool: 0, block: 0, final: 0 }); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("gives each beforeDeliver hook its own deadline", async () => { + vi.useFakeTimers(); + try { + const delivered: string[] = []; + const dispatcher = createReplyDispatcher({ + deliver: async (payload) => { + delivered.push(payload.text ?? ""); + }, + beforeDeliver: async (payload) => { + await new Promise((resolve) => { + setTimeout(resolve, 10_000); + }); + return { ...payload, text: `${payload.text}:first` }; + }, + }); + dispatcher.appendBeforeDeliver?.(async (payload) => { + await new Promise((resolve) => { + setTimeout(resolve, 10_000); + }); + return { ...payload, text: `${payload.text}:second` }; + }); + + dispatcher.sendFinalReply({ text: "final" }); + dispatcher.markComplete(); + await vi.advanceTimersByTimeAsync(10_000); + await vi.advanceTimersByTimeAsync(5_000); + expect(delivered).toEqual([]); + expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 0, final: 0 }); + await vi.advanceTimersByTimeAsync(5_000); + await dispatcher.waitForIdle(); + + expect(delivered).toEqual(["final:first:second"]); + expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 0, final: 0 }); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("applies a hook appended after enqueue before the chain starts", async () => { + const deliver = vi.fn().mockResolvedValue(undefined); + const dispatcher = createReplyDispatcher({ deliver }); + + dispatcher.sendFinalReply({ text: "queued" }); + dispatcher.appendBeforeDeliver?.((payload) => ({ ...payload, text: "appended" })); + dispatcher.markComplete(); + await dispatcher.waitForIdle(); + + expect(deliveredText(deliver)).toBe("appended"); + }); + it("fires onIdle when the queue drains", async () => { const deliver: Parameters[0]["deliver"] = async () => await Promise.resolve(); diff --git a/src/plugin-sdk/reply-runtime.ts b/src/plugin-sdk/reply-runtime.ts index ce47bb41f9cf..f242716eb500 100644 --- a/src/plugin-sdk/reply-runtime.ts +++ b/src/plugin-sdk/reply-runtime.ts @@ -46,6 +46,7 @@ export { createReplyDispatcherWithTyping, } from "../auto-reply/reply/reply-dispatcher.js"; export type { + ReplyDispatchBeforeDeliverOptions, ReplyDispatchKind, ReplyDispatcher, ReplyFollowupAdmissionBarrierTimeoutPolicy, diff --git a/src/plugins/contracts/plugin-sdk-subpaths.test.ts b/src/plugins/contracts/plugin-sdk-subpaths.test.ts index bf332f1d6a6a..571d823566ad 100644 --- a/src/plugins/contracts/plugin-sdk-subpaths.test.ts +++ b/src/plugins/contracts/plugin-sdk-subpaths.test.ts @@ -23,7 +23,11 @@ import type { PluginRuntime as CorePluginRuntime, } from "openclaw/plugin-sdk/core"; import * as providerEntrySdk from "openclaw/plugin-sdk/provider-entry"; -import type { GetReplyOptions as ReplyRuntimeGetReplyOptions } from "openclaw/plugin-sdk/reply-runtime"; +import type { + GetReplyOptions as ReplyRuntimeGetReplyOptions, + ReplyDispatchBeforeDeliverOptions as ReplyRuntimeBeforeDeliverOptions, + ReplyDispatcher as ReplyRuntimeDispatcher, +} from "openclaw/plugin-sdk/reply-runtime"; import * as zalouserSdk from "openclaw/plugin-sdk/zalouser"; import ts from "typescript"; import { beforeAll, describe, expect, expectTypeOf, it } from "vitest"; @@ -1345,6 +1349,12 @@ describe("plugin-sdk subpath exports", () => { "requestedSessionId" | "resumeRequestedSession" >; expectTypeOf().toEqualTypeOf(); + type ReplyRuntimeAppendBeforeDeliverOptions = Parameters< + NonNullable + >[1]; + expectTypeOf().toEqualTypeOf< + ReplyRuntimeBeforeDeliverOptions | undefined + >(); }); it("keeps runtime entry subpaths importable", async () => { diff --git a/src/plugins/hooks.test-helpers.ts b/src/plugins/hooks.test-helpers.ts index c039c5e03bf3..488db85125b3 100644 --- a/src/plugins/hooks.test-helpers.ts +++ b/src/plugins/hooks.test-helpers.ts @@ -11,6 +11,8 @@ export function createMockPluginRegistry( hookName: string; handler: (...args: unknown[]) => unknown; pluginId?: string; + priority?: number; + timeoutMs?: number; }>, ): PluginRegistry { const pluginIds = @@ -32,7 +34,8 @@ export function createMockPluginRegistry( pluginId: h.pluginId ?? "test-plugin", hookName: h.hookName, handler: h.handler, - priority: 0, + priority: h.priority ?? 0, + ...(h.timeoutMs !== undefined ? { timeoutMs: h.timeoutMs } : {}), source: "test", })) as PluginRegistry["typedHooks"], }; @@ -115,6 +118,8 @@ export function createHookRunnerWithRegistry( hookName: string; handler: (...args: unknown[]) => unknown; pluginId?: string; + priority?: number; + timeoutMs?: number; }>, options?: Parameters[1], ) { diff --git a/src/plugins/hooks.ts b/src/plugins/hooks.ts index 929cc9183b6b..ab3c1bced424 100644 --- a/src/plugins/hooks.ts +++ b/src/plugins/hooks.ts @@ -239,6 +239,10 @@ const DEFAULT_MODIFYING_HOOK_TIMEOUT_MS_BY_HOOK: Partial() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + async function expectMessageHookCall(params: { hookName: "message_sending" | "message_sent"; event: PluginHookMessageSendingEvent | PluginHookMessageSentEvent; @@ -63,6 +71,73 @@ describe("message_sending hook runner", () => { channelCtx: demoChannelCtx, }); }); + + it("fails open after the default per-handler timeout", async () => { + vi.useFakeTimers(); + try { + const logger = { warn: vi.fn(), error: vi.fn() }; + const firstStarted = createDeferred(); + const first = vi.fn(() => { + firstStarted.resolve(); + return new Promise(() => {}); + }); + const second = vi.fn().mockResolvedValue({ content: "after timeout" }); + const { runner } = createHookRunnerWithRegistry( + [ + { hookName: "message_sending", handler: first }, + { hookName: "message_sending", handler: second }, + ], + { logger }, + ); + + const resultPromise = runner.runMessageSending( + { to: "user-123", content: "original content" }, + demoChannelCtx, + ); + await firstStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + + await expect(resultPromise).resolves.toEqual({ content: "after timeout" }); + expect(second).toHaveBeenCalledTimes(1); + expect(logger.error).toHaveBeenCalledWith( + "[hooks] message_sending handler from test-plugin failed: timed out after 15000ms", + ); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("preserves a handler-specific timeout longer than the default", async () => { + vi.useFakeTimers(); + try { + const handler = vi.fn( + () => + new Promise((resolve) => { + setTimeout(() => resolve({ content: "slow result" }), 16_000); + }), + ); + const { runner } = createHookRunnerWithRegistry([ + { hookName: "message_sending", handler, timeoutMs: 20_000 }, + ]); + const resultPromise = runner.runMessageSending( + { to: "user-123", content: "original content" }, + demoChannelCtx, + ); + let settled = false; + void resultPromise.then(() => { + settled = true; + }); + + await vi.advanceTimersByTimeAsync(15_000); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(1_000); + await expect(resultPromise).resolves.toEqual({ content: "slow result" }); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); }); describe("message_sent hook runner", () => { diff --git a/src/plugins/wired-hooks-reply-payload-sending.test.ts b/src/plugins/wired-hooks-reply-payload-sending.test.ts index 238f96be4480..cf9ee2676e72 100644 --- a/src/plugins/wired-hooks-reply-payload-sending.test.ts +++ b/src/plugins/wired-hooks-reply-payload-sending.test.ts @@ -8,6 +8,14 @@ import { import type { PluginHookReplyPayload } from "./hook-types.js"; import { createHookRunnerWithRegistry } from "./hooks.test-helpers.js"; +function createDeferred() { + let resolve!: (value: T | PromiseLike) => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + const replyPayloadSendingEvent = { payload: { text: "hello" } satisfies ReplyPayload, kind: "final" as const, @@ -29,6 +37,46 @@ function firstErrorLog(logger: { error: ReturnType }) { } describe("reply_payload_sending hook runner", () => { + it("fails open after the default per-handler timeout", async () => { + vi.useFakeTimers(); + try { + const logger = { warn: vi.fn(), error: vi.fn() }; + const firstStarted = createDeferred(); + const first = vi.fn(() => { + firstStarted.resolve(); + return new Promise(() => {}); + }); + const second = vi.fn().mockResolvedValue({ payload: { text: "after timeout" } }); + const { runner } = createHookRunnerWithRegistry( + [ + { hookName: "reply_payload_sending", handler: first }, + { hookName: "reply_payload_sending", handler: second }, + ], + { logger }, + ); + + const resultPromise = runner.runReplyPayloadSending( + replyPayloadSendingEvent, + replyPayloadSendingCtx, + ); + await firstStarted.promise; + await vi.advanceTimersByTimeAsync(15_000); + + await expect(resultPromise).resolves.toEqual({ + payload: { text: "after timeout" }, + cancel: undefined, + reason: undefined, + }); + expect(second).toHaveBeenCalledTimes(1); + expect(logger.error).toHaveBeenCalledWith( + "[hooks] reply_payload_sending handler from test-plugin failed: timed out after 15000ms", + ); + expect(vi.getTimerCount()).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + it("passes the latest payload between handlers", async () => { const first = vi.fn().mockResolvedValue({ payload: {