Skip to content

Commit 9ef69ec

Browse files
fuller-stack-devchenyangjun-xy
fuller-stack-dev
authored and
chenyangjun-xy
committed
fix(telegram): deliver durable reasoning when enabled
Preserve shared reasoning suppression by default while letting Telegram opt into durable reasoning payloads only when it has a deliverable reasoning lane. Covers persistent /reasoning on, separate reasoning stream lanes, and progress-stream suppression.\n\nVerification:\n- node scripts/run-vitest.mjs src/auto-reply/reply/dispatch-from-config.test.ts extensions/telegram/src/bot-message-dispatch.test.ts\n- git diff --check upstream/main...HEAD\n- .agents/skills/autoreview/scripts/autoreview --mode branch --base upstream/main --stream-engine-output\n- CI run 28411526182 green, including QA Smoke CI and check-test-types\n- Real behavior proof run 28411676681 passed\n\nPR: openclaw#97875
1 parent 34464c4 commit 9ef69ec

5 files changed

Lines changed: 246 additions & 31 deletions

File tree

‎extensions/telegram/src/bot-message-dispatch.test.ts‎

Lines changed: 110 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3608,6 +3608,56 @@ describe("dispatchTelegramMessage draft streaming", () => {
36083608
await run;
36093609
});
36103610

