mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
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 <steipete@gmail.com>
This commit is contained in:
NianJiu
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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),
|
||||
}),
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -195,7 +195,7 @@ export function readPluginSdkSurfaceBudgets(env = process.env) {
|
||||
),
|
||||
publicExports: readPluginSdkSurfaceBudgetEnv(
|
||||
"OPENCLAW_PLUGIN_SDK_MAX_PUBLIC_EXPORTS",
|
||||
10553,
|
||||
10554,
|
||||
env,
|
||||
),
|
||||
publicFunctionExports: readPluginSdkSurfaceBudgetEnv(
|
||||
|
||||
@@ -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<object | void>;
|
||||
onBeforeDeliverCancelled?: (payload: ReplyPayload, info: { kind: Delivery["kind"] }) => void;
|
||||
onSettled?: () => object | void | Promise<object | void>;
|
||||
@@ -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<void>();
|
||||
const onSettled = vi.fn();
|
||||
let hookCalls = 0;
|
||||
const beforeDeliver = vi.fn((payload: ReplyPayload) => {
|
||||
hookCalls += 1;
|
||||
if (hookCalls === 1) {
|
||||
hookStarted.resolve();
|
||||
return new Promise<ReplyPayload>(() => {});
|
||||
}
|
||||
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<void>();
|
||||
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<void>();
|
||||
|
||||
+56
-59
@@ -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<ReplyPayload | null> => {
|
||||
if (!payload.text) {
|
||||
return markReplyDispatchBeforeDeliverDeadlineOwned(
|
||||
async (payload: ReplyPayload): Promise<ReplyPayload | null> => {
|
||||
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<ReplyPayload | null> => {
|
||||
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<ReplyPayload | null> => {
|
||||
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>
|
||||
): 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,
|
||||
|
||||
@@ -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. */
|
||||
|
||||
@@ -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";
|
||||
|
||||
@@ -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<typeof resolveFallbackTransition>;
|
||||
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(),
|
||||
|
||||
@@ -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<T>() {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
const promise = new Promise<T>((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<void>();
|
||||
const deliver = vi.fn().mockResolvedValue(undefined);
|
||||
const dispatcher = createReplyDispatcher({
|
||||
deliver,
|
||||
beforeDeliver: () => {
|
||||
hookStarted.resolve();
|
||||
return new Promise<never>(() => {});
|
||||
},
|
||||
});
|
||||
|
||||
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<void>();
|
||||
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<never>(() => {});
|
||||
}
|
||||
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<void>();
|
||||
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<never>(() => {});
|
||||
}
|
||||
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<void>();
|
||||
let hookCalls = 0;
|
||||
const dispatcher = createReplyDispatcher({
|
||||
deliver: vi.fn().mockResolvedValue(undefined),
|
||||
beforeDeliver: (payload) => {
|
||||
hookCalls += 1;
|
||||
if (hookCalls === 2) {
|
||||
hookStarted.resolve();
|
||||
return new Promise<never>(() => {});
|
||||
}
|
||||
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<void>();
|
||||
let hookCalls = 0;
|
||||
const dispatcher = createReplyDispatcher({
|
||||
deliver: vi.fn().mockResolvedValue(undefined),
|
||||
beforeDeliver: (payload) => {
|
||||
hookCalls += 1;
|
||||
if (hookCalls === 2) {
|
||||
hookStarted.resolve();
|
||||
return new Promise<never>(() => {});
|
||||
}
|
||||
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<void>();
|
||||
const dispatcher = createReplyDispatcher({
|
||||
deliver: vi.fn().mockResolvedValue(undefined),
|
||||
beforeDeliver: () => {
|
||||
hookStarted.resolve();
|
||||
return new Promise<never>(() => {});
|
||||
},
|
||||
});
|
||||
|
||||
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();
|
||||
|
||||
@@ -125,6 +125,22 @@ const sessionStoreMocks = vi.hoisted(() => ({
|
||||
return sessionStoreMocks.currentEntry;
|
||||
},
|
||||
),
|
||||
updateSessionEntry: vi.fn(
|
||||
async (
|
||||
_scope: unknown,
|
||||
update: (entry: Record<string, unknown>) => Promise<Record<string, unknown> | 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<typeof sessionStoreMocks.updateSessionEntry>[1],
|
||||
),
|
||||
};
|
||||
});
|
||||
vi.mock("./dispatch-from-config.runtime.js", () => ({
|
||||
|
||||
@@ -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<void> {
|
||||
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<void> {
|
||||
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<ReplyDispatchDeliveryOutcome>;
|
||||
}> => {
|
||||
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<ReplyDispatchDeliveryOutcome>;
|
||||
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();
|
||||
}
|
||||
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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> | void;
|
||||
|
||||
export type ReplyDispatchDeliveryOutcome =
|
||||
| "delivered"
|
||||
| "cancelled"
|
||||
| "failed-before-deliver"
|
||||
| "failed-deliver";
|
||||
|
||||
type ReplyDispatchDeliveryOutcomeTracker = {
|
||||
promise: Promise<ReplyDispatchDeliveryOutcome>;
|
||||
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<ReplyDispatcher, ReplyDispatchCancelHandler[]>();
|
||||
const deliveryOutcomeTrackers = new WeakMap<ReplyPayload, ReplyDispatchDeliveryOutcomeTracker>();
|
||||
|
||||
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<ReplyPayload | null> {
|
||||
const timeoutMs = stage.timeoutMs;
|
||||
if (!timeoutMs) {
|
||||
return await stage.hook(payload, info);
|
||||
}
|
||||
let timer: ReturnType<typeof setTimeout> | 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<never>((_, 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<ReplyDispatchDeliveryOutcome>;
|
||||
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<void> = 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 }),
|
||||
|
||||
@@ -20,11 +20,20 @@ export type ReplyDispatchBeforeDeliver = (
|
||||
info: ReplyDispatchRuntimeInfo,
|
||||
) => Promise<ReplyPayload | null> | 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<void>;
|
||||
getQueuedCounts: () => Record<ReplyDispatchKind, number>;
|
||||
getCancelledCounts?: () => Record<ReplyDispatchKind, number>;
|
||||
|
||||
@@ -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<Parameters<typeof createReplyDispatcher>[0]["deliver"]>[0];
|
||||
@@ -13,6 +17,14 @@ function deliveredText(deliver: DeliverMock, index = 0) {
|
||||
return payload?.text;
|
||||
}
|
||||
|
||||
function createDeferred<T>() {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
const promise = new Promise<T>((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<void>();
|
||||
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<never>(() => {});
|
||||
}
|
||||
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<void>();
|
||||
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<never>(() => {});
|
||||
}
|
||||
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<typeof createReplyDispatcher>[0]["deliver"] = async () =>
|
||||
await Promise.resolve();
|
||||
|
||||
@@ -46,6 +46,7 @@ export {
|
||||
createReplyDispatcherWithTyping,
|
||||
} from "../auto-reply/reply/reply-dispatcher.js";
|
||||
export type {
|
||||
ReplyDispatchBeforeDeliverOptions,
|
||||
ReplyDispatchKind,
|
||||
ReplyDispatcher,
|
||||
ReplyFollowupAdmissionBarrierTimeoutPolicy,
|
||||
|
||||
@@ -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<PrivateResumeOptionKeys>().toEqualTypeOf<never>();
|
||||
type ReplyRuntimeAppendBeforeDeliverOptions = Parameters<
|
||||
NonNullable<ReplyRuntimeDispatcher["appendBeforeDeliver"]>
|
||||
>[1];
|
||||
expectTypeOf<ReplyRuntimeAppendBeforeDeliverOptions>().toEqualTypeOf<
|
||||
ReplyRuntimeBeforeDeliverOptions | undefined
|
||||
>();
|
||||
});
|
||||
|
||||
it("keeps runtime entry subpaths importable", async () => {
|
||||
|
||||
@@ -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<typeof createHookRunner>[1],
|
||||
) {
|
||||
|
||||
@@ -239,6 +239,10 @@ const DEFAULT_MODIFYING_HOOK_TIMEOUT_MS_BY_HOOK: Partial<Record<PluginHookName,
|
||||
// unresolved; timeout fail-opens with the original final answer.
|
||||
before_agent_finalize: 15_000,
|
||||
before_prompt_build: 15_000,
|
||||
// Outbound modifying hooks run inside the serialized reply delivery lane.
|
||||
// A hung plugin must fail open so later hooks and queued replies can settle.
|
||||
message_sending: 15_000,
|
||||
reply_payload_sending: 15_000,
|
||||
resolve_exec_env: 15_000,
|
||||
};
|
||||
|
||||
|
||||
@@ -11,6 +11,14 @@ import type {
|
||||
PluginHookMessageSentEvent,
|
||||
} from "./types.js";
|
||||
|
||||
function createDeferred<T>() {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
const promise = new Promise<T>((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<void>();
|
||||
const first = vi.fn(() => {
|
||||
firstStarted.resolve();
|
||||
return new Promise<PluginHookMessageSendingResult>(() => {});
|
||||
});
|
||||
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<PluginHookMessageSendingResult>((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", () => {
|
||||
|
||||
@@ -8,6 +8,14 @@ import {
|
||||
import type { PluginHookReplyPayload } from "./hook-types.js";
|
||||
import { createHookRunnerWithRegistry } from "./hooks.test-helpers.js";
|
||||
|
||||
function createDeferred<T>() {
|
||||
let resolve!: (value: T | PromiseLike<T>) => void;
|
||||
const promise = new Promise<T>((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<typeof vi.fn> }) {
|
||||
}
|
||||
|
||||
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<void>();
|
||||
const first = vi.fn(() => {
|
||||
firstStarted.resolve();
|
||||
return new Promise<never>(() => {});
|
||||
});
|
||||
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: {
|
||||
|
||||
Reference in New Issue
Block a user