mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-23 19:08:22 -06:00
Preserve stale Telegram block drafts before rotation
Queued answer block rotations can split assistant messages while the previous Telegram preview is stale. Materialize the previous block through the existing delivery fallback before resetting the draft lane, and keep assistant-message correlation on the internal dispatcher path instead of expanding the public Plugin SDK payload API. Constraint: ClawSweeper flagged missing live Telegram proof and a public Plugin SDK helper surface; the code path must stay a Telegram/channel bugfix without adding a third-party SDK contract. Rejected: Export getReplyPayloadAssistantMessageIndex from openclaw/plugin-sdk/reply-payload | it exposes internal reply metadata solely for this Telegram fix. Rejected: Match queued block rotations only by text | plugin rewrites, repeated text, and media-only transformed block deliveries need assistant-message correlation. Confidence: high Scope-risk: narrow Directive: Keep intermediate block materialization non-durable unless it is the actual final answer path, and keep assistant-message correlation off public reply-payload SDK exports. Tested: OPENCLAW_HEAVY_CHECK_LOCK_SCOPE=worktree node scripts/run-vitest.mjs extensions/telegram/src/bot-message-dispatch.test.ts extensions/telegram/src/lane-delivery.test.ts src/auto-reply/reply/before-deliver.test.ts src/auto-reply/dispatch.test.ts src/auto-reply/reply/dispatch-from-config.test.ts src/plugins/wired-hooks-reply-payload-sending.test.ts Tested: node_modules/.bin/oxlint --tsconfig config/tsconfig/oxlint.extensions.json extensions/telegram/src/bot-message-dispatch.ts extensions/telegram/src/bot-message-dispatch.test.ts extensions/telegram/src/lane-delivery-text-deliverer.ts extensions/telegram/src/lane-delivery.test.ts Tested: node_modules/.bin/oxlint --tsconfig config/tsconfig/oxlint.core.json src/auto-reply/reply/reply-dispatcher.ts src/auto-reply/reply/before-deliver.test.ts src/auto-reply/dispatch.ts src/auto-reply/dispatch.test.ts src/auto-reply/reply/dispatch-from-config.ts src/auto-reply/reply/dispatch-from-config.test.ts src/auto-reply/reply/reply-payload-sending-hook.ts Tested: git diff --check origin/main Not-tested: Redacted live Telegram Bot API/Desktop proof; no Telegram credentials or chat target are configured in this local environment. Not-tested: tsgo extensions test command was attempted locally and terminated after running over six minutes without output; prior known local run failed on unrelated Discord voice libopus-wasm errors.
This commit is contained in:
Alexazhu
@@ -2129,6 +2129,111 @@ describe("dispatchTelegramMessage draft streaming", () => {
|
||||
expect(deliverReplies).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("falls back to normal delivery before rotating a stale queued block preview", async () => {
|
||||
const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
|
||||
let firstBlockPreviewWentStale = false;
|
||||
answerDraftStream.lastDeliveredText.mockImplementation(() =>
|
||||
firstBlockPreviewWentStale ? "stale draft still visible" : "",
|
||||
);
|
||||
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(
|
||||
async ({ dispatcherOptions, replyOptions }) => {
|
||||
const firstPayload = setReplyPayloadMetadata(
|
||||
{ text: "Site A shows X." },
|
||||
{ assistantMessageIndex: 0 },
|
||||
);
|
||||
const secondPayload = setReplyPayloadMetadata(
|
||||
{ text: "Site B shows Y." },
|
||||
{ assistantMessageIndex: 1 },
|
||||
);
|
||||
await replyOptions?.onBlockReplyQueued?.(firstPayload, { assistantMessageIndex: 0 });
|
||||
await dispatcherOptions.deliver(firstPayload, { kind: "block" });
|
||||
firstBlockPreviewWentStale = true;
|
||||
await replyOptions?.onBlockReplyQueued?.(secondPayload, { assistantMessageIndex: 1 });
|
||||
await dispatcherOptions.deliver(secondPayload, { kind: "block" });
|
||||
return { queuedFinal: true };
|
||||
},
|
||||
);
|
||||
|
||||
await dispatchWithContext({ context: createContext() });
|
||||
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "Site A shows X.");
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(2, "Site A shows X.");
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(3, "Site B shows Y.");
|
||||
expect(answerDraftStream.clear).toHaveBeenCalled();
|
||||
expect(deliverReplies).toHaveBeenCalledTimes(1);
|
||||
const fallbackDelivery = mockCallArg(deliverReplies) as {
|
||||
replies?: Array<{ text?: string }>;
|
||||
transcriptMirror?: unknown;
|
||||
};
|
||||
expect(fallbackDelivery.replies?.[0]?.text).toBe("Site A shows X.");
|
||||
expect(fallbackDelivery.transcriptMirror).toBeUndefined();
|
||||
const clearOrder = answerDraftStream.clear.mock.invocationCallOrder[0];
|
||||
const fallbackDeliveryOrder = deliverReplies.mock.invocationCallOrder[0];
|
||||
const rotationOrder = answerDraftStream.forceNewMessage.mock.invocationCallOrder[0];
|
||||
const secondBlockUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[2];
|
||||
expect(clearOrder).toBeLessThan(fallbackDeliveryOrder);
|
||||
expect(fallbackDeliveryOrder).toBeLessThan(rotationOrder);
|
||||
expect(rotationOrder).toBeLessThan(secondBlockUpdateOrder);
|
||||
});
|
||||
|
||||
it("keeps stale block materialization tied to the streamed block before later media sends", async () => {
|
||||
const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
|
||||
let firstBlockPreviewWentStale = false;
|
||||
answerDraftStream.lastDeliveredText.mockImplementation(() =>
|
||||
firstBlockPreviewWentStale ? "stale draft still visible" : "",
|
||||
);
|
||||
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(
|
||||
async ({ dispatcherOptions, replyOptions }) => {
|
||||
const firstPayload = setReplyPayloadMetadata(
|
||||
{ text: "Site A shows X." },
|
||||
{ assistantMessageIndex: 0 },
|
||||
);
|
||||
const mediaPayload = setReplyPayloadMetadata(
|
||||
{
|
||||
text: "Chart attached",
|
||||
mediaUrl: "https://example.test/chart.png",
|
||||
},
|
||||
{ assistantMessageIndex: 0 },
|
||||
);
|
||||
const nextPayload = setReplyPayloadMetadata(
|
||||
{ text: "Site B shows Y." },
|
||||
{ assistantMessageIndex: 1 },
|
||||
);
|
||||
await replyOptions?.onBlockReplyQueued?.(firstPayload, { assistantMessageIndex: 0 });
|
||||
await dispatcherOptions.deliver(firstPayload, { kind: "block" });
|
||||
await replyOptions?.onBlockReplyQueued?.(mediaPayload, { assistantMessageIndex: 0 });
|
||||
await dispatcherOptions.deliver(mediaPayload, { kind: "block" });
|
||||
firstBlockPreviewWentStale = true;
|
||||
await replyOptions?.onBlockReplyQueued?.(nextPayload, { assistantMessageIndex: 1 });
|
||||
await dispatcherOptions.deliver(nextPayload, { kind: "block" });
|
||||
return { queuedFinal: true };
|
||||
},
|
||||
);
|
||||
|
||||
await dispatchWithContext({ context: createContext() });
|
||||
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "Site A shows X.");
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(2, "Site A shows X.");
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(3, "Site B shows Y.");
|
||||
expect(deliverReplies).toHaveBeenCalledTimes(2);
|
||||
expectDeliveredReply(
|
||||
0,
|
||||
{ text: "Chart attached", mediaUrl: "https://example.test/chart.png" },
|
||||
0,
|
||||
);
|
||||
const fallbackDelivery = mockCallArg(deliverReplies, 1) as {
|
||||
replies?: Array<{ text?: string; mediaUrl?: string; mediaUrls?: string[] }>;
|
||||
};
|
||||
expect(fallbackDelivery.replies?.[0]).toEqual({ text: "Site A shows X." });
|
||||
const mediaDeliveryOrder = deliverReplies.mock.invocationCallOrder[0];
|
||||
const fallbackDeliveryOrder = deliverReplies.mock.invocationCallOrder[1];
|
||||
const rotationOrder = answerDraftStream.forceNewMessage.mock.invocationCallOrder[0];
|
||||
const nextBlockUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[2];
|
||||
expect(mediaDeliveryOrder).toBeLessThan(fallbackDeliveryOrder);
|
||||
expect(fallbackDeliveryOrder).toBeLessThan(rotationOrder);
|
||||
expect(rotationOrder).toBeLessThan(nextBlockUpdateOrder);
|
||||
});
|
||||
|
||||
it("rotates queued block boundaries before async block delivery drains", async () => {
|
||||
const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
|
||||
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(
|
||||
@@ -2158,6 +2263,16 @@ describe("dispatchTelegramMessage draft streaming", () => {
|
||||
const secondBlockUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[1];
|
||||
expect(rotationOrder).toBeLessThan(secondBlockUpdateOrder);
|
||||
expect(deliverReplies).not.toHaveBeenCalled();
|
||||
expectRecordFields(mockCallArg(emitInternalMessageSentHook), {
|
||||
content: "Site A shows X.",
|
||||
messageId: 2001,
|
||||
});
|
||||
expectRecordFields(mockCallArg(recordOutboundMessageForPromptContext), {
|
||||
chatId: "123",
|
||||
messageId: 2001,
|
||||
text: "Site A shows X.",
|
||||
messageThreadId: 777,
|
||||
});
|
||||
});
|
||||
|
||||
it("skips canceled queued block rotations when later delivery drains", async () => {
|
||||
@@ -2175,7 +2290,7 @@ describe("dispatchTelegramMessage draft streaming", () => {
|
||||
);
|
||||
await dispatcherOptions.deliver(
|
||||
setReplyPayloadMetadata({ text: "Repeated block." }, { assistantMessageIndex: 1 }),
|
||||
{ kind: "block" },
|
||||
{ kind: "block", assistantMessageIndex: 1 } as { kind: "block" },
|
||||
);
|
||||
return { queuedFinal: true };
|
||||
},
|
||||
@@ -2286,6 +2401,60 @@ describe("dispatchTelegramMessage draft streaming", () => {
|
||||
expect(deliverReplies).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("expires rewritten unindexed queued block rotations after cancellation", async () => {
|
||||
const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
|
||||
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(
|
||||
async ({ dispatcherOptions, replyOptions }) => {
|
||||
await replyOptions?.onPartialReply?.({ text: "Existing preview" });
|
||||
await replyOptions?.onBlockReplyQueued?.({ text: "Original block text" });
|
||||
await replyOptions?.onAssistantMessageStart?.();
|
||||
await dispatcherOptions.onBeforeDeliverCancelled?.(
|
||||
{ text: "PFX Original block text" },
|
||||
{ kind: "block" },
|
||||
);
|
||||
await replyOptions?.onPartialReply?.({ text: "Second preview" });
|
||||
return { queuedFinal: true };
|
||||
},
|
||||
);
|
||||
|
||||
await dispatchWithContext({ context: createContext() });
|
||||
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "Existing preview");
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(2, "Second preview");
|
||||
expect(answerDraftStream.forceNewMessage).toHaveBeenCalledTimes(1);
|
||||
const rotationOrder = answerDraftStream.forceNewMessage.mock.invocationCallOrder[0];
|
||||
const secondPreviewUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[1];
|
||||
expect(rotationOrder).toBeLessThan(secondPreviewUpdateOrder);
|
||||
expect(deliverReplies).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("expires rewritten unindexed queued block rotations after skip", async () => {
|
||||
const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
|
||||
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(
|
||||
async ({ dispatcherOptions, replyOptions }) => {
|
||||
await replyOptions?.onPartialReply?.({ text: "Existing preview" });
|
||||
await replyOptions?.onBlockReplyQueued?.({ text: "Original block text" });
|
||||
await replyOptions?.onAssistantMessageStart?.();
|
||||
dispatcherOptions.onSkip?.(
|
||||
{ text: "PFX Original block text" },
|
||||
{ kind: "block", reason: "silent" },
|
||||
);
|
||||
await replyOptions?.onPartialReply?.({ text: "Second preview" });
|
||||
return { queuedFinal: true };
|
||||
},
|
||||
);
|
||||
|
||||
await dispatchWithContext({ context: createContext() });
|
||||
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "Existing preview");
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(2, "Second preview");
|
||||
expect(answerDraftStream.forceNewMessage).toHaveBeenCalledTimes(1);
|
||||
const rotationOrder = answerDraftStream.forceNewMessage.mock.invocationCallOrder[0];
|
||||
const secondPreviewUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[1];
|
||||
expect(rotationOrder).toBeLessThan(secondPreviewUpdateOrder);
|
||||
expect(deliverReplies).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("expires canceled queued block rotations before later partial previews", async () => {
|
||||
const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
|
||||
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(
|
||||
@@ -2314,6 +2483,36 @@ describe("dispatchTelegramMessage draft streaming", () => {
|
||||
expect(deliverReplies).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("serializes canceled block cleanup behind queued assistant-boundary events", async () => {
|
||||
const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
|
||||
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(
|
||||
async ({ dispatcherOptions, replyOptions }) => {
|
||||
const payload = setReplyPayloadMetadata(
|
||||
{ text: "Site A final" },
|
||||
{ assistantMessageIndex: 0 },
|
||||
);
|
||||
void replyOptions?.onPartialReply?.({ text: "Site A partial" });
|
||||
void replyOptions?.onBlockReplyQueued?.(payload, { assistantMessageIndex: 0 });
|
||||
void replyOptions?.onAssistantMessageStart?.();
|
||||
await dispatcherOptions.onBeforeDeliverCancelled?.(payload, { kind: "block" });
|
||||
await replyOptions?.onPartialReply?.({ text: "Site B partial" });
|
||||
return { queuedFinal: true };
|
||||
},
|
||||
);
|
||||
|
||||
await dispatchWithContext({ context: createContext() });
|
||||
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(1, "Site A partial");
|
||||
expect(answerDraftStream.update).toHaveBeenNthCalledWith(2, "Site B partial");
|
||||
expect(answerDraftStream.forceNewMessage).toHaveBeenCalledTimes(1);
|
||||
const firstPartialUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[0];
|
||||
const rotationOrder = answerDraftStream.forceNewMessage.mock.invocationCallOrder[0];
|
||||
const secondPartialUpdateOrder = answerDraftStream.update.mock.invocationCallOrder[1];
|
||||
expect(firstPartialUpdateOrder).toBeLessThan(rotationOrder);
|
||||
expect(rotationOrder).toBeLessThan(secondPartialUpdateOrder);
|
||||
expect(deliverReplies).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("preserves boundary rotation after a queued prior block is canceled", async () => {
|
||||
const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
|
||||
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(
|
||||
@@ -2541,7 +2740,7 @@ describe("dispatchTelegramMessage draft streaming", () => {
|
||||
{ mediaUrls: ["https://example.test/site-a.png"] },
|
||||
{ assistantMessageIndex: 0 },
|
||||
),
|
||||
{ kind: "block" },
|
||||
{ kind: "block", assistantMessageIndex: 0 } as { kind: "block" },
|
||||
);
|
||||
await replyOptions?.onPartialReply?.({ text: "Site B partial" });
|
||||
return { queuedFinal: true };
|
||||
|
||||
@@ -19,8 +19,8 @@ import {
|
||||
import { CURRENT_MESSAGE_MARKER } from "openclaw/plugin-sdk/channel-mention-gating";
|
||||
import {
|
||||
createChannelMessageReplyPipeline,
|
||||
createOutboundPayloadPlan,
|
||||
createPreviewMessageReceipt,
|
||||
createOutboundPayloadPlan,
|
||||
deriveDurableFinalDeliveryRequirements,
|
||||
projectOutboundPayloadPlanForDelivery,
|
||||
} from "openclaw/plugin-sdk/channel-outbound";
|
||||
@@ -969,6 +969,10 @@ export const dispatchTelegramMessage = async ({
|
||||
let lastAnswerPartialText = "";
|
||||
let activeAnswerDraftIsToolProgressOnly = false;
|
||||
let activeAnswerBlockAssistantMessageIndex: number | undefined;
|
||||
let lastAnswerBlockPayload: ReplyPayload | undefined;
|
||||
let lastAnswerBlockText: string | undefined;
|
||||
let lastAnswerBlockButtons: TelegramInlineButtons | undefined;
|
||||
let materializeAnswerLaneBeforeRotation: (() => Promise<boolean>) | undefined;
|
||||
type QueuedAnswerBlockRotation = {
|
||||
assistantMessageIndex?: number;
|
||||
text?: string;
|
||||
@@ -990,7 +994,7 @@ export const dispatchTelegramMessage = async ({
|
||||
return;
|
||||
}
|
||||
if (answerLane.hasStreamedMessage) {
|
||||
await rotateLaneForNewMessage(answerLane);
|
||||
await rotateAnswerLaneForNewMessage();
|
||||
}
|
||||
activeAnswerDraftIsToolProgressOnly = true;
|
||||
}
|
||||
@@ -1117,6 +1121,9 @@ export const dispatchTelegramMessage = async ({
|
||||
if (lane === answerLane) {
|
||||
resetAnswerToolProgressDraft();
|
||||
pendingAnswerBlockAssistantMessageIndex = undefined;
|
||||
lastAnswerBlockPayload = undefined;
|
||||
lastAnswerBlockText = undefined;
|
||||
lastAnswerBlockButtons = undefined;
|
||||
}
|
||||
};
|
||||
const rotateLaneForNewMessage = async (lane: DraftLaneState) => {
|
||||
@@ -1128,6 +1135,12 @@ export const dispatchTelegramMessage = async ({
|
||||
lane.stream?.forceNewMessage();
|
||||
resetDraftLaneState(lane);
|
||||
};
|
||||
const rotateAnswerLaneForNewMessage = async () => {
|
||||
if (materializeAnswerLaneBeforeRotation) {
|
||||
await materializeAnswerLaneBeforeRotation();
|
||||
}
|
||||
await rotateLaneForNewMessage(answerLane);
|
||||
};
|
||||
const rotateAnswerLaneAfterToolProgress = async () => {
|
||||
nativeToolProgressDraft?.stop();
|
||||
if (!activeAnswerDraftIsToolProgressOnly) {
|
||||
@@ -1148,7 +1161,7 @@ export const dispatchTelegramMessage = async ({
|
||||
if (!answerLane.hasStreamedMessage || activeAnswerDraftIsToolProgressOnly) {
|
||||
return false;
|
||||
}
|
||||
await rotateLaneForNewMessage(answerLane);
|
||||
await rotateAnswerLaneForNewMessage();
|
||||
return true;
|
||||
};
|
||||
const prepareAnswerLaneForText = async (): Promise<boolean> => {
|
||||
@@ -1216,23 +1229,34 @@ export const dispatchTelegramMessage = async ({
|
||||
queuedAnswerBlockAssistantMessageIndex = entry.assistantMessageIndex;
|
||||
}
|
||||
};
|
||||
const queuedAnswerBlockRotationMatchesPayload = (
|
||||
const queuedAnswerBlockRotationTextMatchesPayload = (
|
||||
entry: QueuedAnswerBlockRotation,
|
||||
payload: ReplyPayload,
|
||||
) => {
|
||||
return (
|
||||
entry.assistantMessageIndex === undefined ||
|
||||
(entry.text !== undefined && payload.text !== undefined && entry.text === payload.text)
|
||||
);
|
||||
return entry.text !== undefined && payload.text !== undefined && entry.text === payload.text;
|
||||
};
|
||||
const takeQueuedAnswerBlockRotation = (payload: ReplyPayload): boolean => {
|
||||
const matchIndex = queuedAnswerBlockRotations.findIndex((entry) =>
|
||||
queuedAnswerBlockRotationMatchesPayload(entry, payload),
|
||||
);
|
||||
if (matchIndex < 0) {
|
||||
const queuedAnswerBlockRotationMatchesDelivery = (
|
||||
entry: QueuedAnswerBlockRotation,
|
||||
payload: ReplyPayload,
|
||||
assistantMessageIndex?: number,
|
||||
) => {
|
||||
if (assistantMessageIndex !== undefined && entry.assistantMessageIndex !== undefined) {
|
||||
return assistantMessageIndex === entry.assistantMessageIndex;
|
||||
}
|
||||
return queuedAnswerBlockRotationTextMatchesPayload(entry, payload);
|
||||
};
|
||||
const takeQueuedAnswerBlockRotation = (
|
||||
payload: ReplyPayload,
|
||||
assistantMessageIndex?: number,
|
||||
): boolean => {
|
||||
if (queuedAnswerBlockRotations.length === 0) {
|
||||
return false;
|
||||
}
|
||||
const matchedEntries = queuedAnswerBlockRotations.splice(0, matchIndex + 1);
|
||||
const matchIndex = queuedAnswerBlockRotations.findIndex((entry) =>
|
||||
queuedAnswerBlockRotationMatchesDelivery(entry, payload, assistantMessageIndex),
|
||||
);
|
||||
const consumeIndex = Math.max(matchIndex, 0);
|
||||
const matchedEntries = queuedAnswerBlockRotations.splice(0, consumeIndex + 1);
|
||||
const matchedEntry = matchedEntries.at(-1);
|
||||
const shouldRotateBeforeDelivery = matchedEntry?.shouldRotateBeforeDelivery ?? false;
|
||||
if (matchedEntry?.assistantMessageIndex !== undefined) {
|
||||
@@ -1242,10 +1266,18 @@ export const dispatchTelegramMessage = async ({
|
||||
recomputeQueuedAnswerBlockRotations();
|
||||
return shouldRotateBeforeDelivery;
|
||||
};
|
||||
const dropQueuedAnswerBlockRotation = (payload: ReplyPayload) => {
|
||||
const matchIndex = queuedAnswerBlockRotations.findIndex((entry) =>
|
||||
queuedAnswerBlockRotationMatchesPayload(entry, payload),
|
||||
const dropQueuedAnswerBlockRotation = (
|
||||
payload: ReplyPayload,
|
||||
assistantMessageIndex?: number,
|
||||
) => {
|
||||
let matchIndex = queuedAnswerBlockRotations.findIndex((entry) =>
|
||||
queuedAnswerBlockRotationMatchesDelivery(entry, payload, assistantMessageIndex),
|
||||
);
|
||||
if (matchIndex < 0 && assistantMessageIndex === undefined) {
|
||||
matchIndex = queuedAnswerBlockRotations.findIndex(
|
||||
(entry) => entry.assistantMessageIndex === undefined,
|
||||
);
|
||||
}
|
||||
if (matchIndex >= 0) {
|
||||
const matchedEntry = queuedAnswerBlockRotations[matchIndex];
|
||||
queuedAnswerBlockRotations.splice(matchIndex, 1);
|
||||
@@ -1261,6 +1293,10 @@ export const dispatchTelegramMessage = async ({
|
||||
recomputeQueuedAnswerBlockRotations();
|
||||
}
|
||||
};
|
||||
const getReplyDispatchAssistantMessageIndex = (info: object): number | undefined => {
|
||||
const value = (info as { assistantMessageIndex?: unknown }).assistantMessageIndex;
|
||||
return typeof value === "number" ? value : undefined;
|
||||
};
|
||||
const updateDraftFromPartial = (lane: DraftLaneState, update: DraftPartialTextUpdate) => {
|
||||
const laneStream = lane.stream;
|
||||
if (!laneStream || !update.text) {
|
||||
@@ -1713,6 +1749,60 @@ export const dispatchTelegramMessage = async ({
|
||||
deliveryState.markDelivered();
|
||||
},
|
||||
});
|
||||
materializeAnswerLaneBeforeRotation = async () => {
|
||||
if (
|
||||
!lastAnswerBlockPayload ||
|
||||
!answerLane.stream ||
|
||||
!answerLane.hasStreamedMessage ||
|
||||
answerLane.finalized ||
|
||||
activeAnswerDraftIsToolProgressOnly
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
const text = answerLane.lastPartialText || lastAnswerPartialText || lastAnswerBlockText;
|
||||
if (!text?.trim()) {
|
||||
return false;
|
||||
}
|
||||
// A block skipped by the duplicate-draft dedup was never rendered to its
|
||||
// own draft update. Force the full delivery path (not the no-op finalize
|
||||
// fast path) so the preserved intermediate block is materialized as a
|
||||
// visible draft before the lane rotates for the next message.
|
||||
const wasSkippedDuplicate = skippedDuplicateAnswerBlockDraftDelivery;
|
||||
skippedDuplicateAnswerBlockDraftDelivery = false;
|
||||
const deliveredText = answerLane.stream.lastDeliveredText?.();
|
||||
const messageId = answerLane.stream.messageId();
|
||||
if (
|
||||
!lastAnswerBlockButtons &&
|
||||
!wasSkippedDuplicate &&
|
||||
deliveredText === text.trimEnd() &&
|
||||
typeof messageId === "number"
|
||||
) {
|
||||
await answerLane.stream.stop();
|
||||
answerLane.finalized = true;
|
||||
deliveryState.markDelivered();
|
||||
await emitPreviewFinalizedHook({
|
||||
kind: "preview-finalized",
|
||||
delivery: {
|
||||
content: text,
|
||||
promptContextContent: deliveredText,
|
||||
messageId,
|
||||
receipt: createPreviewMessageReceipt({ id: messageId }),
|
||||
},
|
||||
});
|
||||
return true;
|
||||
}
|
||||
const result = await deliverLaneText({
|
||||
laneName: "answer",
|
||||
text,
|
||||
payload: lastAnswerBlockPayload,
|
||||
infoKind: "block",
|
||||
buttons: lastAnswerBlockButtons,
|
||||
finalizePreview: true,
|
||||
durable: false,
|
||||
});
|
||||
await emitPreviewFinalizedHook(result);
|
||||
return result.kind !== "skipped";
|
||||
};
|
||||
const deliverProgressModeFinalAnswer = async (
|
||||
payload: ReplyPayload,
|
||||
text: string,
|
||||
@@ -1810,8 +1900,14 @@ export const dispatchTelegramMessage = async ({
|
||||
beforeDeliver: async (payload) => payload,
|
||||
onBeforeDeliverCancelled: (payload, info) => {
|
||||
if (info.kind === "block") {
|
||||
dropQueuedAnswerBlockRotation(payload);
|
||||
return enqueueDraftLaneEvent(async () => {
|
||||
dropQueuedAnswerBlockRotation(
|
||||
payload,
|
||||
getReplyDispatchAssistantMessageIndex(info),
|
||||
);
|
||||
});
|
||||
}
|
||||
return undefined;
|
||||
},
|
||||
deliver: async (payload, info) => {
|
||||
if (isDispatchSuperseded()) {
|
||||
@@ -1925,7 +2021,10 @@ export const dispatchTelegramMessage = async ({
|
||||
let blockDelivered = false;
|
||||
const hasAnswerSegment = segments.some((segment) => segment.lane === "answer");
|
||||
if (info.kind === "block" && !hasAnswerSegment) {
|
||||
dropQueuedAnswerBlockRotation(effectivePayload);
|
||||
dropQueuedAnswerBlockRotation(
|
||||
effectivePayload,
|
||||
getReplyDispatchAssistantMessageIndex(info),
|
||||
);
|
||||
}
|
||||
for (const segment of segments) {
|
||||
if (
|
||||
@@ -1969,6 +2068,15 @@ export const dispatchTelegramMessage = async ({
|
||||
await prepareAnswerLaneForToolProgress();
|
||||
}
|
||||
|
||||
const ownedByQueuedAnswerBlockRotation =
|
||||
queuedAnswerBlockRotations.some((entry) =>
|
||||
queuedAnswerBlockRotationMatchesDelivery(
|
||||
entry,
|
||||
effectivePayload,
|
||||
getReplyDispatchAssistantMessageIndex(info),
|
||||
),
|
||||
);
|
||||
|
||||
const skipTextOnlyBlock =
|
||||
streamMode === "partial" &&
|
||||
info.kind === "block" &&
|
||||
@@ -1978,20 +2086,35 @@ export const dispatchTelegramMessage = async ({
|
||||
telegramButtons === undefined &&
|
||||
answerLane.hasStreamedMessage &&
|
||||
!activeAnswerDraftIsToolProgressOnly &&
|
||||
!ownedByQueuedAnswerBlockRotation &&
|
||||
segment.update.text.trimEnd() === answerLane.lastPartialText.trimEnd();
|
||||
|
||||
if (skipTextOnlyBlock) {
|
||||
// Defer the duplicate block: do not emit a redundant draft
|
||||
// update now. Record it so that if a later rotation (tool
|
||||
// progress / next assistant message) follows, the skipped
|
||||
// block is materialized first instead of being lost, and so
|
||||
// that the dispatch-end finalize can commit it when nothing
|
||||
// else follows. Re-enable progress-draft state so a
|
||||
// following tool-progress step can still rotate the lane.
|
||||
skippedDuplicateAnswerBlockDraftDelivery = true;
|
||||
lastAnswerBlockPayload = effectivePayload;
|
||||
lastAnswerBlockText = segment.update.text;
|
||||
lastAnswerBlockButtons = telegramButtons;
|
||||
resetAnswerToolProgressDraft();
|
||||
resetProgressDraftState();
|
||||
blockDelivered = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (segment.lane === "answer" && info.kind === "block") {
|
||||
const preparedAnswerLane = await prepareAnswerLaneForText();
|
||||
const shouldRotateQueuedBlock =
|
||||
takeQueuedAnswerBlockRotation(effectivePayload);
|
||||
const shouldRotateQueuedBlock = takeQueuedAnswerBlockRotation(
|
||||
effectivePayload,
|
||||
getReplyDispatchAssistantMessageIndex(info),
|
||||
);
|
||||
if (shouldRotateQueuedBlock && !preparedAnswerLane) {
|
||||
await rotateLaneForNewMessage(answerLane);
|
||||
await rotateAnswerLaneForNewMessage();
|
||||
rotateAnswerLaneWhenQueuedBlocksSettle = false;
|
||||
}
|
||||
resetAnswerToolProgressDraft();
|
||||
@@ -2014,6 +2137,17 @@ export const dispatchTelegramMessage = async ({
|
||||
if (segment.lane === "answer" && result.kind === "preview-finalized") {
|
||||
await emitPreviewFinalizedHook(result);
|
||||
}
|
||||
if (
|
||||
segment.lane === "answer" &&
|
||||
info.kind === "block" &&
|
||||
(result.kind === "preview-updated" ||
|
||||
result.kind === "preview-finalized" ||
|
||||
result.kind === "preview-retained")
|
||||
) {
|
||||
lastAnswerBlockPayload = effectivePayload;
|
||||
lastAnswerBlockText = segment.update.text;
|
||||
lastAnswerBlockButtons = telegramButtons;
|
||||
}
|
||||
blockDelivered = blockDelivered || result.kind !== "skipped";
|
||||
if (segment.lane === "reasoning") {
|
||||
if (result.kind !== "skipped") {
|
||||
@@ -2086,7 +2220,10 @@ export const dispatchTelegramMessage = async ({
|
||||
onSkip: (payload, info) => {
|
||||
if (info.kind === "block") {
|
||||
void enqueueDraftLaneEvent(async () => {
|
||||
dropQueuedAnswerBlockRotation(payload);
|
||||
dropQueuedAnswerBlockRotation(
|
||||
payload,
|
||||
getReplyDispatchAssistantMessageIndex(info),
|
||||
);
|
||||
});
|
||||
}
|
||||
if (payload.isError === true) {
|
||||
|
||||
@@ -78,6 +78,8 @@ type DeliverLaneTextParams = {
|
||||
payload: ReplyPayload;
|
||||
infoKind: string;
|
||||
buttons?: TelegramInlineButtons;
|
||||
finalizePreview?: boolean;
|
||||
durable?: boolean;
|
||||
};
|
||||
|
||||
function result(
|
||||
@@ -286,7 +288,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
|
||||
lane: DraftLaneState,
|
||||
text: string,
|
||||
payload: ReplyPayload,
|
||||
isFinal: boolean,
|
||||
useFinalTextRecovery: boolean,
|
||||
finalizePreview: boolean,
|
||||
buttons?: TelegramInlineButtons,
|
||||
): Promise<LaneDeliveryResult | undefined> => {
|
||||
@@ -315,7 +317,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
|
||||
const finalText = activeFullText.trimEnd();
|
||||
const deliveredStreamTextBeforeUpdate = stream.lastDeliveredText?.();
|
||||
const deliveredPrefixBeforeUpdate =
|
||||
isFinal &&
|
||||
useFinalTextRecovery &&
|
||||
deliveredStreamTextBeforeUpdate !== undefined &&
|
||||
isDeliveredPrefix({
|
||||
deliveredText: deliveredStreamTextBeforeUpdate,
|
||||
@@ -362,7 +364,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
|
||||
};
|
||||
|
||||
const candidateTexts = [stream.lastDeliveredText?.(), lane.lastPartialText];
|
||||
if (isFinal && remainingChunks.length === 0 && isPotentialTruncatedFinal(activeFullText)) {
|
||||
if (useFinalTextRecovery && remainingChunks.length === 0 && isPotentialTruncatedFinal(activeFullText)) {
|
||||
const resolvedFullCandidate = await params.resolveFinalTextCandidate?.({
|
||||
finalText: text,
|
||||
laneName,
|
||||
@@ -377,7 +379,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
|
||||
}
|
||||
|
||||
const retainedPreview =
|
||||
isFinal && remainingChunks.length === 0 && isPotentialTruncatedFinal(activeFullText)
|
||||
useFinalTextRecovery && remainingChunks.length === 0 && isPotentialTruncatedFinal(activeFullText)
|
||||
? selectLongerFinalText({
|
||||
finalText: activeFullText,
|
||||
candidateTexts,
|
||||
@@ -441,7 +443,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
|
||||
} else {
|
||||
await params.flushDraftLane(lane);
|
||||
}
|
||||
const activeChunkIndexAfterStop = isFinal ? clampActiveChunkIndex() : activeChunkIndex;
|
||||
const activeChunkIndexAfterStop = useFinalTextRecovery ? clampActiveChunkIndex() : activeChunkIndex;
|
||||
const activeChunkAfterStop = chunks[activeChunkIndexAfterStop] ?? activeChunk;
|
||||
const remainingChunksAfterStop = chunks.slice(activeChunkIndexAfterStop + 1);
|
||||
|
||||
@@ -470,6 +472,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
|
||||
!retainedActiveChunkAfterStop
|
||||
) {
|
||||
if (
|
||||
useFinalTextRecovery &&
|
||||
isDeliveredPrefix({ deliveredText: deliveredStreamTextAfterStop, finalText }) &&
|
||||
deliveredStreamTextAfterStop.length > activeChunkTextAfterStop.length
|
||||
) {
|
||||
@@ -523,20 +526,23 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
|
||||
payload,
|
||||
infoKind,
|
||||
buttons,
|
||||
finalizePreview: requestedFinalizePreview,
|
||||
durable: requestedDurable,
|
||||
}: DeliverLaneTextParams): Promise<LaneDeliveryResult> => {
|
||||
const lane = params.lanes[laneName];
|
||||
const reply = resolveSendableOutboundReplyParts(payload, { text });
|
||||
const isFinal = infoKind === "final";
|
||||
const finalizePreview = isFinal;
|
||||
const isDurableFinal = infoKind === "final";
|
||||
const finalizePreview = requestedFinalizePreview ?? isDurableFinal;
|
||||
const durable = requestedDurable ?? isDurableFinal;
|
||||
const streamed = !reply.hasMedia
|
||||
? await streamText(laneName, lane, text, payload, isFinal, finalizePreview, buttons)
|
||||
? await streamText(laneName, lane, text, payload, isDurableFinal, finalizePreview, buttons)
|
||||
: undefined;
|
||||
if (streamed) {
|
||||
return streamed;
|
||||
}
|
||||
|
||||
if (
|
||||
isFinal &&
|
||||
finalizePreview &&
|
||||
reply.hasMedia &&
|
||||
lane.stream &&
|
||||
lane.hasStreamedMessage &&
|
||||
@@ -548,7 +554,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
|
||||
lane,
|
||||
text,
|
||||
textOnlyPayload(payload),
|
||||
true,
|
||||
isDurableFinal,
|
||||
true,
|
||||
buttons,
|
||||
);
|
||||
@@ -564,7 +570,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
|
||||
fallbackButtons: stripButtons ? undefined : buttons,
|
||||
}),
|
||||
{
|
||||
durable: true,
|
||||
durable,
|
||||
},
|
||||
);
|
||||
return finalizedPreview;
|
||||
@@ -576,7 +582,7 @@ export function createLaneTextDeliverer(params: CreateLaneTextDelivererParams) {
|
||||
}
|
||||
|
||||
const delivered = await params.sendPayload(params.applyTextToPayload(payload, text), {
|
||||
durable: isFinal,
|
||||
durable,
|
||||
});
|
||||
if (delivered && finalizePreview) {
|
||||
lane.finalized = true;
|
||||
|
||||
@@ -179,6 +179,65 @@ describe("createLaneTextDeliverer", () => {
|
||||
expect(harness.lanes.answer.finalized).toBe(true);
|
||||
});
|
||||
|
||||
it("keeps media fallback non-durable when materializing an intermediate preview", async () => {
|
||||
const harness = createHarness({ answerMessageId: 999 });
|
||||
harness.lanes.answer.hasStreamedMessage = true;
|
||||
|
||||
const result = await harness.deliverLaneText({
|
||||
laneName: "answer",
|
||||
text: "visible block",
|
||||
payload: { text: "visible block", mediaUrls: ["file:///site-a.png"] },
|
||||
infoKind: "block",
|
||||
finalizePreview: true,
|
||||
durable: false,
|
||||
});
|
||||
|
||||
const delivery = expectPreviewFinalized(result);
|
||||
expect(delivery.content).toBe("visible block");
|
||||
expect(harness.sendPayload).toHaveBeenCalledWith(
|
||||
{ mediaUrls: ["file:///site-a.png"] },
|
||||
{ durable: false },
|
||||
);
|
||||
expect(harness.lanes.answer.finalized).toBe(true);
|
||||
});
|
||||
|
||||
it("does not use final transcript recovery when materializing an intermediate block preview", async () => {
|
||||
const previousBlock =
|
||||
"Here is the complete block preview with enough stable prefix text before the ellipsis...";
|
||||
const nextAssistantBlock =
|
||||
"Here is the complete block preview with enough stable prefix text before the ellipsis and later assistant continuation text.";
|
||||
const answer = createTestDraftStream({ messageId: 999 });
|
||||
answer.lastDeliveredText.mockReturnValue(nextAssistantBlock);
|
||||
const harness = createHarness({
|
||||
answerStream: answer,
|
||||
resolveFinalTextCandidate: () => nextAssistantBlock,
|
||||
});
|
||||
harness.lanes.answer.lastPartialText = previousBlock;
|
||||
harness.lanes.answer.hasStreamedMessage = true;
|
||||
|
||||
const result = await harness.deliverLaneText({
|
||||
laneName: "answer",
|
||||
text: previousBlock,
|
||||
payload: { text: previousBlock },
|
||||
infoKind: "block",
|
||||
finalizePreview: true,
|
||||
durable: false,
|
||||
});
|
||||
|
||||
expect(result.kind).toBe("sent");
|
||||
expect(answer.update).toHaveBeenCalledWith(previousBlock);
|
||||
expect(answer.update).not.toHaveBeenCalledWith(nextAssistantBlock);
|
||||
expect(harness.clearDraftLane).toHaveBeenCalledTimes(1);
|
||||
expect(harness.sendPayload).toHaveBeenCalledWith(
|
||||
{ text: previousBlock },
|
||||
{ durable: false },
|
||||
);
|
||||
expect(harness.sendPayload).not.toHaveBeenCalledWith(
|
||||
{ text: nextAssistantBlock },
|
||||
expect.anything(),
|
||||
);
|
||||
});
|
||||
|
||||
it("keeps block delivery in the draft lane when delivered text is stale", async () => {
|
||||
const answer = createTestDraftStream({ messageId: 999 });
|
||||
answer.lastDeliveredText.mockReturnValue("working");
|
||||
|
||||
@@ -60,6 +60,62 @@ describe("beforeDeliver in reply dispatcher", () => {
|
||||
expect(dispatcher.getCancelledCounts?.()).toEqual({ tool: 0, block: 0, final: 1 });
|
||||
});
|
||||
|
||||
it("notifies cancellation when beforeDeliver throws before delivery", async () => {
|
||||
const delivered: string[] = [];
|
||||
const cancelled: Array<{
|
||||
assistantMessageIndex?: number;
|
||||
kind: string;
|
||||
text: string;
|
||||
}> = [];
|
||||
const errors: Array<{
|
||||
assistantMessageIndex?: number;
|
||||
kind: string;
|
||||
message: string;
|
||||
}> = [];
|
||||
|
||||
const dispatcher = createReplyDispatcher({
|
||||
deliver: async (payload) => {
|
||||
delivered.push(payload.text ?? "");
|
||||
},
|
||||
onBeforeDeliverCancelled: (payload, info) => {
|
||||
cancelled.push({
|
||||
assistantMessageIndex: (info as { assistantMessageIndex?: number })
|
||||
.assistantMessageIndex,
|
||||
kind: info.kind,
|
||||
text: payload.text ?? "",
|
||||
});
|
||||
},
|
||||
onError: (err, info) => {
|
||||
errors.push({
|
||||
assistantMessageIndex: (info as { assistantMessageIndex?: number })
|
||||
.assistantMessageIndex,
|
||||
kind: info.kind,
|
||||
message: err instanceof Error ? err.message : String(err),
|
||||
});
|
||||
},
|
||||
beforeDeliver: async () => {
|
||||
throw new Error("pre-delivery failed");
|
||||
},
|
||||
});
|
||||
|
||||
dispatcher.sendBlockReply(
|
||||
setReplyPayloadMetadata({ text: "blocked block" }, { assistantMessageIndex: 9 }),
|
||||
);
|
||||
dispatcher.markComplete();
|
||||
await dispatcher.waitForIdle();
|
||||
|
||||
expect(delivered).toEqual([]);
|
||||
expect(cancelled).toEqual([
|
||||
{ assistantMessageIndex: 9, kind: "block", text: "blocked block" },
|
||||
]);
|
||||
expect(errors).toEqual([
|
||||
{ assistantMessageIndex: 9, kind: "block", message: "pre-delivery failed" },
|
||||
]);
|
||||
expect(dispatcher.getQueuedCounts()).toEqual({ tool: 0, block: 1, final: 0 });
|
||||
expect(dispatcher.getCancelledCounts?.()).toEqual({ tool: 0, block: 0, final: 0 });
|
||||
expect(dispatcher.getFailedCounts?.()).toEqual({ tool: 0, block: 1, final: 0 });
|
||||
});
|
||||
|
||||
it("allows modifying payload in beforeDeliver", async () => {
|
||||
const delivered: string[] = [];
|
||||
|
||||
@@ -84,10 +140,14 @@ describe("beforeDeliver in reply dispatcher", () => {
|
||||
|
||||
it("preserves payload metadata through beforeDeliver rewrites", async () => {
|
||||
let deliveredMetadata: unknown;
|
||||
let deliveredAssistantMessageIndex: unknown;
|
||||
|
||||
const dispatcher = createReplyDispatcher({
|
||||
deliver: async (payload) => {
|
||||
deliver: async (payload, info) => {
|
||||
deliveredMetadata = getReplyPayloadMetadata(payload);
|
||||
deliveredAssistantMessageIndex = (
|
||||
info as { assistantMessageIndex?: unknown }
|
||||
).assistantMessageIndex;
|
||||
},
|
||||
beforeDeliver: async () => ({ text: "rewritten" }),
|
||||
});
|
||||
@@ -99,6 +159,7 @@ describe("beforeDeliver in reply dispatcher", () => {
|
||||
await dispatcher.waitForIdle();
|
||||
|
||||
expect(deliveredMetadata).toMatchObject({ assistantMessageIndex: 12 });
|
||||
expect(deliveredAssistantMessageIndex).toBe(12);
|
||||
});
|
||||
|
||||
it("delivers normally without beforeDeliver", async () => {
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
// Coalesces buffered block-streaming payloads into sendable reply parts.
|
||||
import { resolveSendableOutboundReplyParts } from "openclaw/plugin-sdk/reply-payload";
|
||||
import { isReplyPayloadStatusNotice } from "../reply-payload.js";
|
||||
import { copyReplyPayloadMetadata, isReplyPayloadStatusNotice } from "../reply-payload.js";
|
||||
import type { ReplyPayload } from "../types.js";
|
||||
import type { BlockStreamingCoalescing } from "./block-streaming.js";
|
||||
|
||||
@@ -32,6 +32,7 @@ export function createBlockReplyCoalescer(params: {
|
||||
let bufferIsCompactionNotice: ReplyPayload["isCompactionNotice"];
|
||||
let bufferIsFallbackNotice: ReplyPayload["isFallbackNotice"];
|
||||
let bufferIsStatusNotice: ReplyPayload["isStatusNotice"];
|
||||
let bufferMetadataSource: ReplyPayload | undefined;
|
||||
let idleTimer: NodeJS.Timeout | undefined;
|
||||
|
||||
const clearIdleTimer = () => {
|
||||
@@ -50,6 +51,17 @@ export function createBlockReplyCoalescer(params: {
|
||||
bufferIsCompactionNotice = undefined;
|
||||
bufferIsFallbackNotice = undefined;
|
||||
bufferIsStatusNotice = undefined;
|
||||
bufferMetadataSource = undefined;
|
||||
};
|
||||
|
||||
const startBufferFromPayload = (payload: ReplyPayload) => {
|
||||
bufferReplyToId = payload.replyToId;
|
||||
bufferAudioAsVoice = payload.audioAsVoice;
|
||||
bufferIsReasoning = payload.isReasoning;
|
||||
bufferIsCompactionNotice = payload.isCompactionNotice;
|
||||
bufferIsFallbackNotice = payload.isFallbackNotice;
|
||||
bufferIsStatusNotice = payload.isStatusNotice;
|
||||
bufferMetadataSource = payload;
|
||||
};
|
||||
|
||||
const scheduleIdleFlush = () => {
|
||||
@@ -84,8 +96,12 @@ export function createBlockReplyCoalescer(params: {
|
||||
isFallbackNotice: bufferIsFallbackNotice,
|
||||
isStatusNotice: bufferIsStatusNotice,
|
||||
};
|
||||
const metadataSource = bufferMetadataSource;
|
||||
const payloadWithMetadata = metadataSource
|
||||
? copyReplyPayloadMetadata(metadataSource, payload)
|
||||
: payload;
|
||||
resetBuffer();
|
||||
await onFlush(payload);
|
||||
await onFlush(payloadWithMetadata);
|
||||
};
|
||||
|
||||
const canMergeBufferedTextWithMedia = (payload: ReplyPayload) =>
|
||||
@@ -111,8 +127,11 @@ export function createBlockReplyCoalescer(params: {
|
||||
text: mergedText,
|
||||
replyToId: payload.replyToId ?? bufferReplyToId,
|
||||
};
|
||||
const metadataMergedPayload = bufferMetadataSource
|
||||
? copyReplyPayloadMetadata(bufferMetadataSource, mergedPayload)
|
||||
: mergedPayload;
|
||||
resetBuffer();
|
||||
return mergedPayload;
|
||||
return copyReplyPayloadMetadata(payload, metadataMergedPayload);
|
||||
};
|
||||
|
||||
const enqueue = (payload: ReplyPayload) => {
|
||||
@@ -142,12 +161,7 @@ export function createBlockReplyCoalescer(params: {
|
||||
if (bufferText) {
|
||||
void flush({ force: true });
|
||||
}
|
||||
bufferReplyToId = payload.replyToId;
|
||||
bufferAudioAsVoice = payload.audioAsVoice;
|
||||
bufferIsReasoning = payload.isReasoning;
|
||||
bufferIsCompactionNotice = payload.isCompactionNotice;
|
||||
bufferIsFallbackNotice = payload.isFallbackNotice;
|
||||
bufferIsStatusNotice = payload.isStatusNotice;
|
||||
startBufferFromPayload(payload);
|
||||
bufferText = text;
|
||||
void flush({ force: true });
|
||||
return;
|
||||
@@ -177,24 +191,14 @@ export function createBlockReplyCoalescer(params: {
|
||||
}
|
||||
|
||||
if (!bufferText) {
|
||||
bufferReplyToId = payload.replyToId;
|
||||
bufferAudioAsVoice = payload.audioAsVoice;
|
||||
bufferIsReasoning = payload.isReasoning;
|
||||
bufferIsCompactionNotice = payload.isCompactionNotice;
|
||||
bufferIsFallbackNotice = payload.isFallbackNotice;
|
||||
bufferIsStatusNotice = payload.isStatusNotice;
|
||||
startBufferFromPayload(payload);
|
||||
}
|
||||
|
||||
const nextText = bufferText ? `${bufferText}${joiner}${text}` : text;
|
||||
if (nextText.length > maxChars) {
|
||||
if (bufferText) {
|
||||
void flush({ force: true });
|
||||
bufferReplyToId = payload.replyToId;
|
||||
bufferAudioAsVoice = payload.audioAsVoice;
|
||||
bufferIsReasoning = payload.isReasoning;
|
||||
bufferIsCompactionNotice = payload.isCompactionNotice;
|
||||
bufferIsFallbackNotice = payload.isFallbackNotice;
|
||||
bufferIsStatusNotice = payload.isStatusNotice;
|
||||
startBufferFromPayload(payload);
|
||||
if (text.length >= maxChars) {
|
||||
void onFlush(payload);
|
||||
return;
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
/** Tests block reply pipeline buffering, dedupe, and final flush behavior. */
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { setReplyPayloadMetadata } from "../reply-payload.js";
|
||||
import { getReplyPayloadMetadata, setReplyPayloadMetadata } from "../reply-payload.js";
|
||||
import {
|
||||
createBlockReplyContentKey,
|
||||
createBlockReplyPayloadKey,
|
||||
@@ -289,6 +289,31 @@ describe("createBlockReplyPipeline dedup with threading", () => {
|
||||
|
||||
expect(sent).toEqual(["Alpha", "Beta"]);
|
||||
});
|
||||
|
||||
it("preserves assistant metadata on coalesced text flushes", async () => {
|
||||
const sent: Array<{ assistantMessageIndex?: number; text?: string }> = [];
|
||||
const pipeline = createBlockReplyPipeline({
|
||||
onBlockReply: async (payload) => {
|
||||
sent.push({
|
||||
assistantMessageIndex: getReplyPayloadMetadata(payload)?.assistantMessageIndex,
|
||||
text: payload.text,
|
||||
});
|
||||
},
|
||||
timeoutMs: 5000,
|
||||
coalescing: {
|
||||
minChars: 100,
|
||||
maxChars: 200,
|
||||
idleMs: 1000,
|
||||
joiner: " ",
|
||||
},
|
||||
});
|
||||
|
||||
pipeline.enqueue(setReplyPayloadMetadata({ text: "Alpha" }, { assistantMessageIndex: 0 }));
|
||||
pipeline.enqueue(setReplyPayloadMetadata({ text: "Beta" }, { assistantMessageIndex: 0 }));
|
||||
await pipeline.flush({ force: true });
|
||||
|
||||
expect(sent).toEqual([{ assistantMessageIndex: 0, text: "Alpha Beta" }]);
|
||||
});
|
||||
});
|
||||
|
||||
describe("createBlockReplyPipeline content coverage dedup", () => {
|
||||
|
||||
@@ -6,7 +6,7 @@ import { generateSecureInt } from "../../infra/secure-random.js";
|
||||
import { createSubsystemLogger } from "../../logging/subsystem.js";
|
||||
import type { SilentReplyConversationType } from "../../shared/silent-reply-policy.js";
|
||||
import { sleep } from "../../utils.js";
|
||||
import { copyReplyPayloadMetadata } from "../reply-payload.js";
|
||||
import { copyReplyPayloadMetadata, getReplyPayloadMetadata } from "../reply-payload.js";
|
||||
import { isSilentReplyText, SILENT_REPLY_TOKEN } from "../tokens.js";
|
||||
import type { GetReplyOptions, ReplyPayload } from "../types.js";
|
||||
import { registerDispatcher } from "./dispatcher-registry.js";
|
||||
@@ -47,6 +47,19 @@ const DEFAULT_HUMAN_DELAY_MIN_MS = 800;
|
||||
const DEFAULT_HUMAN_DELAY_MAX_MS = 2500;
|
||||
const silentReplyLogger = createSubsystemLogger("silent-reply/dispatcher");
|
||||
|
||||
type ReplyDispatchRuntimeInfo = { kind: ReplyDispatchKind; assistantMessageIndex?: number };
|
||||
|
||||
function buildReplyDispatchRuntimeInfo(
|
||||
payload: ReplyPayload,
|
||||
kind: ReplyDispatchKind,
|
||||
): ReplyDispatchRuntimeInfo {
|
||||
const assistantMessageIndex = getReplyPayloadMetadata(payload)?.assistantMessageIndex;
|
||||
return {
|
||||
kind,
|
||||
...(assistantMessageIndex !== undefined ? { assistantMessageIndex } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
/** Generate a random delay within the configured range. */
|
||||
function getHumanDelay(config: HumanDelayConfig | undefined): number {
|
||||
const mode = config?.mode ?? "off";
|
||||
@@ -175,7 +188,11 @@ export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDis
|
||||
responsePrefixContextProvider: options.responsePrefixContextProvider,
|
||||
transformReplyPayload: options.transformReplyPayload,
|
||||
onHeartbeatStrip: options.onHeartbeatStrip,
|
||||
onSkip: (reason) => options.onSkip?.(payload, { kind, reason }),
|
||||
onSkip: (reason) =>
|
||||
options.onSkip?.(payload, {
|
||||
...buildReplyDispatchRuntimeInfo(payload, kind),
|
||||
reason,
|
||||
}),
|
||||
});
|
||||
if (!normalized) {
|
||||
if (kind === "final" && originalWasExactSilent) {
|
||||
@@ -205,25 +222,35 @@ export function createReplyDispatcher(options: ReplyDispatcherOptions): ReplyDis
|
||||
await sleep(delayMs);
|
||||
}
|
||||
}
|
||||
const dispatchInfo = buildReplyDispatchRuntimeInfo(normalized, kind);
|
||||
let deliverPayload: ReplyPayload | null = normalized;
|
||||
if (beforeDeliver) {
|
||||
deliverPayload = await beforeDeliver(normalized, { kind });
|
||||
try {
|
||||
deliverPayload = await beforeDeliver(normalized, dispatchInfo);
|
||||
} catch (err: unknown) {
|
||||
try {
|
||||
await options.onBeforeDeliverCancelled?.(normalized, dispatchInfo);
|
||||
} catch (cancelErr: unknown) {
|
||||
void options.onError?.(cancelErr, dispatchInfo);
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
if (!deliverPayload) {
|
||||
cancelledCounts[kind] += 1;
|
||||
try {
|
||||
await options.onBeforeDeliverCancelled?.(normalized, { kind });
|
||||
await options.onBeforeDeliverCancelled?.(normalized, dispatchInfo);
|
||||
} catch (err: unknown) {
|
||||
void options.onError?.(err, { kind });
|
||||
void options.onError?.(err, dispatchInfo);
|
||||
}
|
||||
return;
|
||||
}
|
||||
deliverPayload = copyReplyPayloadMetadata(normalized, deliverPayload);
|
||||
}
|
||||
await options.deliver(deliverPayload, { kind });
|
||||
await options.deliver(deliverPayload, dispatchInfo);
|
||||
})
|
||||
.catch((err: unknown) => {
|
||||
failedCounts[kind] += 1;
|
||||
void options.onError?.(err, { kind });
|
||||
void options.onError?.(err, buildReplyDispatchRuntimeInfo(normalized, kind));
|
||||
})
|
||||
.finally(() => {
|
||||
pending -= 1;
|
||||
|
||||
Reference in New Issue
Block a user