3611+
it("keeps shared durable reasoning payloads disabled when reasoning is off", async () => {
3612+
dispatchReplyWithBufferedBlockDispatcher.mockResolvedValue({
3613+
queuedFinal: false,
3614+
counts: { block: 0, final: 0, tool: 0 },
3615+
});
3616+
3617+
await dispatchWithContext({ context: createContext() });
3618+
3619+
const dispatchParams = mockCallArg(dispatchReplyWithBufferedBlockDispatcher) as {
3620+
replyOptions?: { reasoningPayloadsEnabled?: boolean };
3621+
};
3622+
expect(dispatchParams.replyOptions?.reasoningPayloadsEnabled).toBe(false);
3623+
});
3624+
3625+
it("opts shared dispatch into durable reasoning payload delivery when reasoning streams", async () => {
3626+
setupDraftStreams({
3627+
answerMessageId: 2001,
3628+
reasoningMessageId: 3001,
3629+
});
3630+
dispatchReplyWithBufferedBlockDispatcher.mockResolvedValue({
3631+
queuedFinal: false,
3632+
counts: { block: 0, final: 0, tool: 0 },
3633+
});
3634+
3635+
await dispatchWithContext({ context: createReasoningStreamContext() });
3636+
3637+
const dispatchParams = mockCallArg(dispatchReplyWithBufferedBlockDispatcher) as {
3638+
replyOptions?: { reasoningPayloadsEnabled?: boolean };
3639+
};
3640+
expect(dispatchParams.replyOptions?.reasoningPayloadsEnabled).toBe(true);
3641+
});
3642+
3643+
it("keeps shared durable reasoning payloads disabled in progress stream mode", async () => {
3644+
setupDraftStreams({ answerMessageId: 2001 });
3645+
dispatchReplyWithBufferedBlockDispatcher.mockResolvedValue({
3646+
queuedFinal: false,
3647+
counts: { block: 0, final: 0, tool: 0 },
3648+
});
3649+
3650+
await dispatchWithContext({
3651+
context: createReasoningStreamContext(),
3652+
streamMode: "progress",
3653+
});
3654+
3655+
const dispatchParams = mockCallArg(dispatchReplyWithBufferedBlockDispatcher) as {
3656+
replyOptions?: { reasoningPayloadsEnabled?: boolean };
3657+
};
3658+
expect(dispatchParams.replyOptions?.reasoningPayloadsEnabled).toBe(false);
3659+
});
3660+
36113661
it("suppresses typed reasoning-only finals without raw text fallback", async () => {
36123662
setupDraftStreams({ answerMessageId: 2001, reasoningMessageId: 3001 });
36133663
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ dispatcherOptions }) => {
@@ -3624,6 +3674,66 @@ describe("dispatchTelegramMessage draft streaming", () => {
36243674
expect(editMessageTelegram).not.toHaveBeenCalled();
36253675
});
36263676

3677+
it("routes typed reasoning-only finals to the reasoning lane when reasoning streams", async () => {
3678+
const { reasoningDraftStream } = setupDraftStreams({
3679+
answerMessageId: 2001,
3680+
reasoningMessageId: 3001,
3681+
});
3682+
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ dispatcherOptions }) => {
3683+
await dispatcherOptions.deliver(
3684+
{ text: "<think>hidden</think>", isReasoning: true },
3685+
{ kind: "final" },
3686+
);
3687+
return { queuedFinal: true };
3688+
});
3689+
3690+
await dispatchWithContext({ context: createReasoningStreamContext() });
3691+
3692+
expect(reasoningDraftStream.update).toHaveBeenCalledWith("Thinking\n\n_hidden_");
3693+
expect(deliverReplies).not.toHaveBeenCalled();
3694+
});
3695+
3696+
it("routes typed reasoning-only finals to durable delivery when reasoning is persistent", async () => {
3697+
loadSessionStore.mockReturnValue({
3698+
s1: { reasoningLevel: "on" },
3699+
});
3700+
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ dispatcherOptions }) => {
3701+
await dispatcherOptions.deliver(
3702+
{ text: "<think>hidden</think>", isReasoning: true },
3703+
{ kind: "final" },
3704+
);
3705+
return { queuedFinal: true };
3706+
});
3707+
3708+
await dispatchWithContext({
3709+
context: createContext({
3710+
ctxPayload: { SessionKey: "s1" } as unknown as TelegramMessageContext["ctxPayload"],
3711+
}),
3712+
});
3713+
3714+
const delivered = expectDeliveredReply(0, { text: "Thinking\n\n_hidden_" });
3715+
expect(delivered).not.toHaveProperty("isReasoning");
3716+
});
3717+
3718+
it("does not persist typed reasoning-only finals in progress stream mode", async () => {
3719+
const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 });
3720+
dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ dispatcherOptions }) => {
3721+
await dispatcherOptions.deliver(
3722+
{ text: "<think>hidden</think>", isReasoning: true },
3723+
{ kind: "final" },
3724+
);
3725+
return { queuedFinal: true };
3726+
});
3727+
3728+
await dispatchWithContext({
3729+
context: createReasoningStreamContext(),
3730+
streamMode: "progress",
3731+
});
3732+
3733+
expect(deliverReplies).not.toHaveBeenCalled();
3734+
expect(answerDraftStream.update).not.toHaveBeenCalled();
3735+
});
3736+
36273737
it("keeps unflagged angle-bracket text visible on the answer lane", async () => {
36283738
const { answerDraftStream } = setupDraftStreams({
36293739
answerMessageId: 2001,

‎extensions/telegram/src/bot-message-dispatch.ts‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1005,6 +1005,8 @@ export const dispatchTelegramMessage = async ({
10051005
};
10061006
const answerLane = lanes.answer;
10071007
const reasoningLane = lanes.reasoning;
1008+
const durableReasoningPayloadsEnabled =
1009+
resolvedReasoningLevel === "on" || Boolean(reasoningLane.stream);
10081010
const streamToolProgressEnabled = resolveChannelStreamingPreviewToolProgress(telegramCfg);
10091011
let lastAnswerPartialText = "";
10101012
let activeAnswerDraftIsToolProgressOnly = false;
@@ -1600,13 +1602,19 @@ export const dispatchTelegramMessage = async ({
16001602
return { ...payload, replyToId: implicitQuoteReplyTargetId };
16011603
};
16021604
const normalizeDeliveryPayload = (payload: ReplyPayload): ReplyPayload | undefined => {
1603-
return projectOutboundPayloadPlanForDelivery(
1604-
createOutboundPayloadPlan([payload], {
1605+
const keepReasoningLane = payload.isReasoning === true && durableReasoningPayloadsEnabled;
1606+
const payloadForPlan = keepReasoningLane ? { ...payload } : payload;
1607+
if (keepReasoningLane) {
1608+
delete payloadForPlan.isReasoning;
1609+
}
1610+
const normalized = projectOutboundPayloadPlanForDelivery(
1611+
createOutboundPayloadPlan([payloadForPlan], {
16051612
cfg,
16061613
sessionKey: ctxPayload.SessionKey,
16071614
surface: "telegram",
16081615
}),
16091616
)[0];
1617+
return normalized;
16101618
};
16111619
const usesNativeTelegramQuote = (payload: ReplyPayload): boolean => {
16121620
if (replyQuoteText != null) {
@@ -2411,6 +2419,7 @@ export const dispatchTelegramMessage = async ({
24112419
},
24122420
commentaryProgressEnabled:
24132421
streamMode === "progress" ? progressDraft.commentaryProgressEnabled : undefined,
2422+
reasoningPayloadsEnabled: durableReasoningPayloadsEnabled,
24142423
onToolStart: async (payload) => {
24152424
const toolName = payload.name?.trim();
24162425
const progressPromise = pushStreamToolProgress(

‎src/auto-reply/get-reply-options.types.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -155,6 +155,8 @@ export type GetReplyOptions = {
155155
}) => Promise<void> | void;
156156
/** In progress mode, classify Claude pre-tool text; true also renders it as commentary. */
157157
commentaryProgressEnabled?: boolean;
158+
/** Deliver durable reasoning payloads to channels that own a separate reasoning lane. */
159+
reasoningPayloadsEnabled?: boolean;
158160
/** Called when the agent emits a structured plan update. */
159161
onPlanUpdate?: (payload: {
160162
phase?: string;

‎src/auto-reply/reply/dispatch-from-config.test.ts‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7932,6 +7932,53 @@ describe("dispatchReplyFromConfig", () => {
79327932
expect((finalCalls[0]?.[0] as ReplyPayload | undefined)?.text).toBe("The answer is 42");
79337933
});
79347934

7935+
it("delivers isReasoning final replies when the channel opts in", async () => {
7936+
setNoAbort();
7937+
const dispatcher = createDispatcher();
7938+
const ctx = buildTestCtx({ Provider: "telegram", Surface: "telegram" });
7939+
const replyResolver = async () =>
7940+
[
7941+
{ text: "thinking...", isReasoning: true },
7942+
{ text: "The answer is 42" },
7943+
] satisfies ReplyPayload[];
7944+
7945+
await dispatchReplyFromConfig({
7946+
ctx,
7947+
cfg: emptyConfig,
7948+
dispatcher,
7949+
replyOptions: { reasoningPayloadsEnabled: true },
7950+
replyResolver,
7951+
});
7952+
7953+
const finalCalls = (dispatcher.sendFinalReply as ReturnType<typeof vi.fn>).mock.calls;
7954+
expect(finalCalls.map((call) => (call[0] as ReplyPayload).text)).toEqual([
7955+
"thinking...",
7956+
"The answer is 42",
7957+
]);
7958+
});
7959+
7960+
it("does not synthesize opted-in final reasoning payloads into TTS media", async () => {
7961+
setNoAbort();
7962+
ttsMocks.state.synthesizeFinalAudio = true;
7963+
const dispatcher = createDispatcher();
7964+
const ctx = buildTestCtx({ Provider: "telegram", Surface: "telegram" });
7965+
const reasoningPayload = {
7966+
text: "thinking...",
7967+
isReasoning: true,
7968+
} satisfies ReplyPayload;
7969+
7970+
await dispatchReplyFromConfig({
7971+
ctx,
7972+
cfg: emptyConfig,
7973+
dispatcher,
7974+
replyOptions: { reasoningPayloadsEnabled: true },
7975+
replyResolver: async () => reasoningPayload,
7976+
});
7977+
7978+
expect(ttsMocks.maybeApplyTtsToPayload).not.toHaveBeenCalled();
7979+
expect(dispatcher.sendFinalReply).toHaveBeenCalledWith(reasoningPayload);
7980+
});
7981+
79357982
it("suppresses isReasoning payloads from block replies (generic dispatch path)", async () => {
79367983
setNoAbort();
79377984
const dispatcher = createDispatcher();
@@ -7960,6 +8007,43 @@ describe("dispatchReplyFromConfig", () => {
79608007
expect(blockReplySentTexts).toContain("The answer is 42");
79618008
});
79628009

8010+
it("delivers opted-in block reasoning payloads without applying TTS", async () => {
8011+
setNoAbort();
8012+
const dispatcher = createDispatcher();
8013+
const ctx = buildTestCtx({ Provider: "telegram", Surface: "telegram" });
8014+
const blockReplySentTexts: string[] = [];
8015+
const replyResolver = async (
8016+
_ctx: MsgContext,
8017+
opts?: GetReplyOptions,
8018+
): Promise<ReplyPayload | undefined> => {
8019+
await opts?.onBlockReply?.({ text: "thinking...", isReasoning: true });
8020+
await opts?.onBlockReply?.({ text: "The answer is 42" });
8021+
return undefined;
8022+
};
8023+
(dispatcher.sendBlockReply as ReturnType<typeof vi.fn>).mockImplementation(
8024+
(payload: ReplyPayload) => {
8025+
if (payload.text) {
8026+
blockReplySentTexts.push(payload.text);
8027+
}
8028+
return true;
8029+
},
8030+
);
8031+
8032+
await dispatchReplyFromConfig({
8033+
ctx,
8034+
cfg: emptyConfig,
8035+
dispatcher,
8036+
replyOptions: { reasoningPayloadsEnabled: true },
8037+
replyResolver,
8038+
});
8039+
8040+
expect(blockReplySentTexts).toEqual(["thinking...", "The answer is 42"]);
8041+
const blockTtsCalls = ttsMocks.maybeApplyTtsToPayload.mock.calls
8042+
.map(([call]) => call as { kind?: unknown; payload?: ReplyPayload })
8043+
.filter((call) => call.kind === "block");
8044+
expect(blockTtsCalls.map((call) => call.payload?.text)).toEqual(["The answer is 42"]);
8045+
});
8046+
79638047
it("strips split TTS directives from streamed block text before delivery", async () => {
79648048
setNoAbort();
79658049
ttsMocks.state.synthesizeFinalAudio = true;

‎src/auto-reply/reply/dispatch-from-config.ts‎

Lines changed: 39 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -2050,6 +2050,7 @@ export async function dispatchReplyFromConfig(
20502050
suppressHookUserDelivery,
20512051
suppressHookReplyLifecycle,
20522052
} = sourceReplyPolicy;
2053+
const reasoningPayloadsEnabled = params.replyOptions?.reasoningPayloadsEnabled === true;
20532054
const attachSourceReplyDeliveryMode = (
20542055
result: DispatchFromConfigResult,
20552056
): DispatchFromConfigResult =>
@@ -2481,16 +2482,19 @@ export async function dispatchReplyFromConfig(
24812482
markInboundDedupeReplayUnsafe();
24822483
finalReplyDeliveryStarted = true;
24832484
}
2484-
const ttsPayload = await maybeApplyTtsToReplyPayload({
2485-
payload,
2486-
cfg,
2487-
channel: deliveryChannel,
2488-
kind: "final",
2489-
inboundAudio,
2490-
ttsAuto: sessionTtsAuto,
2491-
agentId: sessionAgentId,
2492-
accountId: replyRoute.accountId,
2493-
});
2485+
const ttsPayload =
2486+
payload.isReasoning === true
2487+
? payload
2488+
: await maybeApplyTtsToReplyPayload({
2489+
payload,
2490+
cfg,
2491+
channel: deliveryChannel,
2492+
kind: "final",
2493+
inboundAudio,
2494+
ttsAuto: sessionTtsAuto,
2495+
agentId: sessionAgentId,
2496+
accountId: replyRoute.accountId,
2497+
});
24942498
throwIfFinalDeliveryAborted();
24952499
const normalizedPayload = await normalizeReplyMediaPayload(ttsPayload);
24962500
throwIfFinalDeliveryAborted();
@@ -3078,6 +3082,7 @@ export async function dispatchReplyFromConfig(
30783082
deliverStandaloneCommentaryProgress ||
30793083
canForwardSuppressedSourceItemEvents ||
30803084
params.replyOptions?.commentaryProgressEnabled,
3085+
reasoningPayloadsEnabled,
30813086
onCommandOutput: wrapProgressCallback(params.replyOptions?.onCommandOutput, {
30823087
forwardWhenSourceDeliverySuppressed: true,
30833088
requiresToolSummaryVisibility: true,
@@ -3306,17 +3311,16 @@ export async function dispatchReplyFromConfig(
33063311
if (suppressDelivery) {
33073312
return;
33083313
}
3309-
// Suppress reasoning payloads — channels using this generic dispatch
3310-
// path (WhatsApp, web, etc.) do not have a dedicated reasoning lane.
3311-
// Telegram has its own dispatch path that handles reasoning splitting.
3312-
if (payload.isReasoning === true) {
3314+
// Durable reasoning is a channel-owned lane; generic channels
3315+
// keep the historical suppression unless they explicitly opt in.
3316+
if (payload.isReasoning === true && !reasoningPayloadsEnabled) {
33133317
return;
33143318
}
33153319
// Accumulate block text for TTS generation after streaming.
33163320
// Exclude status notices — they are informational UI signals
33173321
// and must not be synthesised into the spoken reply.
33183322
const isStatusNotice = isReplyPayloadStatusNotice(payload);
3319-
if (payload.text && !isStatusNotice) {
3323+
if (payload.text && !isStatusNotice && payload.isReasoning !== true) {
33203324
const joinsBufferedTtsDirective =
33213325
cleanBlockTtsDirectiveText?.hasBufferedDirectiveText() === true;
33223326
if (accumulatedBlockText.length > 0) {
@@ -3330,7 +3334,10 @@ export async function dispatchReplyFromConfig(
33303334
blockCount++;
33313335
}
33323336
const visiblePayload =
3333-
payload.text && cleanBlockTtsDirectiveText && !isStatusNotice
3337+
payload.text &&
3338+
cleanBlockTtsDirectiveText &&
3339+
!isStatusNotice &&
3340+
payload.isReasoning !== true
33343341
? (() => {
33353342
const text = cleanBlockTtsDirectiveText.push(payload.text);
33363343
return copyReplyPayloadMetadata(payload, {
@@ -3359,16 +3366,19 @@ export async function dispatchReplyFromConfig(
33593366
if (isDispatchOperationAborted()) {
33603367
return;
33613368
}
3362-
const ttsPayload = await maybeApplyTtsToReplyPayload({
3363-
payload: visiblePayload,
3364-
cfg,
3365-
channel: deliveryChannel,
3366-
kind: "block",
3367-
inboundAudio,
3368-
ttsAuto: sessionTtsAuto,
3369-
agentId: sessionAgentId,
3370-
accountId: replyRoute.accountId,
3371-
});
3369+
const ttsPayload =
3370+
payload.isReasoning === true
3371+
? visiblePayload
3372+
: await maybeApplyTtsToReplyPayload({
3373+
payload: visiblePayload,
3374+
cfg,
3375+
channel: deliveryChannel,
3376+
kind: "block",
3377+
inboundAudio,
3378+
ttsAuto: sessionTtsAuto,
3379+
agentId: sessionAgentId,
3380+
accountId: replyRoute.accountId,
3381+
});
33723382
const normalizedPayload = await normalizeReplyMediaPayload(ttsPayload);
33733383
if (isDispatchOperationAborted()) {
33743384
return;
@@ -3479,9 +3489,9 @@ export async function dispatchReplyFromConfig(
34793489
(ctx.InboundEventKind !== "room_event" || explicitCommandTurnCtx);
34803490
for (const [replyIndex, reply] of replies.entries()) {
34813491
throwIfDispatchOperationAborted();
3482-
// Suppress reasoning payloads from channel delivery — channels using this
3483-
// generic dispatch path do not have a dedicated reasoning lane.
3484-
if (reply.isReasoning === true) {
3492+
// Durable reasoning is a channel-owned lane; generic channels keep the
3493+
// historical suppression unless they explicitly opt in.
3494+
if (reply.isReasoning === true && !reasoningPayloadsEnabled) {
34853495
continue;
34863496
}
34873497
if (suppressDelivery && !shouldDeliverDespiteSourceReplySuppression(reply)) {

0 commit comments

Comments
 (0)