From c691f2e41cd61b26b5457541f5ff5ecbd7f2a9a3 Mon Sep 17 00:00:00 2001 From: Josh Avant <830519+joshavant@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:40:33 -0500 Subject: [PATCH] fix(progress): preserve callback acceptance results (#120171) * fix(progress): preserve callback acceptance results * fix(progress): require transport acknowledgements * fix(progress): preserve direct acceptance outcomes --- .../.generated/plugin-sdk-api-baseline.sha256 | 4 +- docs/plugins/sdk-channel-plugins.md | 8 ++ extensions/clickclack/src/activity.ts | 8 +- extensions/clickclack/src/inbound.ts | 1 + extensions/clickclack/src/progress.ts | 7 +- .../monitor/message-handler.draft-preview.ts | 33 +++--- ...e-handler.draft-preview.visibility.test.ts | 65 +++++++++++ .../message-handler.process-progress.ts | 30 ++++-- ...essage-handler.process.draft-final.test.ts | 2 +- ...age-handler.process.draft-progress.test.ts | 32 ++++++ .../message-handler.process.test-harness.ts | 2 +- extensions/feishu/src/reply-dispatcher.ts | 32 +++--- .../imessage/src/monitor/monitor-provider.ts | 2 + .../monitor/handler-draft-controller.ts | 35 +++--- ...andler-draft-controller.visibility.test.ts | 64 +++++++++++ .../src/matrix/monitor/handler-runtime.ts | 1 + .../matrix/src/matrix/monitor/handler.ts | 4 +- .../mattermost/src/mattermost/monitor-turn.ts | 27 +++-- extensions/msteams/src/reply-dispatcher.ts | 25 +++-- .../src/engine/gateway/outbound-dispatch.ts | 3 +- .../src/engine/messaging/streaming-c2c.ts | 11 +- .../event-handler.inbound-context.test.ts | 6 +- .../signal/src/monitor/event-handler.ts | 3 + .../message-handler/dispatch-progress.ts | 102 +++++++++--------- .../dispatch.preview-fallback.test.ts | 25 +++++ .../src/monitor/message-handler/dispatch.ts | 43 +++++--- .../src/bot-message-dispatch-progress.ts | 26 +++-- .../telegram/src/bot-message-dispatch-turn.ts | 63 +++++++---- ...-message-dispatch.progress-updates.test.ts | 27 ++++- .../auto-reply/monitor/inbound-dispatch.ts | 3 + .../embedded-agent-runner/run/params.ts | 2 +- .../embedded-agent-subscribe.callback.ts | 4 +- src/agents/embedded-agent-subscribe.types.ts | 2 +- src/auto-reply/get-reply-options.types.ts | 37 ++++--- src/auto-reply/reply/agent-event-bridge.ts | 8 +- .../reply/agent-runner-cli-candidate.ts | 25 +++-- .../reply/agent-runner-cli-dispatch.ts | 4 +- .../reply/agent-runner-embedded-candidate.ts | 35 +++--- .../reply/agent-runner-presentation.ts | 11 +- src/auto-reply/reply/agent-runner-run.ts | 9 +- .../agent-runner.runreplyagent.e2e.test.ts | 38 ++++--- ...from-config.delivery-and-tts.test-utils.ts | 4 +- .../dispatch-from-config.prepare-execution.ts | 2 +- ...ispatch-from-config.progress.test-utils.ts | 4 +- .../reply/followup-turn-execution.test.ts | 30 +++++- .../reply/followup-turn-execution.ts | 93 ++++++++++++---- src/channels/progress-draft-compositor.ts | 58 ++++++---- ...ogress-draft-compositor.visibility.test.ts | 57 ++++++++++ src/channels/progress-visibility.ts | 9 ++ src/plugin-sdk/reply-runtime.contract.test.ts | 35 ++++++ 50 files changed, 858 insertions(+), 303 deletions(-) create mode 100644 extensions/discord/src/monitor/message-handler.draft-preview.visibility.test.ts create mode 100644 extensions/matrix/src/matrix/monitor/handler-draft-controller.visibility.test.ts create mode 100644 src/channels/progress-draft-compositor.visibility.test.ts create mode 100644 src/channels/progress-visibility.ts create mode 100644 src/plugin-sdk/reply-runtime.contract.test.ts diff --git a/docs/.generated/plugin-sdk-api-baseline.sha256 b/docs/.generated/plugin-sdk-api-baseline.sha256 index c826a4feaa7a..260a3ce16dc4 100644 --- a/docs/.generated/plugin-sdk-api-baseline.sha256 +++ b/docs/.generated/plugin-sdk-api-baseline.sha256 @@ -34,8 +34,8 @@ ba41c40956d6b4565605fa38c2d12f4b8471a0f8afe842798716b9032ee4d74a module/channel df567ce2f4a4ba8a0937f825c46e83763412a724e36b72ce727dc4203c7dd134 module/channel-ingress-runtime 0e6efb79730fae59bb549ad00d9af2848c139b4bb1762e83b66886284e1c1421 module/channel-lifecycle 159d034b431d113f3a6dc41ec0bcadba2d6664051f158330b0e3dd3da8b5d42f module/channel-logging -2d2b4b0440e61ef3697891883ca114f2b4bc4fff09c375b86e8d3bfec5f59bb5 module/channel-message -d22cd6190bb3d0d0914466f3b5e1fe1787930572b5c212774a716fe2136112f0 module/channel-outbound +e649231e5136e8a8045c098754adad4b2e7bf7acc29c759e02015d244f7f107a module/channel-message +146c87e18187fbf90d7780d946b614c89c415e084c7206de2f0be1928a2b3434 module/channel-outbound 930beff13ed42a138f65164013c82f4f4c96422292c5fc634a5edee1dce71367 module/channel-pairing ee4292b069d4d48cce4fc2dc26df5b5c87eb1fa4769f1f6be9a10c3e1221e1a9 module/channel-plugin-common 94ef57c8f6087fcaa56e59e493c391a04377ed03f23c681edd8d8f6e2d64e0da module/channel-policy diff --git a/docs/plugins/sdk-channel-plugins.md b/docs/plugins/sdk-channel-plugins.md index 81145deb9d8e..6dc3bb3354d2 100644 --- a/docs/plugins/sdk-channel-plugins.md +++ b/docs/plugins/sdk-channel-plugins.md @@ -77,6 +77,14 @@ and `verifyChannelMessageLiveFinalizerProofs(...)` tests so native preview, progress, edit, fallback/retention, cleanup, and receipt behavior cannot drift silently. +### Progress visibility acceptance + +Progress callbacks report what the operator can see, not merely what a plugin queued. Return +`true` after accepting visible progress and `false` while delivery is pending or when no visible +update occurred. Existing synchronous and asynchronous callbacks that return `void` remain +backward-compatible and are treated as visible; new acceptance-aware implementations should use +an explicit boolean. + Inbound receivers that defer platform acknowledgements should declare `message.receive.defaultAckPolicy` and `supportedAckPolicies` instead of hiding ack timing in monitor-local state. Cover every declared policy with diff --git a/extensions/clickclack/src/activity.ts b/extensions/clickclack/src/activity.ts index 85e0bb92a29c..795e5bf5f510 100644 --- a/extensions/clickclack/src/activity.ts +++ b/extensions/clickclack/src/activity.ts @@ -139,7 +139,7 @@ type ToolRow = { /** Publisher wired into one agent turn via `replyOptions.onItemEvent`. */ export type ClickClackActivityPublisher = { - onItemEvent: (payload: ClickClackItemEventPayload) => void; + onItemEvent: (payload: ClickClackItemEventPayload) => false; /** * Records the resolved model/thinking for this turn (from * `replyOptions.onModelSelected`); stamped onto subsequent activity rows. @@ -300,12 +300,14 @@ export function createClickClackActivityPublisher(params: { const kind = normalizedItemKind(payload); if (STREAMING_COMMENTARY_ITEM_KINDS.has(kind)) { handleCommentary(payload); - return; + return false; } if (SKIPPED_ITEM_KINDS.has(kind)) { - return; + return false; } handleDiscreteItem(payload); + // Activity transport is serialized in the background; queueing is not visibility. + return false; }, setProvenance: (next) => { provenance = next; diff --git a/extensions/clickclack/src/inbound.ts b/extensions/clickclack/src/inbound.ts index d08dd34867d1..a4420f16b26d 100644 --- a/extensions/clickclack/src/inbound.ts +++ b/extensions/clickclack/src/inbound.ts @@ -238,6 +238,7 @@ export async function handleClickClackInbound(params: { onItemEvent: (payload: ClickClackItemEventPayload) => { progress?.onItemEvent(payload); activity?.onItemEvent(payload); + return false; }, commentaryProgressEnabled: true, // ClickClack owns the native progress rendering, so item events must flow diff --git a/extensions/clickclack/src/progress.ts b/extensions/clickclack/src/progress.ts index e4861860a5cf..fc74e43c9348 100644 --- a/extensions/clickclack/src/progress.ts +++ b/extensions/clickclack/src/progress.ts @@ -131,7 +131,7 @@ function createLineIdResolver(): (payload: ClickClackItemEventPayload) => string type ClickClackAgentProgressPublisher = { start(): void; - onItemEvent(payload: ClickClackItemEventPayload): void; + onItemEvent(payload: ClickClackItemEventPayload): false; finalize(): Promise; }; @@ -287,7 +287,7 @@ export function createClickClackAgentProgressPublisher(params: { }, onItemEvent(payload) { if (!started || cleared) { - return; + return false; } const id = resolveLineId(payload); const final = isFinal(payload); @@ -300,7 +300,7 @@ export function createClickClackAgentProgressPublisher(params: { if (retractsExistingCommentary && queuedLines.get(id)?.payload.op === "append") { queuedLines.delete(id); seenLines.delete(id); - return; + return false; } const line: Record = { id, @@ -316,6 +316,7 @@ export function createClickClackAgentProgressPublisher(params: { line, }); seenLines.add(id); + return false; }, async finalize() { if (!started || cleared) { diff --git a/extensions/discord/src/monitor/message-handler.draft-preview.ts b/extensions/discord/src/monitor/message-handler.draft-preview.ts index b509b6ee07f7..3a3c55c1a8b1 100644 --- a/extensions/discord/src/monitor/message-handler.draft-preview.ts +++ b/extensions/discord/src/monitor/message-handler.draft-preview.ts @@ -129,14 +129,19 @@ export function createDiscordDraftPreviewController(params: { ? buildChannelProgressDraftLineForEntry(params.discordConfig, input, options) : buildChannelProgressDraftLine(input, options), update: async (previewText, options) => { + if (!draftStream) { + return false; + } lastPartialText = previewText; draftText = previewText; hasStreamedMessage = true; draftChunker?.reset(); - draftStream?.update(previewText); + draftStream.update(previewText); if (options?.flush) { - await draftStream?.flush(); + await draftStream.flush(); } + // REST-backed draft work is pending until Discord returns a message id. + return Boolean(draftStream.messageId()); }, deleteCurrent: async () => { lastPartialText = ""; @@ -167,8 +172,9 @@ export function createDiscordDraftPreviewController(params: { const pushPreambleHeadline = async (text?: string, options?: { itemId?: string }) => { if (discordStreamMode === "progress") { - await progressDraft.pushPreambleHeadline(text, options); + return await progressDraft.pushPreambleHeadline(text, options); } + return false; }; const beginNewProgressTurn = (options?: { force?: boolean }) => { @@ -268,36 +274,39 @@ export function createDiscordDraftPreviewController(params: { line?: string | ChannelProgressDraftLine, options?: { toolName?: string }, ) { - await progressDraft.pushToolProgress(line, options); + return await progressDraft.pushToolProgress(line, options); }, async pushPlanProgress(steps?: AgentPlanStep[], options?: { explanation?: string }) { - await progressDraft.pushPlanProgress(steps, options); + return await progressDraft.pushPlanProgress(steps, options); }, async pushReasoningProgress(text?: string, options?: { snapshot?: boolean }) { - await progressDraft.pushReasoningProgress(text, options); + return await progressDraft.pushReasoningProgress(text, options); }, async pushNarrationProgress(text?: string) { - await progressDraft.pushNarrationProgress(text); + return await progressDraft.pushNarrationProgress(text); }, pushPreambleHeadline, async pushPreambleItemEvent( payload: { itemId?: string; progressText?: string }, noteCommentary: (itemId?: string, text?: string) => void, ) { - await pushPreambleHeadline(payload.progressText, { itemId: payload.itemId }); + const headlineAccepted = await pushPreambleHeadline(payload.progressText, { + itemId: payload.itemId, + }); if (!progressDraft.commentaryProgressEnabled) { - return; + return headlineAccepted; } - const accepted = await progressDraft.pushCommentaryProgress(payload.progressText, { + const commentaryAccepted = await progressDraft.pushCommentaryProgress(payload.progressText, { itemId: payload.itemId, }); // Count only sanitized commentary that actually streamed to the window. - if (accepted) { + if (commentaryAccepted) { noteCommentary(payload.itemId, payload.progressText); } + return headlineAccepted || commentaryAccepted; }, async pushCommentaryProgress(text?: string, options?: { itemId?: string }) { - await progressDraft.pushCommentaryProgress(text, options); + return await progressDraft.pushCommentaryProgress(text, options); }, resolvePreviewFinalText(text?: string) { if (typeof text !== "string") { diff --git a/extensions/discord/src/monitor/message-handler.draft-preview.visibility.test.ts b/extensions/discord/src/monitor/message-handler.draft-preview.visibility.test.ts new file mode 100644 index 000000000000..f6aef22aba7e --- /dev/null +++ b/extensions/discord/src/monitor/message-handler.draft-preview.visibility.test.ts @@ -0,0 +1,65 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const draftStream = vi.hoisted(() => ({ + update: vi.fn(), + flush: vi.fn(async () => {}), + messageId: vi.fn<() => string | undefined>(() => undefined), + clear: vi.fn(async () => {}), + deleteCurrentMessage: vi.fn(async () => {}), + discardPending: vi.fn(async () => {}), + seal: vi.fn(async () => {}), + stop: vi.fn(async () => {}), + retarget: vi.fn(async () => {}), + cleanupRetargeted: vi.fn(async () => {}), + forceNewMessage: vi.fn(), +})); + +vi.mock("../draft-stream.js", () => ({ + createDiscordDraftStream: () => draftStream, +})); + +import { createDiscordDraftPreviewController } from "./message-handler.draft-preview.js"; + +describe("Discord progress visibility", () => { + beforeEach(() => { + vi.useFakeTimers(); + for (const mock of Object.values(draftStream)) { + mock.mockClear(); + } + draftStream.messageId.mockReturnValue(undefined); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("retries identical progress until Discord acknowledges a draft message", async () => { + const controller = createDiscordDraftPreviewController({ + cfg: {}, + discordConfig: { streaming: { mode: "progress" } }, + accountId: "default", + sourceRepliesAreToolOnly: false, + textLimit: 2_000, + deliveryRest: {} as never, + deliverChannelId: "channel-1", + replyReference: { peek: () => undefined }, + tableMode: "off", + maxLinesPerMessage: undefined, + chunkMode: "length", + log: vi.fn(), + }); + const progress = { itemId: "item-1", progressText: "still working" }; + + expect(await controller.pushItemEvent(progress)).toBe(false); + await vi.advanceTimersByTimeAsync(1_500); + expect(draftStream.update).toHaveBeenCalledTimes(1); + + expect(await controller.pushItemEvent(progress)).toBe(false); + expect(draftStream.update).toHaveBeenCalledTimes(2); + + draftStream.messageId.mockReturnValue("message-1"); + expect(await controller.pushItemEvent(progress)).toBe(true); + expect(draftStream.update).toHaveBeenCalledTimes(3); + await controller.cleanup(); + }); +}); diff --git a/extensions/discord/src/monitor/message-handler.process-progress.ts b/extensions/discord/src/monitor/message-handler.process-progress.ts index 71f66de457fb..6fd16a997ebe 100644 --- a/extensions/discord/src/monitor/message-handler.process-progress.ts +++ b/extensions/discord/src/monitor/message-handler.process-progress.ts @@ -88,11 +88,17 @@ export function createDiscordMessageProgressRuntime(params: { const buildProgressSummaryLine = () => `-# ${progressReceipt.buildSummaryLine()}`; const replyOptions: Partial = { - onAssistantMessageStart: draftPreview.draftStream ? handleAssistantMessageBoundary : undefined, + onAssistantMessageStart: draftPreview.draftStream + ? () => { + handleAssistantMessageBoundary(); + return false; + } + : undefined, onReasoningEnd: draftPreview.draftStream ? () => { progressReceipt.closeReasoning(); handleAssistantMessageBoundary(); + return false; } : undefined, onQueuedFollowupAdmitted: draftPreview.draftStream @@ -140,27 +146,27 @@ export function createDiscordMessageProgressRuntime(params: { narrationHideCommandText: draftPreview.narrationHideCommandText ? true : undefined, onReasoningStream: async (payload) => { if (payload?.requiresReasoningProgressOptIn === true && !reasoningWindowEnabled) { - return; + return false; } if (payload?.text) { progressReceipt.noteReasoning(); } await params.reactions.controller.setThinking(); - await draftPreview.pushReasoningProgress(payload?.text, { + return await draftPreview.pushReasoningProgress(payload?.text, { snapshot: payload?.isReasoningSnapshot === true, }); }, streamReasoningInNonStreamModes: reasoningWindowEnabled, onToolStart: async (payload) => { if (isProcessAborted(abortSignal)) { - return; + return false; } await params.reactions.maybeBindToToolReaction(payload); await params.reactions.controller.setTool(payload.name); if (payload.phase === "start") { progressReceipt.noteToolCall(payload.name); } - await draftPreview.pushToolEvent(payload); + return await draftPreview.pushToolEvent(payload); }, onItemEvent: async (payload) => { if (isFailedProgress(payload)) { @@ -174,38 +180,40 @@ export function createDiscordMessageProgressRuntime(params: { progressReceipt.noteCommentary(itemId, text); }); } - await draftPreview.pushItemEvent(payload); + return await draftPreview.pushItemEvent(payload); }, onPlanUpdate: async (payload) => { if (payload.phase === "update") { - await draftPreview.pushPlanProgress(payload.steps, { + return await draftPreview.pushPlanProgress(payload.steps, { explanation: payload.explanation, }); } + return false; }, onApprovalEvent: async (payload) => { - await draftPreview.pushApprovalEvent(payload); + return await draftPreview.pushApprovalEvent(payload); }, onCommandOutput: async (payload) => { if (isFailedProgress(payload)) { return false; } - await draftPreview.pushCommandOutputEvent(payload); - return undefined; + return await draftPreview.pushCommandOutputEvent(payload); }, onPatchSummary: async (payload) => { - await draftPreview.pushPatchEvent(payload); + return await draftPreview.pushPatchEvent(payload); }, onCompactionStart: async () => { if (!isProcessAborted(abortSignal)) { await params.reactions.controller.setCompacting(); } + return false; }, onCompactionEnd: async () => { if (!isProcessAborted(abortSignal)) { params.reactions.controller.cancelPending(); await params.reactions.controller.setThinking(); } + return false; }, }; diff --git a/extensions/discord/src/monitor/message-handler.process.draft-final.test.ts b/extensions/discord/src/monitor/message-handler.process.draft-final.test.ts index a3808e1e55c6..b06f38f5abaf 100644 --- a/extensions/discord/src/monitor/message-handler.process.draft-final.test.ts +++ b/extensions/discord/src/monitor/message-handler.process.draft-final.test.ts @@ -430,7 +430,7 @@ describe("processDiscordMessage draft streaming final delivery", () => { it("declines failed item progress without updating the Discord draft", async () => { const draftStream = createMockDraftStreamForTest(); - let callbackResult: false | void = undefined; + let callbackResult: boolean | void = undefined; dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => { callbackResult = await params?.replyOptions?.onItemEvent?.({ diff --git a/extensions/discord/src/monitor/message-handler.process.draft-progress.test.ts b/extensions/discord/src/monitor/message-handler.process.draft-progress.test.ts index f660680690f7..c118bbc3aa6a 100644 --- a/extensions/discord/src/monitor/message-handler.process.draft-progress.test.ts +++ b/extensions/discord/src/monitor/message-handler.process.draft-progress.test.ts @@ -252,6 +252,38 @@ describe("processDiscordMessage draft streaming progress", () => { expectFinalWithProgressReceipt("done", "💬 2 notes", "🛠️ 1 tool call"); }); + it("retries an unacknowledged preamble and reports visibility after Discord accepts it", async () => { + const draftStream = createMockDraftStreamForTest(); + draftStream.messageId.mockReturnValue(undefined); + const results: Array = []; + + dispatchInboundMessage.mockImplementationOnce(async (params?: DispatchInboundParams) => { + const preamble = { + itemId: "preamble-1", + kind: "preamble", + progressText: "Checking source data.", + }; + results.push(await params?.replyOptions?.onItemEvent?.(preamble)); + draftStream.messageId.mockReturnValue("preview-1"); + results.push(await params?.replyOptions?.onItemEvent?.(preamble)); + return createNoQueuedDispatchResult(); + }); + + const ctx = await createAutomaticSourceDeliveryContext({ + discordConfig: { + streaming: { + mode: "progress", + progress: { label: false, commentary: true }, + }, + }, + }); + + await runProcessDiscordMessage(ctx); + + expect(results).toEqual([false, true]); + expect(draftStream.update.mock.calls.length).toBeGreaterThanOrEqual(2); + }); + it.each([ ["active", true], ["inactive", false], diff --git a/extensions/discord/src/monitor/message-handler.process.test-harness.ts b/extensions/discord/src/monitor/message-handler.process.test-harness.ts index d15d58fd7150..1b3ced43850c 100644 --- a/extensions/discord/src/monitor/message-handler.process.test-harness.ts +++ b/extensions/discord/src/monitor/message-handler.process.test-harness.ts @@ -172,7 +172,7 @@ export type DispatchInboundParams = { summary?: string; title?: string; name?: string; - }) => Promise | false | void; + }) => Promise | boolean | void; onNarrationUpdate?: (payload: { text: string }) => Promise | void; onProgressNarratorLifecycle?: (lifecycle: { beginTurn: () => void; diff --git a/extensions/feishu/src/reply-dispatcher.ts b/extensions/feishu/src/reply-dispatcher.ts index 0ddd18a9b559..16aa660b026a 100644 --- a/extensions/feishu/src/reply-dispatcher.ts +++ b/extensions/feishu/src/reply-dispatcher.ts @@ -707,10 +707,11 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP statusLine = nextStatusLine; const hasStreamingSession = Boolean(streaming?.isActive() || streamingStartPromise); if (!hasStreamingSession && (options?.startIfNeeded === false || renderMode !== "card")) { - return; + return false; } startStreaming(); flushStreamingCardUpdate(buildCombinedStreamText(reasoningText, streamText)); + return false; }; const sendChunkedTextReply = async (paramsLocal: { @@ -1523,32 +1524,34 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP onPartialReply: previewStreamingEnabled ? (payload: ReplyPayload) => { if (!payload.text) { - return; + return false; } const cleaned = stripReasoningTagsFromText(payload.text, { mode: "strict", trim: "both", }); if (!cleaned) { - return; + return false; } startStreaming(); queueStreamingUpdate(cleaned, { dedupeWithLastPartial: true, mode: "snapshot", }); + return false; } : undefined, onReasoningStream: reasoningPreviewEnabled ? (payload: ReplyPayload) => { if (!payload.text) { - return; + return false; } startStreaming(); queueReasoningUpdate(formatReasoningMessage(payload.text)); + return false; } : undefined, - onReasoningEnd: reasoningPreviewEnabled ? () => {} : undefined, + onReasoningEnd: reasoningPreviewEnabled ? () => false : undefined, onToolStart: previewStreamingEnabled ? (payload: { name?: string; @@ -1557,7 +1560,7 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP detailMode?: "explain" | "raw"; }) => { if (!isChannelProgressDraftWorkToolName(payload.name)) { - return; + return false; } const statusLineLocal = formatChannelProgressDraftLineForEntry( account.config, @@ -1572,25 +1575,18 @@ export function createFeishuReplyDispatcher(params: CreateFeishuReplyDispatcherP }, ); if (statusLineLocal) { - updateStreamingStatusLine(statusLineLocal); + return updateStreamingStatusLine(statusLineLocal); } + return false; } : undefined, onAssistantMessageStart: previewStreamingEnabled - ? () => { - updateStreamingStatusLine("", { startIfNeeded: false }); - } + ? () => updateStreamingStatusLine("", { startIfNeeded: false }) : undefined, onCompactionStart: previewStreamingEnabled - ? () => { - updateStreamingStatusLine("📦 **Compacting context...**"); - } - : undefined, - onCompactionEnd: previewStreamingEnabled - ? () => { - updateStreamingStatusLine(""); - } + ? () => updateStreamingStatusLine("📦 **Compacting context...**") : undefined, + onCompactionEnd: previewStreamingEnabled ? () => updateStreamingStatusLine("") : undefined, }, ensureNoVisibleReplyFallback, getVisibleReplyState: () => ({ diff --git a/extensions/imessage/src/monitor/monitor-provider.ts b/extensions/imessage/src/monitor/monitor-provider.ts index 023533a1f1cf..787d7c655e35 100644 --- a/extensions/imessage/src/monitor/monitor-provider.ts +++ b/extensions/imessage/src/monitor/monitor-provider.ts @@ -1227,11 +1227,13 @@ export async function monitorIMessageProvider(opts: MonitorIMessageOpts = {}): P // instead of falling back to a durable iMessage bubble. onToolResult: async () => { await directTypingController?.startTypingLoop(); + return false; }, ...(supportsTyping ? { onToolStart: async () => { await directTypingController?.startTypingLoop(); + return false; }, } : {}), diff --git a/extensions/matrix/src/matrix/monitor/handler-draft-controller.ts b/extensions/matrix/src/matrix/monitor/handler-draft-controller.ts index d8d4b9204da9..2301ccc63179 100644 --- a/extensions/matrix/src/matrix/monitor/handler-draft-controller.ts +++ b/extensions/matrix/src/matrix/monitor/handler-draft-controller.ts @@ -100,12 +100,20 @@ export async function createMatrixDraftController(params: { input.event === "approval" ? formatChannelProgressDraftLine(input, options) : buildChannelProgressDraftLineForEntry(progressConfigEntry, input, options), - update: (text) => { + update: async (text, options) => { const previewText = !progressDraftStreaming && (previewPlan || previewPlanExplanation) ? renderPreviewPlan() : text.replace(/^• /gmu, "- "); - draftStream?.update(previewText); + if (!draftStream) { + return false; + } + draftStream.update(previewText); + if (options?.flush) { + await draftStream.flush(); + } + // A queued update is not visible until Matrix has accepted a draft event. + return Boolean(draftStream.eventId()); }, }); @@ -129,21 +137,22 @@ export async function createMatrixDraftController(params: { return { ...options, onToolStart: async (payload) => { - await progressDraft.pushToolEvent(payload); + return await progressDraft.pushToolEvent(payload); }, onItemEvent: async (payload) => { - await progressDraft.pushItemEvent(payload); + return await progressDraft.pushItemEvent(payload); }, onPlanUpdate: async (payload) => { if (payload.phase !== "update") { - return; + return false; } if (progressDraftStreaming) { - await progressDraft.pushPlanProgress(payload.steps, { explanation: payload.explanation }); - return; + return await progressDraft.pushPlanProgress(payload.steps, { + explanation: payload.explanation, + }); } if (!draftStream || previewPlanSuppressed) { - return; + return false; } previewPlan = payload.steps?.length ? payload.steps.map((step) => ({ ...step })) @@ -153,15 +162,16 @@ export async function createMatrixDraftController(params: { if (text) { draftStream.update(text); } + return false; }, onApprovalEvent: async (payload) => { - await progressDraft.pushApprovalEvent(payload); + return await progressDraft.pushApprovalEvent(payload); }, onCommandOutput: async (payload) => { - await progressDraft.pushCommandOutputEvent(payload); + return await progressDraft.pushCommandOutputEvent(payload); }, onPatchSummary: async (payload) => { - await progressDraft.pushPatchEvent(payload); + return await progressDraft.pushPatchEvent(payload); }, }; }; @@ -265,7 +275,7 @@ export async function createMatrixDraftController(params: { }, onPartialReply: (text: string) => { if (progressDraftStreaming) { - return; + return false; } latestDraftFullText = text; if (text.trim()) { @@ -275,6 +285,7 @@ export async function createMatrixDraftController(params: { progressDraft.suppress(); } updateDraftFromLatestFullText(); + return false; }, }; } diff --git a/extensions/matrix/src/matrix/monitor/handler-draft-controller.visibility.test.ts b/extensions/matrix/src/matrix/monitor/handler-draft-controller.visibility.test.ts new file mode 100644 index 000000000000..ff27d018610b --- /dev/null +++ b/extensions/matrix/src/matrix/monitor/handler-draft-controller.visibility.test.ts @@ -0,0 +1,64 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; + +const draftStream = vi.hoisted(() => ({ + update: vi.fn(), + flush: vi.fn(async () => {}), + stop: vi.fn(async () => undefined), + discardPending: vi.fn(async () => {}), + finalizeLive: vi.fn(async () => true), + reset: vi.fn(), + eventId: vi.fn<() => string | undefined>(() => undefined), + content: vi.fn(() => undefined), + matchesPreparedText: vi.fn(() => false), + mustDeliverFinalNormally: vi.fn(() => false), +})); + +vi.mock("./handler-runtime.js", () => ({ + loadMatrixDraftStream: async () => ({ + createMatrixDraftStream: () => draftStream, + }), +})); + +import { createMatrixDraftController } from "./handler-draft-controller.js"; + +describe("Matrix progress visibility", () => { + beforeEach(() => { + vi.useFakeTimers(); + for (const mock of Object.values(draftStream)) { + mock.mockClear(); + } + draftStream.eventId.mockReturnValue(undefined); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("retries identical progress until Matrix acknowledges a draft event", async () => { + const controller = await createMatrixDraftController({ + streaming: "progress", + previewToolProgressEnabled: true, + replyToMode: "off", + messageId: "$inbound", + cfg: {}, + accountId: "default", + roomId: "!room:example.org", + client: {} as never, + logVerboseMessage: vi.fn(), + }); + const options = controller.buildPreviewToolProgressReplyOptions(); + const progress = { itemId: "item-1", progressText: "still working" }; + + expect(await options.onItemEvent?.(progress)).toBe(false); + await vi.advanceTimersByTimeAsync(1_500); + expect(draftStream.update).toHaveBeenCalledTimes(1); + + expect(await options.onItemEvent?.(progress)).toBe(false); + expect(draftStream.update).toHaveBeenCalledTimes(2); + + draftStream.eventId.mockReturnValue("$draft"); + expect(await options.onItemEvent?.(progress)).toBe(true); + expect(draftStream.update).toHaveBeenCalledTimes(3); + controller.cancelProgressDraft(); + }); +}); diff --git a/extensions/matrix/src/matrix/monitor/handler-runtime.ts b/extensions/matrix/src/matrix/monitor/handler-runtime.ts index 2746944a5739..ac5c8ead92fa 100644 --- a/extensions/matrix/src/matrix/monitor/handler-runtime.ts +++ b/extensions/matrix/src/matrix/monitor/handler-runtime.ts @@ -4,6 +4,7 @@ import { MATRIX_OPENCLAW_FINALIZED_PREVIEW_KEY } from "../send/types.js"; export type MatrixDraftStreamHandle = { update: (text: string) => void; + flush: () => Promise; stop: () => Promise; discardPending: () => Promise; eventId: () => string | undefined; diff --git a/extensions/matrix/src/matrix/monitor/handler.ts b/extensions/matrix/src/matrix/monitor/handler.ts index 162101a79332..f46d996ebb4e 100644 --- a/extensions/matrix/src/matrix/monitor/handler.ts +++ b/extensions/matrix/src/matrix/monitor/handler.ts @@ -539,9 +539,10 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam onBlockReplyQueued: draftStream ? (payload, context) => { if (payload.isCompactionNotice === true) { - return; + return false; } draftController.queueDraftBlockBoundary(payload, context); + return false; } : undefined, // Reset draft boundary bookkeeping on assistant message @@ -551,6 +552,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam ? () => { draftController.resetDraftBlockOffsets(); draftController.resetPreviewToolProgress(); + return false; } : undefined, onQueuedFollowupAdmitted: draftStream diff --git a/extensions/mattermost/src/mattermost/monitor-turn.ts b/extensions/mattermost/src/mattermost/monitor-turn.ts index 8cc93827ef5e..7fa8dcc5b989 100644 --- a/extensions/mattermost/src/mattermost/monitor-turn.ts +++ b/extensions/mattermost/src/mattermost/monitor-turn.ts @@ -482,18 +482,19 @@ export async function dispatchMattermostInboundTurn( onModelSelected, onPartialReply: (payloadResult) => account.streamingMode === "progress" - ? undefined + ? false : updateDraftFromPartial(payloadResult.text), onAssistantMessageStart: () => { lastPartialText = ""; progressDraft.resetReasoningProgress(); if (account.streamingMode === "block") { blockPreviewAssistantMessagePending = true; - return; + return false; } if (account.streamingMode !== "progress") { progressDraft.reset(); } + return false; }, onReasoningEnd: () => { // Hidden reasoning has no boundary; only rendered text, reasoning, or tools rotate preview posts. @@ -502,13 +503,16 @@ export async function dispatchMattermostInboundTurn( if (account.streamingMode !== "block" && account.streamingMode !== "progress") { progressDraft.reset(); } + return false; }, onReasoningStream: async (payloadResult) => { if (account.streamingMode === "progress") { - await progressDraft.pushReasoningProgress(payloadResult.text || "Thinking…", { - snapshot: payloadResult.isReasoningSnapshot === true, - }); - return; + return await progressDraft.pushReasoningProgress( + payloadResult.text || "Thinking…", + { + snapshot: payloadResult.isReasoningSnapshot === true, + }, + ); } if (!lastPartialText) { const boundarySettled = enterBlockPreviewActivity("reasoning"); @@ -516,10 +520,11 @@ export async function dispatchMattermostInboundTurn( previewBoundaryController.noteUpdate(); await boundarySettled; } + return false; }, onToolStart: async (payloadValue) => { if (!draftToolProgressEnabled) { - return; + return false; } const boundarySettled = enterBlockPreviewActivity("tool"); // Boundary detach and progress staging both happen synchronously before @@ -540,11 +545,12 @@ export async function dispatchMattermostInboundTurn( { startImmediately: true }, ); previewBoundaryController.noteUpdate(); - await Promise.all([boundarySettled, progressSettled]); + const [, visible] = await Promise.all([boundarySettled, progressSettled]); + return visible; }, onItemEvent: async (payloadLocal) => { if (!draftToolProgressEnabled) { - return; + return false; } const boundarySettled = enterBlockPreviewActivity("tool"); const progressSettled = progressDraft.pushToolProgress( @@ -563,7 +569,8 @@ export async function dispatchMattermostInboundTurn( { startImmediately: true }, ); previewBoundaryController.noteUpdate(); - await Promise.all([boundarySettled, progressSettled]); + const [, visible] = await Promise.all([boundarySettled, progressSettled]); + return visible; }, }, }), diff --git a/extensions/msteams/src/reply-dispatcher.ts b/extensions/msteams/src/reply-dispatcher.ts index 6c57290cf4e0..8f1502ae539d 100644 --- a/extensions/msteams/src/reply-dispatcher.ts +++ b/extensions/msteams/src/reply-dispatcher.ts @@ -565,11 +565,11 @@ export function createMSTeamsReplyDispatcher(params: { onReasoningStream: async (payload: PipelinePayload) => { const text = typeof payload?.text === "string" ? payload.text : undefined; if (!text) { - return; + return false; } if (payload?.isReasoningSnapshot !== true) { await streamController.pushProgressLine(text); - return; + return false; } await streamController.pushProgressLine( buildChannelProgressDraftLine({ @@ -580,6 +580,7 @@ export function createMSTeamsReplyDispatcher(params: { progressText: text, }), ); + return false; }, onToolStart: async (payload: PipelinePayload) => { const name = typeof payload?.name === "string" ? payload.name : undefined; @@ -604,6 +605,7 @@ export function createMSTeamsReplyDispatcher(params: { ), name ? { toolName: name } : undefined, ); + return false; }, onItemEvent: async (payload: PipelinePayload) => { await streamController.pushProgressLine( @@ -625,18 +627,20 @@ export function createMSTeamsReplyDispatcher(params: { ...(typeof payload?.meta === "string" ? { meta: payload.meta } : {}), }), ); + return false; }, onPlanUpdate: async (payload: PipelinePayload) => { if (payload?.phase !== "update") { - return; + return false; } await streamController.pushPlanProgress(normalizeAgentPlanSteps(payload.steps), { explanation: typeof payload.explanation === "string" ? payload.explanation : undefined, }); + return false; }, onApprovalEvent: async (payload: PipelinePayload) => { if (payload?.phase !== "requested") { - return; + return false; } await streamController.pushProgressLine( buildChannelProgressDraftLine({ @@ -648,10 +652,11 @@ export function createMSTeamsReplyDispatcher(params: { ...(typeof payload?.message === "string" ? { message: payload.message } : {}), }), ); + return false; }, onCommandOutput: async (payload: PipelinePayload) => { if (payload?.phase !== "end") { - return; + return false; } await streamController.pushProgressLine( buildChannelProgressDraftLine({ @@ -667,10 +672,11 @@ export function createMSTeamsReplyDispatcher(params: { ...(typeof payload?.exitCode === "number" ? { exitCode: payload.exitCode } : {}), }), ); + return false; }, onPatchSummary: async (payload: PipelinePayload) => { if (payload?.phase !== "end") { - return; + return false; } await streamController.pushProgressLine( buildChannelProgressDraftLine({ @@ -697,6 +703,7 @@ export function createMSTeamsReplyDispatcher(params: { ...(typeof payload?.summary === "string" ? { summary: payload.summary } : {}), }), ); + return false; }, } : {}; @@ -710,8 +717,10 @@ export function createMSTeamsReplyDispatcher(params: { replyOptions: { ...(streamController.hasStream() ? { - onPartialReply: (payload: { text?: string }) => - streamController.onPartialReply(payload), + onPartialReply: (payload: { text?: string }) => { + streamController.onPartialReply(payload); + return false; + }, } : {}), ...progressCallbacks, diff --git a/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts b/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts index 2f9e6f609bfc..6b59882b8994 100644 --- a/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts +++ b/extensions/qqbot/src/engine/gateway/outbound-dispatch.ts @@ -697,11 +697,12 @@ export async function dispatchOutbound( ? { onPartialReply: async (payload: { text?: string }) => { try { - await streamingController.onPartialReply(payload); + return await streamingController.onPartialReply(payload); } catch (partialErr) { log?.error( `Streaming onPartialReply error: ${partialErr instanceof Error ? partialErr.message : String(partialErr)}`, ); + return false; } }, } diff --git a/extensions/qqbot/src/engine/messaging/streaming-c2c.ts b/extensions/qqbot/src/engine/messaging/streaming-c2c.ts index 7260cc6f9fcc..e03377b02731 100644 --- a/extensions/qqbot/src/engine/messaging/streaming-c2c.ts +++ b/extensions/qqbot/src/engine/messaging/streaming-c2c.ts @@ -464,17 +464,17 @@ export class StreamingController { * payload.text 是从头到尾的完整当前文本（每次回调都是全量）。 * 核心逻辑：normalize → 更新 lastNormalizedFull → 从 sentIndex 开始 processMediaTags */ - async onPartialReply(payload: { text?: string }): Promise { + async onPartialReply(payload: { text?: string }): Promise { if (this.isTerminalPhase) { - return; + return false; } if (!payload.text) { - return; + return false; } // ★ 互斥锁在入口检查：如果已被 deliver 锁定，直接跳过，无需排队 if (!this.acquireCallbackLock("partial")) { - return; + return false; } // 将实际逻辑挂到 Promise 链尾部，保证串行执行 @@ -486,7 +486,8 @@ export class StreamingController { return this.handlePartialReply(payload); }, ); - return this.callbackChain; + await this.callbackChain; + return this.sentStreamChunkCount > 0 || this.streamMsgId !== null; } /** onPartialReply 的实际逻辑（由 callbackChain 保证串行调用） */ diff --git a/extensions/signal/src/monitor/event-handler.inbound-context.test.ts b/extensions/signal/src/monitor/event-handler.inbound-context.test.ts index fc1497fd542d..d74913f9d9ff 100644 --- a/extensions/signal/src/monitor/event-handler.inbound-context.test.ts +++ b/extensions/signal/src/monitor/event-handler.inbound-context.test.ts @@ -28,9 +28,9 @@ type DispatchInboundMessageMockParams = { allowProgressCallbacksWhenSourceDeliverySuppressed?: boolean; allowToolLifecycleWhenProgressHidden?: boolean; onReplyStart?: () => void | Promise; - onToolStart?: (payload: { name?: string }) => void | Promise; - onCompactionStart?: () => void | Promise; - onCompactionEnd?: () => void | Promise; + onToolStart?: (payload: { name?: string }) => boolean | void | Promise; + onCompactionStart?: () => boolean | void | Promise; + onCompactionEnd?: () => boolean | void | Promise; }; }; diff --git a/extensions/signal/src/monitor/event-handler.ts b/extensions/signal/src/monitor/event-handler.ts index 8857d9348f9a..8a21c94f0b3f 100644 --- a/extensions/signal/src/monitor/event-handler.ts +++ b/extensions/signal/src/monitor/event-handler.ts @@ -549,13 +549,16 @@ export function createSignalEventHandler(deps: SignalEventHandlerDeps) { if (toolName) { await statusReactionController.setTool(toolName); } + return false; }, onCompactionStart: async () => { await statusReactionController.setCompacting(); + return false; }, onCompactionEnd: async () => { statusReactionController.cancelPending(); await statusReactionController.setThinking(); + return false; }, } : {}), diff --git a/extensions/slack/src/monitor/message-handler/dispatch-progress.ts b/extensions/slack/src/monitor/message-handler/dispatch-progress.ts index 8c31a94b3593..a692116793a4 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch-progress.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch-progress.ts @@ -185,26 +185,27 @@ export function createSlackProgressRuntime(runtimeParams: { ); const markNativeProgressDelivered = (session: SlackStreamSession, threadTs?: string) => { - if (session.delivered) { - delivery.observedReplyDelivery = true; + if (!session.delivered) { + return false; } + delivery.observedReplyDelivery = true; if (threadTs) { delivery.usedReplyThreadTs ??= threadTs; delivery.rememberDeliveredThreadTs("block", threadTs); } + return true; }; const startNativeProgressStream = async ( chunks: NonNullable>, - chunkKey: string, - ) => { + ): Promise => { const streamThreadTs = replyPlan.nextThreadTs(); if (!streamThreadTs) { logVerbose( "slack-stream: no reply thread target for native progress stream start, falling back", ); delivery.streamFailed = true; - return; + return false; } delivery.nativeProgressStreamThreadTs = streamThreadTs; const startPromise = (async () => { @@ -235,26 +236,23 @@ export function createSlackProgressRuntime(runtimeParams: { delivery.nativeProgressStreamStartPromise = null; } } - if (startedSession) { - markNativeProgressDelivered(startedSession, streamThreadTs); - } - nativeProgressChunkKey = chunkKey; - replyPlan.markSent(); + return startedSession ? markNativeProgressDelivered(startedSession, streamThreadTs) : false; }; const appendNativeProgressStream = async ( chunks: NonNullable>, - chunkKey: string, - ) => { + ): Promise => { if (!delivery.streamSession) { - return; + return false; } await appendSlackStream({ session: delivery.streamSession, chunks }); - markNativeProgressDelivered(delivery.streamSession); - nativeProgressChunkKey = chunkKey; + return markNativeProgressDelivered( + delivery.streamSession, + delivery.nativeProgressStreamThreadTs, + ); }; - const updateNativeProgressStream = async () => { + const updateNativeProgressStream = async (): Promise => { const snapshot = progressDraft.getSnapshot(); const progressLines = resolveNativeProgressLines(snapshot); const hasRetirableNativeTasks = [...nativeTaskState.values()].some( @@ -269,11 +267,11 @@ export function createSlackProgressRuntime(runtimeParams: { !explicitProgressTitle && !hasRetirableNativeTasks) ) { - return; + return false; } const canContinue = await waitForNativeProgressStreamStart(); if (!canContinue) { - return; + return false; } const reconciled = reconcileSlackNativeTaskChunks({ previousTasks: nativeTaskState, @@ -281,21 +279,27 @@ export function createSlackProgressRuntime(runtimeParams: { }); const chunks = reconciled.chunks; if (!chunks?.length) { - return; + return false; } const chunkKey = JSON.stringify(chunks); if (chunkKey === nativeProgressChunkKey) { - return; + return Boolean(delivery.streamSession?.delivered); } try { - if (!delivery.streamSession) { - await startNativeProgressStream(chunks, chunkKey); - } else { - await appendNativeProgressStream(chunks, chunkKey); + const accepted = !delivery.streamSession + ? await startNativeProgressStream(chunks) + : await appendNativeProgressStream(chunks); + if (!accepted) { + return false; } - // Commit only after Slack accepted the chunks; a failed emit must retry - // the same reconciliation against the previous snapshot. + // Commit transport identity and task state together. Buffered or failed + // chunks must leave the identical render eligible for another attempt. + if (nativeProgressChunkKey === undefined) { + replyPlan.markSent(); + } + nativeProgressChunkKey = chunkKey; nativeTaskState = reconciled.tasks; + return true; } catch (err) { runtime.error?.( danger( @@ -303,6 +307,7 @@ export function createSlackProgressRuntime(runtimeParams: { ), ); delivery.streamFailed = true; + return false; } }; @@ -335,11 +340,10 @@ export function createSlackProgressRuntime(runtimeParams: { updateOnLineChange: useNativeProgressStreaming || useRichProgressDraft, update: async (previewText, options) => { if (useNativeProgressStreaming) { - await updateNativeProgressStream(); - return; + return await updateNativeProgressStream(); } if (!draftStream) { - return; + return false; } const snapshot = progressDraft.getSnapshot(); const structuredLines = resolveStructuredProgressLines(options?.lines ?? snapshot.lines); @@ -365,6 +369,7 @@ export function createSlackProgressRuntime(runtimeParams: { if (options?.flush) { await draftStream.flush(); } + return Boolean(draftStream.messageId() && draftStream.channelId()); }, }); const commentaryProgressEnabled = progressDraft.commentaryProgressEnabled; @@ -433,11 +438,10 @@ export function createSlackProgressRuntime(runtimeParams: { const pushPlanProgress = async (steps?: AgentPlanStep[], explanation?: string) => { if (streamMode === "status_final") { - await progressDraft.pushPlanProgress(steps, { explanation }); - return; + return await progressDraft.pushPlanProgress(steps, { explanation }); } if (previewToolProgressSuppressed || !draftStream) { - return; + return false; } const text = formatChannelProgressDraftText({ entry: account.config, @@ -451,6 +455,7 @@ export function createSlackProgressRuntime(runtimeParams: { draftStream.update(text); hasStreamedMessage = true; } + return false; }; const pushPreviewProgress = async ( @@ -458,30 +463,28 @@ export function createSlackProgressRuntime(runtimeParams: { options?: { toolName?: string }, ) => { if (!draftStream && !useNativeProgressStreaming) { - return; + return false; } if (options?.toolName !== undefined && !isChannelProgressDraftWorkToolName(options.toolName)) { - return; + return false; } const normalized = line?.text.replace(/\s+/g, " ").trim(); if (streamMode === "status_final") { if (!line || !normalized) { - await progressDraft.noteActivity(); - return; + return await progressDraft.noteActivity(); } - await progressDraft.pushToolProgress(line, options); - return; + return await progressDraft.pushToolProgress(line, options); } if (!line || !normalized || !draftStream || !previewToolProgressEnabled) { - return; + return false; } - await progressDraft.pushToolProgress(line, options); + return await progressDraft.pushToolProgress(line, options); }; const updateDraftFromPartial = (text?: string) => { const trimmed = text?.trimEnd(); if (!trimmed) { - return; + return false; } if (streamMode === "append") { @@ -495,28 +498,29 @@ export function createSlackProgressRuntime(runtimeParams: { appendRenderedText = next.rendered; appendSourceText = next.source; if (!next.changed) { - return; + return false; } draftStream?.update(next.rendered); hasStreamedMessage = true; - return; + return false; } if (streamMode === "status_final") { - return; + return false; } previewToolProgressSuppressed = true; progressDraft.suppress(); draftStream?.update(trimmed); hasStreamedMessage = true; + return false; }; const pushReasoningProgress = async (payload?: { text?: string; isReasoningSnapshot?: boolean; }) => { if (!payload?.text) { - return; + return false; } if (streamMode !== "status_final") { const normalized = progressDraft @@ -526,9 +530,9 @@ export function createSlackProgressRuntime(runtimeParams: { .replace(/^_(.*)_$/su, "$1") .trim(); if (!normalized) { - return; + return false; } - await pushPreviewProgress({ + const visible = await pushPreviewProgress({ id: "reasoning", kind: "item", text: normalized, @@ -536,10 +540,10 @@ export function createSlackProgressRuntime(runtimeParams: { }); // Tool admission closes reasoning bursts; restore this still-open preview lane. progressDraft.mergeReasoningProgress(normalized, { snapshot: true }); - return; + return visible; } progressReceipt.noteReasoning(); - await progressDraft.pushReasoningProgress(payload.text, { + return await progressDraft.pushReasoningProgress(payload.text, { snapshot: payload.isReasoningSnapshot === true, }); }; diff --git a/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts b/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts index 14376664f143..0e0c59300dab 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts @@ -3284,6 +3284,31 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { expectDeliverReplyCall(0, FINAL_REPLY_TEXT); }); + it("retries identical native progress after Slack buffers the first update", async () => { + const session = { + channel: "C123", + threadTs: THREAD_TS, + stopped: false, + delivered: false, + pendingText: "", + }; + startSlackStreamMock.mockResolvedValueOnce(session); + appendSlackStreamMock.mockImplementationOnce(async () => { + session.delivered = true; + }); + + await dispatchNativeProgressScenario({ + events: [ + { kind: "item", itemId: "item-1", progressText: "still working" }, + { kind: "item", itemId: "item-1", progressText: "still working" }, + ], + }); + + expect(startSlackStreamMock).toHaveBeenCalledOnce(); + expect(appendSlackStreamMock).toHaveBeenCalledOnce(); + expect(session.delivered).toBe(true); + }); + it("collapses a native progress stream to a receipt after its fresh final lands", async () => { stopSlackStreamMock.mockResolvedValueOnce({ messageId: "171234.888" }); finalizeSlackPreviewEditMock.mockResolvedValueOnce(undefined); diff --git a/extensions/slack/src/monitor/message-handler/dispatch.ts b/extensions/slack/src/monitor/message-handler/dispatch.ts index 3ac3e8fde402..197e9b78a3a4 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.ts @@ -408,22 +408,29 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag : !previewStreamingEnabled ? undefined : async (payload) => { - progress.updateDraftFromPartial(payload.text); + return progress.updateDraftFromPartial(payload.text); }, - onAssistantMessageStart: progress.onDraftBoundary, + onAssistantMessageStart: progress.onDraftBoundary + ? async () => { + await progress.onDraftBoundary?.(); + return false; + } + : undefined, onReasoningEnd: async () => { progress.progressReceipt.closeReasoning(); await progress.onDraftBoundary?.(); + return false; }, onQueuedFollowupAdmitted: progress.onQueuedFollowupAdmitted, onReasoningStream: statusReactionsEnabled || progress.previewToolProgressEnabled ? async (payload) => { - await progress.pushReasoningProgress(payload); + const visible = await progress.pushReasoningProgress(payload); if (!statusReactionsEnabled) { - return; + return visible; } await statusReactions.setThinking(); + return visible; } : undefined, onToolStart: async (payload) => { @@ -433,16 +440,19 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag if (payload.phase === "start") { progress.progressReceipt.noteToolCall(payload.name); } - await progress.progressDraft.pushToolEvent(payload); + return await progress.progressDraft.pushToolEvent(payload); }, onItemEvent: async (payload) => { if (progress.streamMode === "status_final" && payload.kind === "preamble") { if (progress.shouldYieldDraftProgress()) { - return; + return false; } - await progress.progressDraft.pushPreambleHeadline(payload.progressText, { - itemId: payload.itemId, - }); + const headlineVisible = await progress.progressDraft.pushPreambleHeadline( + payload.progressText, + { + itemId: payload.itemId, + }, + ); if (progress.commentaryProgressEnabled) { const accepted = await progress.progressDraft.pushCommentaryProgress( payload.progressText, @@ -453,25 +463,26 @@ export async function dispatchPreparedSlackMessage(prepared: PreparedSlackMessag if (accepted) { progress.progressReceipt.noteCommentary(payload.itemId, payload.progressText); } + return accepted || headlineVisible; } - return; + return headlineVisible; } - await progress.progressDraft.pushItemEvent(payload); + return await progress.progressDraft.pushItemEvent(payload); }, onPlanUpdate: async (payload) => { if (payload.phase !== "update") { - return; + return false; } - await progress.pushPlanProgress(payload.steps, payload.explanation); + return await progress.pushPlanProgress(payload.steps, payload.explanation); }, onApprovalEvent: async (payload) => { - await progress.progressDraft.pushApprovalEvent(payload); + return await progress.progressDraft.pushApprovalEvent(payload); }, onCommandOutput: async (payload) => { - await progress.progressDraft.pushCommandOutputEvent(payload); + return await progress.progressDraft.pushCommandOutputEvent(payload); }, onPatchSummary: async (payload) => { - await progress.progressDraft.pushPatchEvent(payload); + return await progress.progressDraft.pushPatchEvent(payload); }, }, }); diff --git a/extensions/telegram/src/bot-message-dispatch-progress.ts b/extensions/telegram/src/bot-message-dispatch-progress.ts index 22081d28ab22..a9ef8e5fdba6 100644 --- a/extensions/telegram/src/bot-message-dispatch-progress.ts +++ b/extensions/telegram/src/bot-message-dispatch-progress.ts @@ -235,15 +235,18 @@ export function createTelegramProgressController(params: { if (params.statusReactionController && toolName) { await params.statusReactionController.setTool(toolName); } - await progressPromise; + return await progressPromise; }; const handleItemEvent = async (payload: CallbackPayload<"onItemEvent">) => { if (payload.kind === "preamble") { if (verboseProgressActive()) { - return; + return false; } + let rendered = false; if (params.streamMode === "progress") { - await compositor.pushPreambleHeadline(payload.progressText, { itemId: payload.itemId }); + rendered = await compositor.pushPreambleHeadline(payload.progressText, { + itemId: payload.itemId, + }); } if (params.streamMode === "progress" && compositor.commentaryProgressEnabled) { const accepted = await compositor.pushCommentaryProgress(payload.progressText, { @@ -252,10 +255,11 @@ export function createTelegramProgressController(params: { if (accepted) { summary.noteCommentary(payload.itemId, payload.progressText); } + rendered ||= accepted; } - return; + return rendered; } - await pushToolProgress( + return await pushToolProgress( buildChannelProgressDraftLineForEntry(params.telegramCfg, { event: "item", itemId: payload.itemId, @@ -273,14 +277,15 @@ export function createTelegramProgressController(params: { }; const handlePlanUpdate = async (payload: CallbackPayload<"onPlanUpdate">) => { if (payload.phase === "update" && canPushToolProgress()) { - await compositor.pushPlanProgress(payload.steps, { + return await compositor.pushPlanProgress(payload.steps, { explanation: payload.explanation, }); } + return false; }; const handleApprovalEvent = async (payload: CallbackPayload<"onApprovalEvent">) => { if (payload.phase === "requested") { - await pushToolProgress( + return await pushToolProgress( buildChannelProgressDraftLine({ event: "approval", phase: payload.phase, @@ -291,10 +296,11 @@ export function createTelegramProgressController(params: { }), ); } + return false; }; const handleCommandOutput = async (payload: CallbackPayload<"onCommandOutput">) => { if (payload.phase === "end") { - await pushToolProgress( + return await pushToolProgress( buildChannelProgressDraftLineForEntry(params.telegramCfg, { event: "command-output", itemId: payload.itemId, @@ -307,10 +313,11 @@ export function createTelegramProgressController(params: { }), ); } + return false; }; const handlePatchSummary = async (payload: CallbackPayload<"onPatchSummary">) => { if (payload.phase === "end") { - await pushToolProgress( + return await pushToolProgress( buildChannelProgressDraftLine({ event: "patch", itemId: payload.itemId, @@ -325,6 +332,7 @@ export function createTelegramProgressController(params: { }), ); } + return false; }; return { diff --git a/extensions/telegram/src/bot-message-dispatch-turn.ts b/extensions/telegram/src/bot-message-dispatch-turn.ts index 6b4138128b5e..cf0c2c6de063 100644 --- a/extensions/telegram/src/bot-message-dispatch-turn.ts +++ b/extensions/telegram/src/bot-message-dispatch-turn.ts @@ -154,31 +154,39 @@ export async function runTelegramDispatchTurn(params: { suppressTyping: isRoomEvent, onPartialReply: params.draft.answerLane.stream || params.draft.reasoningLane.stream - ? (payload) => - params.draft.enqueueEvent(async () => { + ? (payload) => { + const queued = params.draft.enqueueEvent(async () => { await params.draft.ingestDraftLaneSegments(payload); - }) + }); + return queued.then(() => false); + } : undefined, onBlockReplyQueued: params.draft.answerLane.stream - ? (payload, blockContext) => - params.draft.enqueueEvent(async () => { + ? (payload, blockContext) => { + const queued = params.draft.enqueueEvent(async () => { await params.draft.prepareQueuedAnswerBlock(payload, blockContext); - }) + }); + return queued.then(() => false); + } : undefined, onReasoningStream: params.draft.reasoningLane.stream - ? (payload) => - params.draft.enqueueEvent(async () => { + ? (payload) => { + const queued = params.draft.enqueueEvent(async () => { if (splitReasoningOnNextStream) { params.draft.repositionLaneForNewMessage(params.draft.reasoningLane); splitReasoningOnNextStream = false; } await params.draft.ingestDraftLaneSegments(payload, true); - }) + }); + return queued.then(() => false); + } : params.draft.streamReasoningInProgressDraft - ? (payload) => - params.draft.enqueueEvent(async () => { + ? (payload) => { + const queued = params.draft.enqueueEvent(async () => { await params.progress.pushReasoningProgress(payload); - }) + }); + return queued.then(() => false); + } : undefined, onReasoningProgress: params.draft.answerLane.stream ? (payload) => @@ -187,8 +195,8 @@ export async function runTelegramDispatchTurn(params: { }) : undefined, onAssistantMessageStart: params.draft.answerLane.stream - ? () => - params.draft.enqueueEvent(async () => { + ? () => { + const queued = params.draft.enqueueEvent(async () => { params.reply.reasoningStepState.resetForNextStep(); params.progress.setFinalAnswerDelivered(false); if (params.streamMode !== "progress") { @@ -203,16 +211,23 @@ export async function runTelegramDispatchTurn(params: { ) { params.draft.setRotateWhenQueuedBlocksSettle(true); } - }) + }); + return queued.then(() => false); + } : undefined, onReasoningEnd: params.draft.reasoningLane.stream - ? () => - params.draft.enqueueEvent(async () => { + ? () => { + const queued = params.draft.enqueueEvent(async () => { params.progress.closeReasoningBurst(); splitReasoningOnNextStream = params.draft.reasoningLane.hasStreamedMessage; params.progress.reset(); - }) - : () => params.progress.closeReasoningBurst(), + }); + return queued.then(() => false); + } + : () => { + params.progress.closeReasoningBurst(); + return false; + }, onQueuedFollowupAdmitted: () => { params.draft.beginQueuedFollowup(); params.progress.beginQueuedFollowup(); @@ -250,30 +265,36 @@ export async function runTelegramDispatchTurn(params: { onToolResult: async (payload) => { const text = payload.text?.trim(); if (!text) { - return; + return false; } const updatedDraft = await params.progress.pushToolProgress(text, { startImmediately: true, }); + if (updatedDraft) { + return true; + } if ( - !updatedDraft && isFastModeAutoProgressPayload(payload) && !params.progress.canPushToolProgress() ) { await params.delivery.sendPayload(payload); + return true; } + return false; }, onCommandOutput: params.progress.handleCommandOutput, onPatchSummary: params.progress.handlePatchSummary, onCompactionStart: params.statusReactionController ? async () => { await params.statusReactionController?.setCompacting(); + return false; } : undefined, onCompactionEnd: params.statusReactionController ? async () => { params.statusReactionController?.cancelPending(); await params.statusReactionController?.setThinking(); + return false; } : undefined, onModelSelected, diff --git a/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts b/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts index 0b359770c2ad..3bd6a15b1c9f 100644 --- a/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts +++ b/extensions/telegram/src/bot-message-dispatch.progress-updates.test.ts @@ -168,9 +168,10 @@ describeTelegramDispatch("dispatchTelegramMessage progress-updates", () => { it("keeps streamed final text in place when late media arrives", async () => { const { answerDraftStream } = setupDraftStreams({ answerMessageId: 2001 }); const mediaMaxBytes = 50 * 1024 * 1024; + let partialAccepted: boolean | void = undefined; dispatchReplyWithBufferedBlockDispatcher.mockImplementation( async ({ dispatcherOptions, replyOptions }) => { - await replyOptions?.onPartialReply?.({ text: "Photo" }); + partialAccepted = await replyOptions?.onPartialReply?.({ text: "Photo" }); await dispatcherOptions.deliver( { text: "Photo", mediaUrl: "https://example.com/a.png" }, { kind: "final" }, @@ -186,6 +187,7 @@ describeTelegramDispatch("dispatchTelegramMessage progress-updates", () => { expect(answerDraftStream.clear).not.toHaveBeenCalled(); expect(answerDraftStream.update).toHaveBeenCalledWith("Photo"); + expect(partialAccepted).toBe(false); expectDeliverRepliesParams({ mediaMaxBytes }); expectDeliveredReply(0, { text: undefined, mediaUrl: "https://example.com/a.png" }); expect(emitTelegramMessageSentHooks).toHaveBeenCalledTimes(1); @@ -699,6 +701,7 @@ describeTelegramDispatch("dispatchTelegramMessage progress-updates", () => { it("keeps string tool-result progress beneath a Telegram preamble", async () => { const draftStream = createSequencedDraftStream(2001); createTelegramDraftStream.mockReturnValue(draftStream); + let rendered: boolean | void = undefined; dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ replyOptions }) => { await replyOptions?.onReplyStart?.(); await replyOptions?.onItemEvent?.({ @@ -706,7 +709,7 @@ describeTelegramDispatch("dispatchTelegramMessage progress-updates", () => { itemId: "preamble-1", progressText: "Checking recent context", }); - await replyOptions?.onToolResult?.({ text: "Background task still running" }); + rendered = await replyOptions?.onToolResult?.({ text: "Background task still running" }); return { queuedFinal: false }; }); @@ -722,9 +725,29 @@ describeTelegramDispatch("dispatchTelegramMessage progress-updates", () => { const preview = draftStream.updatePreview.mock.calls.at(-1)?.[0]; expect(preview?.text).toBe("Shelling\nChecking recent context\nBackground task still running"); expect(JSON.stringify(preview?.richMessage)).toContain("Background task still running"); + expect(rendered).toBe(true); expect(deliverReplies).not.toHaveBeenCalled(); }); + it("reports empty tool-result progress as not rendered", async () => { + const draftStream = createSequencedDraftStream(2001); + createTelegramDraftStream.mockReturnValue(draftStream); + let rendered: boolean | void = undefined; + dispatchReplyWithBufferedBlockDispatcher.mockImplementation(async ({ replyOptions }) => { + rendered = await replyOptions?.onToolResult?.({ text: " " }); + return { queuedFinal: false }; + }); + + await dispatchWithContext({ + context: createContext(), + streamMode: "progress", + telegramCfg: { streaming: { mode: "progress", progress: { label: "Shelling" } } }, + }); + + expect(rendered).toBe(false); + expect(draftStream.updatePreview).not.toHaveBeenCalled(); + }); + it("retracts the Telegram preamble headline by item identity", async () => { const draftStream = createSequencedDraftStream(2001); createTelegramDraftStream.mockReturnValue(draftStream); diff --git a/extensions/whatsapp/src/auto-reply/monitor/inbound-dispatch.ts b/extensions/whatsapp/src/auto-reply/monitor/inbound-dispatch.ts index 657c77cd7604..02fb65f12bd8 100644 --- a/extensions/whatsapp/src/auto-reply/monitor/inbound-dispatch.ts +++ b/extensions/whatsapp/src/auto-reply/monitor/inbound-dispatch.ts @@ -873,13 +873,16 @@ export function createWhatsAppReplyPlan(params: { if (toolName) { await statusReactionController.setTool(toolName); } + return false; }, onCompactionStart: async () => { await statusReactionController.setCompacting(); + return false; }, onCompactionEnd: async () => { statusReactionController.cancelPending(); await statusReactionController.setThinking(); + return false; }, } : {}), diff --git a/src/agents/embedded-agent-runner/run/params.ts b/src/agents/embedded-agent-runner/run/params.ts index bacac7264edb..47e2a0fc1b88 100644 --- a/src/agents/embedded-agent-runner/run/params.ts +++ b/src/agents/embedded-agent-runner/run/params.ts @@ -311,7 +311,7 @@ export type RunEmbeddedAgentParams = { replyOperation?: ReplyOperation; shouldEmitToolResult?: () => boolean; shouldEmitToolOutput?: () => boolean; - onPartialReply?: (payload: PartialReplyPayload) => void | Promise; + onPartialReply?: (payload: PartialReplyPayload) => boolean | void | Promise; onAssistantMessageStart?: () => void | Promise; onBlockReply?: (payload: BlockReplyPayload, context?: BlockReplyContext) => void | Promise; onBlockReplyFlush?: (context: BlockReplyFlushContext) => void | Promise; diff --git a/src/agents/embedded-agent-subscribe.callback.ts b/src/agents/embedded-agent-subscribe.callback.ts index a34bb0fc1c9d..79b79b747338 100644 --- a/src/agents/embedded-agent-subscribe.callback.ts +++ b/src/agents/embedded-agent-subscribe.callback.ts @@ -6,13 +6,13 @@ type CallbackLogger = { /** Contains failures from untracked subscriber presentation and telemetry callbacks. */ export function runBestEffortCallback(params: { - callback: () => void | Promise; + callback: () => unknown; label: string; log: CallbackLogger; }): void { try { const result = params.callback(); - if (isPromiseLike(result)) { + if (isPromiseLike(result)) { void Promise.resolve(result).catch((error: unknown) => { params.log.warn(`${params.label} callback failed: ${String(error)}`); }); diff --git a/src/agents/embedded-agent-subscribe.types.ts b/src/agents/embedded-agent-subscribe.types.ts index 430276f39d1b..b289430666dd 100644 --- a/src/agents/embedded-agent-subscribe.types.ts +++ b/src/agents/embedded-agent-subscribe.types.ts @@ -66,7 +66,7 @@ export type SubscribeEmbeddedAgentSessionParams = { onBlockReplyFlush?: (context: BlockReplyFlushContext) => void | Promise; blockReplyBreak?: "text_end" | "message_end"; blockReplyChunking?: BlockReplyChunking; - onPartialReply?: (payload: PartialReplyPayload) => void | Promise; + onPartialReply?: (payload: PartialReplyPayload) => boolean | void | Promise; onAssistantMessageStart?: () => void | Promise; onExecutionPhase?: (info: { phase: "tool_execution_started"; diff --git a/src/auto-reply/get-reply-options.types.ts b/src/auto-reply/get-reply-options.types.ts index 3f539ac09237..2659e559a509 100644 --- a/src/auto-reply/get-reply-options.types.ts +++ b/src/auto-reply/get-reply-options.types.ts @@ -103,8 +103,8 @@ type ReasoningProgressPayload = { progressTokens: number; }; -/** Return false when a channel intentionally keeps a progress event out of user-visible UI. */ -type ProgressCallbackResult = false | void; +/** Return false until the channel has accepted operator-visible progress. */ +type ProgressCallbackResult = boolean | void; /** Reply generation options shared by auto-reply, webchat, channels, and tests. */ export type GetReplyOptions = { @@ -180,20 +180,29 @@ export type GetReplyOptions = { onVerboseProgressVisibility?: (isActive: () => boolean) => void; /** Preserve source-event callback start order for stateful channel progress renderers. */ preserveProgressCallbackStartOrder?: boolean; - onPartialReply?: (payload: PartialReplyPayload) => Promise | void; - onReasoningStream?: (payload: ReasoningStreamPayload) => Promise | void; + onPartialReply?: ( + payload: PartialReplyPayload, + ) => Promise | ProgressCallbackResult; + onReasoningStream?: ( + payload: ReasoningStreamPayload, + ) => Promise | ProgressCallbackResult; onReasoningProgress?: (payload: ReasoningProgressPayload) => Promise | void; streamReasoningInNonStreamModes?: boolean; /** Called when a thinking/reasoning block ends. */ - onReasoningEnd?: () => Promise | void; + onReasoningEnd?: () => Promise | ProgressCallbackResult; /** Called when a new assistant message starts (e.g., after tool call or thinking block). */ - onAssistantMessageStart?: () => Promise | void; + onAssistantMessageStart?: () => Promise | ProgressCallbackResult; /** Called synchronously when a block reply is logically emitted, before async * delivery drains. Useful for channels that need to rotate preview state at * block boundaries without waiting for transport acks. */ - onBlockReplyQueued?: (payload: ReplyPayload, context?: BlockReplyContext) => Promise | void; + onBlockReplyQueued?: ( + payload: ReplyPayload, + context?: BlockReplyContext, + ) => Promise | ProgressCallbackResult; onBlockReply?: (payload: ReplyPayload, context?: BlockReplyContext) => Promise | void; - onToolResult?: (payload: ReplyPayload) => Promise | void; + onToolResult?: ( + payload: ReplyPayload, + ) => Promise | ProgressCallbackResult; /** Called when a tool phase starts/updates, before summary payloads are emitted. */ onToolStart?: (payload: { itemId?: string; @@ -202,7 +211,7 @@ export type GetReplyOptions = { phase?: string; args?: Record; detailMode?: "explain" | "raw"; - }) => Promise | void; + }) => Promise | ProgressCallbackResult; /** Called when a concrete work item starts, updates, or completes. */ onItemEvent?: (payload: { itemId?: string; @@ -256,7 +265,7 @@ export type GetReplyOptions = { explanation?: string; steps?: AgentPlanStep[]; source?: string; - }) => Promise | void; + }) => Promise | ProgressCallbackResult; /** Called when an approval becomes pending or resolves. */ onApprovalEvent?: (payload: { phase?: string; @@ -272,7 +281,7 @@ export type GetReplyOptions = { reason?: string; scope?: "turn" | "session"; message?: string; - }) => Promise | void; + }) => Promise | ProgressCallbackResult; /** Called when command output streams or completes. */ onCommandOutput?: (payload: { itemId?: string; @@ -297,11 +306,11 @@ export type GetReplyOptions = { modified?: string[]; deleted?: string[]; summary?: string; - }) => Promise | void; + }) => Promise | ProgressCallbackResult; /** Called when context auto-compaction starts (allows UX feedback during the pause). */ - onCompactionStart?: () => Promise | void; + onCompactionStart?: () => Promise | ProgressCallbackResult; /** Called when context auto-compaction completes. */ - onCompactionEnd?: () => Promise | void; + onCompactionEnd?: () => Promise | ProgressCallbackResult; /** Called when the actual model is selected (including after fallback). * Use this to get model/provider/thinkLevel for responsePrefix template interpolation. */ onModelSelected?: (ctx: ModelSelectedContext) => void; diff --git a/src/auto-reply/reply/agent-event-bridge.ts b/src/auto-reply/reply/agent-event-bridge.ts index 7b7db079ec6b..4b9f424d8415 100644 --- a/src/auto-reply/reply/agent-event-bridge.ts +++ b/src/auto-reply/reply/agent-event-bridge.ts @@ -3,7 +3,7 @@ import { type AgentEventPayload, onAgentEvent } from "../../infra/agent-events.js"; export type AgentEventDeliveryStartOrder = { - schedule: (deliver: () => Promise) => Promise; + schedule: (deliver: () => Promise) => Promise; }; export function createAgentEventDeliveryStartOrder(): AgentEventDeliveryStartOrder { @@ -18,7 +18,7 @@ export function createAgentEventDeliveryStartOrder(): AgentEventDeliveryStartOrd releaseStart = resolve; }); await previousStart; - let delivery: Promise; + let delivery: Promise; try { delivery = deliver(); } finally { @@ -33,7 +33,7 @@ export function createAgentEventBridge(params: { runId: string; suppressed?: boolean; read: (evt: AgentEventPayload) => T | undefined; - deliver?: (payload: T) => Promise; + deliver?: (payload: T) => Promise; startOrder?: AgentEventDeliveryStartOrder; }) { const deliver = params.deliver; @@ -44,7 +44,7 @@ export function createAgentEventBridge(params: { }; } let unsubscribed = false; - let delivery = Promise.resolve(); + let delivery: Promise = Promise.resolve(); const rawUnsubscribe = onAgentEvent((evt) => { if (evt.runId !== params.runId) { return; diff --git a/src/auto-reply/reply/agent-runner-cli-candidate.ts b/src/auto-reply/reply/agent-runner-cli-candidate.ts index bfee72fc2b39..002739108ef7 100644 --- a/src/auto-reply/reply/agent-runner-cli-candidate.ts +++ b/src/auto-reply/reply/agent-runner-cli-candidate.ts @@ -265,22 +265,26 @@ export async function runCliFallbackCandidate(params: { } const textForTyping = classified.text; const sanitized = params.presentation.sanitizeStreamingText(textForTyping, false); + const onPartialReply = turn.opts?.onPartialReply; if (!params.preserveProgressCallbackStartOrder) { await turn.typingSignals.signalTextDelta(textForTyping); - if (sanitized.skip || !sanitized.text || !turn.opts?.onPartialReply) { - return; + if (sanitized.skip || !sanitized.text || !onPartialReply) { + return false; } - await turn.opts.onPartialReply({ text: sanitized.text }); - return; + return await onPartialReply({ text: sanitized.text }); } if (sanitized.skip || !sanitized.text) { await turn.typingSignals.signalTextDelta(textForTyping); - return; + return false; + } + if (!onPartialReply) { + await turn.typingSignals.signalTextDelta(textForTyping); + return false; } // Assistant and tool CLI bridges drain independently. Stage presentation first. - await params.presentation.startPresentationWhileTyping( + return await params.presentation.startPresentationWhileTyping( turn.typingSignals.signalTextDelta(textForTyping), - () => turn.opts?.onPartialReply?.({ text: sanitized.text }), + () => onPartialReply({ text: sanitized.text }), ); }, onReasoningText: createCliReasoningStreamBridge(turn.opts?.onReasoningStream), @@ -320,14 +324,15 @@ export async function runCliFallbackCandidate(params: { summaryPromise, params.presentation.startPresentationWhileTyping( turn.typingSignals.signalToolStart(), - () => - turn.opts?.onToolStart?.({ + async () => { + await turn.opts?.onToolStart?.({ ...(toolCallId ? { toolCallId } : {}), name, phase, args, detailMode: turn.toolProgressDetail, - }), + }); + }, ), ]); }, diff --git a/src/auto-reply/reply/agent-runner-cli-dispatch.ts b/src/auto-reply/reply/agent-runner-cli-dispatch.ts index 3c09b72506cc..99fa6707dbfd 100644 --- a/src/auto-reply/reply/agent-runner-cli-dispatch.ts +++ b/src/auto-reply/reply/agent-runner-cli-dispatch.ts @@ -50,7 +50,7 @@ async function stopAgentEventBridges(bridges: readonly AgentEventBridge[]): Prom function createAssistantTextBridge(params: { runId: string; suppressed?: boolean; - deliver?: (text: string) => Promise; + deliver?: (text: string) => Promise; startOrder?: AgentEventDeliveryStartOrder; }) { let lastText: string | undefined; @@ -431,7 +431,7 @@ type RunCliAgentWithLifecycleParams = { */ onActivity?: () => void; preserveProgressCallbackStartOrder?: boolean; - onAssistantText?: (text: string) => Promise; + onAssistantText?: (text: string) => Promise; onReasoningText?: (payload: ReasoningTextPayload) => Promise; onReasoningProgress?: (payload: ReasoningProgressPayload) => Promise; onToolEvent?: (payload: CliToolEventPayload) => Promise; diff --git a/src/auto-reply/reply/agent-runner-embedded-candidate.ts b/src/auto-reply/reply/agent-runner-embedded-candidate.ts index c68bab93a941..f06b33fe1403 100644 --- a/src/auto-reply/reply/agent-runner-embedded-candidate.ts +++ b/src/auto-reply/reply/agent-runner-embedded-candidate.ts @@ -278,7 +278,7 @@ export async function runEmbeddedFallbackCandidate(params: { onPartialReply: async (payload) => { const classified = params.presentation.classifyStreamingPartial(payload); if (classified.skip || !classified.text) { - return; + return false; } const textForTyping = classified.text; let didMaterialize = false; @@ -294,17 +294,21 @@ export async function runEmbeddedFallbackCandidate(params: { }, mediaUrls: payload.mediaUrls, }; + const onPartialReply = turn.opts?.onPartialReply; if (!params.preserveProgressCallbackStartOrder) { await turn.typingSignals.signalTextDelta(textForTyping); - if (!turn.opts?.onPartialReply) { - return; + if (!onPartialReply) { + return false; } - await turn.opts.onPartialReply(partialPayload); - return; + return await onPartialReply(partialPayload); } - await params.presentation.startPresentationWhileTyping( + if (!onPartialReply) { + await turn.typingSignals.signalTextDelta(textForTyping); + return false; + } + return await params.presentation.startPresentationWhileTyping( turn.typingSignals.signalTextDelta(textForTyping), - () => turn.opts?.onPartialReply?.(partialPayload), + () => onPartialReply(partialPayload), ); }, onAssistantMessageStart: async () => { @@ -315,7 +319,9 @@ export async function runEmbeddedFallbackCandidate(params: { } await params.presentation.startPresentationWhileTyping( turn.typingSignals.signalMessageStart(), - () => turn.opts?.onAssistantMessageStart?.(), + async () => { + await turn.opts?.onAssistantMessageStart?.(); + }, ); }, onReasoningStream: @@ -336,18 +342,23 @@ export async function runEmbeddedFallbackCandidate(params: { } await params.presentation.startPresentationWhileTyping( turn.typingSignals.signalReasoningDelta(), - () => - turn.opts?.onReasoningStream?.({ + async () => { + await turn.opts?.onReasoningStream?.({ text: payload.text, mediaUrls: payload.mediaUrls, isReasoningSnapshot: payload.isReasoningSnapshot, requiresReasoningProgressOptIn: payload.requiresReasoningProgressOptIn, - }), + }); + }, ); } : undefined, streamReasoningInNonStreamModes: turn.opts?.streamReasoningInNonStreamModes, - onReasoningEnd: turn.opts?.onReasoningEnd, + onReasoningEnd: turn.opts?.onReasoningEnd + ? async () => { + await turn.opts?.onReasoningEnd?.(); + } + : undefined, onAgentEvent: createAgentRunEventHandler({ turn, lifecycleBackstop, diff --git a/src/auto-reply/reply/agent-runner-presentation.ts b/src/auto-reply/reply/agent-runner-presentation.ts index eaf55c44df91..09caf726cb9b 100644 --- a/src/auto-reply/reply/agent-runner-presentation.ts +++ b/src/auto-reply/reply/agent-runner-presentation.ts @@ -24,8 +24,8 @@ type AgentTurnPresentation = { normalizeStreamingText: (payload: ReplyPayload) => { text?: string; skip: boolean }; startPresentationWhileTyping: ( typingPromise: Promise, - startPresentation: () => void | Promise, - ) => Promise; + startPresentation: () => boolean | void | Promise, + ) => Promise; blockReplyHandler: ReturnType | undefined; }; @@ -93,9 +93,9 @@ export function createAgentTurnPresentation(params: { const startPresentationWhileTyping = async ( typingPromise: Promise, - startPresentation: () => void | Promise, + startPresentation: () => boolean | void | Promise, ) => { - let presentationPromise: void | Promise; + let presentationPromise: boolean | void | Promise; try { presentationPromise = startPresentation(); } catch (err) { @@ -103,7 +103,8 @@ export function createAgentTurnPresentation(params: { void typingPromise.catch(() => undefined); throw err; } - await Promise.all([typingPromise, presentationPromise]); + const [, result] = await Promise.all([typingPromise, presentationPromise]); + return result; }; const blockReplyPipeline = params.turn.blockReplyPipeline; diff --git a/src/auto-reply/reply/agent-runner-run.ts b/src/auto-reply/reply/agent-runner-run.ts index 1c81d04f8d54..0f7226701612 100644 --- a/src/auto-reply/reply/agent-runner-run.ts +++ b/src/auto-reply/reply/agent-runner-run.ts @@ -5,6 +5,7 @@ import { formatEmbeddedAgentQueueFailureSummary, queueEmbeddedAgentMessageWithOutcomeAsync, } from "../../agents/embedded-agent-runner/runs.js"; +import { settleProgressVisibilityCallbackResult } from "../../channels/progress-visibility.js"; import { hasRestartRecoverySourceClaim } from "../../config/sessions/restart-recovery-state.js"; import { loadSessionEntry, updateSessionEntry } from "../../config/sessions/session-accessor.js"; import { logVerbose } from "../../globals.js"; @@ -113,14 +114,16 @@ export async function runReplyAgent( const isHeartbeat = opts?.isHeartbeat === true; let didDeliverVisiblePartialReply = false; - const runOpts = opts?.onPartialReply + const onPartialReply = opts?.onPartialReply; + const runOpts = onPartialReply ? { ...opts, onPartialReply: async (payload: Parameters>[0]) => { - await opts.onPartialReply?.(payload); - if (hasOutboundReplyContent(payload, { trimText: true })) { + const observed = await settleProgressVisibilityCallbackResult(onPartialReply(payload)); + if (observed.visible && hasOutboundReplyContent(payload, { trimText: true })) { didDeliverVisiblePartialReply = true; } + return observed.result; }, } : opts; diff --git a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts index c882bfdfa1f3..55f5b0d1fa7a 100644 --- a/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts +++ b/src/auto-reply/reply/agent-runner.runreplyagent.e2e.test.ts @@ -50,7 +50,7 @@ import { createMockTypingController } from "./test-helpers.js"; type AgentRunParams = { sessionId?: string; sessionFile?: string; - onPartialReply?: (payload: { text?: string }) => Promise | void; + onPartialReply?: (payload: { text?: string }) => Promise | boolean | void; onAssistantMessageStart?: () => Promise | void; onReasoningStream?: (payload: { text?: string }) => Promise | void; onBlockReply?: (payload: { @@ -1223,24 +1223,28 @@ describe("runReplyAgent heartbeat followup guard", () => { }); it.each([ - { label: "direct chat", sessionCtx: {} }, + { label: "legacy void in a direct chat", callbackResult: undefined, sessionCtx: {} }, + { label: "explicit acceptance in a direct chat", callbackResult: true, sessionCtx: {} }, + { label: "explicit rejection in a direct chat", callbackResult: false, sessionCtx: {} }, { - label: "group chat", + label: "legacy void in a group chat", + callbackResult: undefined, sessionCtx: { ChatType: "group" as const, SessionKey: "agent:test:telegram:group:-100123", }, }, ])( - "returns a terminal failure in a $label after a delivered partial with block streaming disabled", - async ({ sessionCtx }) => { + "preserves $label through terminal failure with block streaming disabled", + async ({ callbackResult, sessionCtx }) => { const accounting = await import("./session-run-accounting.js"); const persistSpy = vi .spyOn(accounting, "persistRunSessionUsage") .mockRejectedValueOnce(new Error("persist exploded")); - const onPartialReply = vi.fn(); + const onPartialReply = vi.fn(async () => callbackResult); + let observedCallbackResult: boolean | void = undefined; state.runEmbeddedAgentMock.mockImplementationOnce(async (params: AgentRunParams) => { - await params.onPartialReply?.({ text: "partial answer" }); + observedCallbackResult = await params.onPartialReply?.({ text: "partial answer" }); return { payloads: [{ text: "final answer" }], meta: { agentMeta: { usage: { input: 1, output: 1 } } }, @@ -1250,20 +1254,24 @@ describe("runReplyAgent heartbeat followup guard", () => { try { const { run } = createMinimalRun({ blockStreamingEnabled: false, - opts: { onPartialReply }, + opts: { onPartialReply, preserveProgressCallbackStartOrder: true }, sessionCtx, }); - const result = await run(); - const payload = Array.isArray(result) ? result[0] : result; - + if (callbackResult === false) { + await expect(run()).rejects.toThrow("persist exploded"); + } else { + const result = await run(); + const payload = Array.isArray(result) ? result[0] : result; + expect(payload).toMatchObject({ + text: GENERIC_EXTERNAL_RUN_FAILURE_TEXT, + isError: true, + }); + } expect(onPartialReply).toHaveBeenCalledWith({ text: "partial answer", mediaUrls: undefined, }); - expect(payload).toMatchObject({ - text: GENERIC_EXTERNAL_RUN_FAILURE_TEXT, - isError: true, - }); + expect(observedCallbackResult).toBe(callbackResult); } finally { persistSpy.mockRestore(); } diff --git a/src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts b/src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts index 152e6dd49e7e..b1a12537dfd0 100644 --- a/src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts +++ b/src/auto-reply/reply/dispatch-from-config.delivery-and-tts.test-utils.ts @@ -1641,7 +1641,9 @@ describe("dispatchReplyFromConfig", () => { toolProgressPromise = Promise.resolve(opts?.onToolStart?.({ name: "lookup" })).then(() => { toolProgressSettled = true; }); - partialProgressPromise = Promise.resolve(opts?.onPartialReply?.({ text: "after tool" })); + partialProgressPromise = Promise.resolve(opts?.onPartialReply?.({ text: "after tool" })).then( + () => undefined, + ); return { text: "final" }; }; diff --git a/src/auto-reply/reply/dispatch-from-config.prepare-execution.ts b/src/auto-reply/reply/dispatch-from-config.prepare-execution.ts index 087e0af0f7be..373260b5ee42 100644 --- a/src/auto-reply/reply/dispatch-from-config.prepare-execution.ts +++ b/src/auto-reply/reply/dispatch-from-config.prepare-execution.ts @@ -265,7 +265,7 @@ export async function prepareDispatchExecution(state: ChooseDispatchRouteReadySt releaseStart: () => releaseStart?.(), }; }; - const wrapProgressCallback = ( + const wrapProgressCallback = ( callback: ((...args: Args) => Promise | Result) | undefined, options?: { allowWhenToolSummariesHidden?: boolean; diff --git a/src/auto-reply/reply/dispatch-from-config.progress.test-utils.ts b/src/auto-reply/reply/dispatch-from-config.progress.test-utils.ts index 50aaf15be7d7..f9b3a4bd268a 100644 --- a/src/auto-reply/reply/dispatch-from-config.progress.test-utils.ts +++ b/src/auto-reply/reply/dispatch-from-config.progress.test-utils.ts @@ -1358,8 +1358,8 @@ describe("dispatchReplyFromConfig", () => { SessionKey: "agent:main:discord:direct:U1", }); let receivedOptions: GetReplyOptions | undefined; - let commandOutputResult: false | void = undefined; - let itemEventResult: false | void = undefined; + let commandOutputResult: boolean | void = undefined; + let itemEventResult: boolean | void = undefined; const replyResolver = vi.fn(async (_ctx: MsgContext, opts?: GetReplyOptions) => { receivedOptions = opts; commandOutputResult = await opts?.onCommandOutput?.({ diff --git a/src/auto-reply/reply/followup-turn-execution.test.ts b/src/auto-reply/reply/followup-turn-execution.test.ts index be2f22134810..072ad7823bf1 100644 --- a/src/auto-reply/reply/followup-turn-execution.test.ts +++ b/src/auto-reply/reply/followup-turn-execution.test.ts @@ -432,6 +432,34 @@ describe("executeFollowupTurn", () => { expect(onPlanUpdate).toHaveBeenCalledWith({ title: "quiet plan" }); }); + it.each([ + { label: "sync void", callback: () => undefined, expected: true }, + { label: "async void", callback: async () => undefined, expected: true }, + { label: "explicit true", callback: () => true, expected: true }, + { label: "explicit false", callback: () => false, expected: false }, + ])("classifies $label followup progress", async ({ callback, expected }) => { + let observed: boolean | void = undefined; + state.execute.mockImplementation(async (params: AgentTurnParams) => { + observed = await params.opts?.onPlanUpdate?.({ title: "queued plan" }); + return { runId: "run-1", outcome: { kind: "rejected", payload: { text: "done" } } }; + }); + + const result = await executeFollowupTurn({ + turn: createTurn(), + defaults: { + typing: createTypingController(), + typingMode: "never", + defaultModel: "claude", + opts: { onPlanUpdate: callback }, + }, + onToolResult: vi.fn(async () => {}), + onCompactionNoticePayload: vi.fn(async () => {}), + }); + await result.progress.drain(); + + expect(observed).toBe(expected); + }); + it("tracks a visible failed item before suppressing duplicate default warnings", async () => { const onItemEvent = vi.fn(async () => undefined); let warningSuppressed: boolean | undefined; @@ -623,7 +651,7 @@ describe("executeFollowupTurn", () => { onCompactionNoticePayload: vi.fn(async () => {}), }); - await expect(detachedProgress).resolves.toBeUndefined(); + await expect(detachedProgress).resolves.toBe(false); await expect(result.progress.drain()).rejects.toBe(failure); }); diff --git a/src/auto-reply/reply/followup-turn-execution.ts b/src/auto-reply/reply/followup-turn-execution.ts index db473db9173c..65ebcb908b01 100644 --- a/src/auto-reply/reply/followup-turn-execution.ts +++ b/src/auto-reply/reply/followup-turn-execution.ts @@ -1,3 +1,4 @@ +import { settleProgressVisibilityCallbackResult } from "../../channels/progress-visibility.js"; import { loadSessionEntryReadOnly } from "../../config/sessions/session-accessor.js"; import { formatErrorMessage } from "../../infra/errors.js"; import type { TemplateContext } from "../templating.js"; @@ -125,6 +126,17 @@ export async function executeFollowupTurn(params: { pendingProgressTasks.add(trackedTask); return progressChain; }; + const enqueueProgressResult = async ( + deliver: () => Promise | boolean | void, + ): Promise => { + let completed = false; + let result: boolean | void = false; + await enqueueProgress(async () => { + result = await deliver(); + completed = true; + }); + return completed ? result : false; + }; const wrap = (callback: ((value: T) => unknown) | undefined, allowed = progressAllowed) => callback ? (value: T) => @@ -134,6 +146,19 @@ export async function executeFollowupTurn(params: { } }) : undefined; + const wrapVisibility = ( + callback: ((value: T) => Promise | boolean | void) | undefined, + allowed = progressAllowed, + ) => + callback + ? (value: T) => + enqueueProgressResult(async () => { + if (!allowed()) { + return false; + } + return (await settleProgressVisibilityCallbackResult(callback(value))).visible; + }) + : undefined; const baseTypingSignals = createTypingSignaler({ typing: defaults.typing, mode: progressAllowed() ? defaults.typingMode : "never", @@ -161,14 +186,16 @@ export async function executeFollowupTurn(params: { onBlockReply: undefined, onPartialReply: undefined, onAssistantMessageStart: undefined, - onToolStart: wrap(sourceOpts?.onToolStart, shouldEmitToolLifecycle), + onToolStart: wrapVisibility(sourceOpts?.onToolStart, shouldEmitToolLifecycle), onCommandOutput: sourceOpts?.onCommandOutput ? (output) => - enqueueProgress(async () => { + enqueueProgressResult(async () => { if (!shouldEmitToolResult()) { - return; + return false; } - const visible = (await sourceOpts.onCommandOutput?.(output)) !== false; + const visible = ( + await settleProgressVisibilityCallbackResult(sourceOpts.onCommandOutput!(output)) + ).visible; if ( visible && (output.status === "failed" || @@ -177,39 +204,58 @@ export async function executeFollowupTurn(params: { ) { visibleToolError = true; } + return visible; }) : undefined, onItemEvent: sourceOpts?.onItemEvent ? (item) => - enqueueProgress(async () => { + enqueueProgressResult(async () => { if (!shouldEmitToolResult()) { - return; + return false; } - const visible = (await sourceOpts.onItemEvent?.(item)) !== false; + const visible = ( + await settleProgressVisibilityCallbackResult(sourceOpts.onItemEvent!(item)) + ).visible; if ( visible && (item.phase === "error" || item.status === "failed" || item.status === "error") ) { visibleToolError = true; } + return visible; }) : undefined, onNarrationUpdate: wrap(sourceOpts?.onNarrationUpdate), - onPlanUpdate: wrap(sourceOpts?.onPlanUpdate), - onApprovalEvent: wrap(sourceOpts?.onApprovalEvent, shouldEmitToolResult), - onPatchSummary: wrap(sourceOpts?.onPatchSummary, shouldEmitToolResult), + onPlanUpdate: wrapVisibility(sourceOpts?.onPlanUpdate), + onApprovalEvent: wrapVisibility(sourceOpts?.onApprovalEvent, shouldEmitToolResult), + onPatchSummary: wrapVisibility(sourceOpts?.onPatchSummary, shouldEmitToolResult), onCompactionStart: sourceOpts?.onCompactionStart ? () => - enqueueProgress(() => (progressAllowed() ? sourceOpts.onCompactionStart?.() : undefined)) + enqueueProgressResult(async () => + progressAllowed() + ? (await settleProgressVisibilityCallbackResult(sourceOpts.onCompactionStart!())) + .visible + : false, + ) : undefined, onCompactionEnd: sourceOpts?.onCompactionEnd ? () => - enqueueProgress(() => (progressAllowed() ? sourceOpts.onCompactionEnd?.() : undefined)) + enqueueProgressResult(async () => + progressAllowed() + ? (await settleProgressVisibilityCallbackResult(sourceOpts.onCompactionEnd!())) + .visible + : false, + ) : undefined, - onReasoningStream: wrap(sourceOpts?.onReasoningStream), + onReasoningStream: wrapVisibility(sourceOpts?.onReasoningStream), onReasoningProgress: wrap(sourceOpts?.onReasoningProgress), onReasoningEnd: sourceOpts?.onReasoningEnd - ? () => enqueueProgress(() => (progressAllowed() ? sourceOpts.onReasoningEnd?.() : undefined)) + ? () => + enqueueProgressResult(async () => + progressAllowed() + ? (await settleProgressVisibilityCallbackResult(sourceOpts.onReasoningEnd!())).visible + : false, + ) : undefined, shouldSuppressToolErrorWarnings: () => { const explicit = sourceOpts?.suppressToolErrorWarnings; @@ -225,9 +271,9 @@ export async function executeFollowupTurn(params: { return undefined; }, onToolResult: async (payload) => { - await enqueueProgress(async () => { + return await enqueueProgressResult(async () => { if (!progressAllowed()) { - return; + return false; } const verboseToolResult = shouldEmitVerboseToolResult(); const toolResultProgressVisible = Boolean(channelToolResultProgress) || verboseToolResult; @@ -235,16 +281,17 @@ export async function executeFollowupTurn(params: { turn.queued.run.sourceReplyDeliveryMode === "message_tool_only" && !toolResultProgressVisible ) { - return; + return false; } - if (channelToolResultProgress && !verboseToolResult) { - await channelToolResultProgress(payload); - } else { - await params.onToolResult(payload, { runId: turn.runId }); - } - if (payload.isError === true) { + const visible = + channelToolResultProgress && !verboseToolResult + ? (await settleProgressVisibilityCallbackResult(channelToolResultProgress(payload))) + .visible + : await params.onToolResult(payload, { runId: turn.runId }).then(() => true); + if (visible && payload.isError === true) { visibleToolError = true; } + return visible; }); }, }; diff --git a/src/channels/progress-draft-compositor.ts b/src/channels/progress-draft-compositor.ts index 26606e18e08a..e305452c453d 100644 --- a/src/channels/progress-draft-compositor.ts +++ b/src/channels/progress-draft-compositor.ts @@ -12,6 +12,7 @@ import { normalizeReasoningProgressLine, sanitizeProgressStatusText, } from "./progress-draft-status-text.js"; +import { settleProgressVisibilityCallbackResult } from "./progress-visibility.js"; import { createChannelProgressDraftGate, type AgentPlanStep, @@ -58,7 +59,10 @@ export function createChannelProgressDraftCompositor(params: { mode: ChannelProgressDraftMode; active: boolean; seed: string; - update: (text: string, options?: ChannelProgressDraftUpdateOptions) => Promise | void; + update: ( + text: string, + options?: ChannelProgressDraftUpdateOptions, + ) => Promise | boolean | void; deleteCurrent?: () => Promise | void; tryNativeUpdate?: (text: string) => Promise | boolean; /** Publish when structured lines change even if the rendered text does not. */ @@ -126,6 +130,7 @@ export function createChannelProgressDraftCompositor(params: { let finalReplyStarted = false; let finalReplyDelivered = false; let preambleExpiryTimer: ReturnType | undefined; + let lastStartRendered = false; const mergeReasoningProgress = (text?: string, options?: { snapshot?: boolean }): string => { if (!text) { @@ -195,6 +200,7 @@ export function createChannelProgressDraftCompositor(params: { narrationText = ""; planSteps = undefined; planExplanation = ""; + lastStartRendered = false; }; const publish = async (options?: { flush?: boolean }): Promise => { @@ -203,9 +209,15 @@ export function createChannelProgressDraftCompositor(params: { if (!text || (text === lastRenderedText && !linesChanged)) { return false; } + const observed = await settleProgressVisibilityCallbackResult( + params.update(text, { ...options, lines: [...lines] }), + ); + if (!observed.visible) { + return false; + } + // Only accepted renders become the dedupe baseline; pending sends remain retryable. lastRenderedText = text; lastRenderedLines = lines; - await params.update(text, { ...options, lines: [...lines] }); return true; }; @@ -242,7 +254,7 @@ export function createChannelProgressDraftCompositor(params: { const gate = createChannelProgressDraftGate({ onStart: async () => { - await render({ flush: true }); + lastStartRendered = await render({ flush: true }); schedulePreambleExpiryRefresh(); }, setTimeoutFn, @@ -323,13 +335,15 @@ export function createChannelProgressDraftCompositor(params: { maxLines: resolveChannelProgressDraftMaxLines(params.entry), }) : lines; - if (shouldStoreLine && nextLines === lines) { + const lineChanged = nextLines !== lines; + const hasUnconfirmedRender = formatDraftText(nextLines) !== lastRenderedText; + if (shouldStoreLine && !lineChanged && !hasUnconfirmedRender) { return false; } // A work line lands between reasoning bursts: commit the current thinking // line so the next thought appends as its own line, interleaved with tools // in arrival order, instead of replacing the prior thought. - if (shouldStoreLine) { + if (shouldStoreLine && lineChanged) { reasoningRawText = ""; lastReasoningLine = undefined; } @@ -350,11 +364,14 @@ export function createChannelProgressDraftCompositor(params: { } if (options?.startImmediately || params.shouldStartNow?.(line)) { const alreadyStarted = gate.hasStarted; + if (!alreadyStarted) { + lastStartRendered = false; + } await gate.startNow(); if (!gate.hasStarted) { return false; } - return alreadyStarted ? await render() : true; + return alreadyStarted ? await render() : lastStartRendered; } const alreadyStarted = gate.hasStarted; const progressActive = await gate.noteWork(); @@ -384,7 +401,7 @@ export function createChannelProgressDraftCompositor(params: { return gate.hasStarted; }, get isVisible() { - return gate.hasStarted && !finalReplyStarted && !finalReplyDelivered; + return Boolean(lastRenderedText) && !finalReplyStarted && !finalReplyDelivered; }, get hasStatusHeadline() { return Boolean(resolveStatusText()); @@ -444,8 +461,15 @@ export function createChannelProgressDraftCompositor(params: { return false; } if (options?.startImmediately) { + const alreadyStarted = gate.hasStarted; + if (!alreadyStarted) { + lastStartRendered = false; + } await gate.startNow(); - return gate.hasStarted ? await render({ flush: true }) : false; + if (!gate.hasStarted) { + return false; + } + return alreadyStarted ? await render({ flush: true }) : lastStartRendered; } const alreadyStarted = gate.hasStarted; const progressActive = await gate.noteWork(); @@ -484,14 +508,14 @@ export function createChannelProgressDraftCompositor(params: { return true; } const alreadyStarted = gate.hasStarted; + if (!alreadyStarted) { + lastStartRendered = false; + } await gate.startNow(); if (!gate.hasStarted) { return false; } - if (alreadyStarted) { - await render(); - } - return true; + return alreadyStarted ? await render() : lastStartRendered; }, async pushPreambleHeadline(text?: string, options?: { itemId?: string }) { if (!params.active || params.mode !== "progress" || progressSuppressed) { @@ -656,16 +680,14 @@ export function createChannelProgressDraftCompositor(params: { lastIdLessCommentaryBare = bareNormalized; } const alreadyStarted = gate.hasStarted; + if (!alreadyStarted) { + lastStartRendered = false; + } await gate.startNow(); if (!gate.hasStarted) { return false; } - if (alreadyStarted) { - await render(); - } - // True means the sanitized commentary was accepted into the visible - // lane. A first item renders inside gate.onStart, not this call site. - return true; + return alreadyStarted ? await render() : lastStartRendered; }, }; } diff --git a/src/channels/progress-draft-compositor.visibility.test.ts b/src/channels/progress-draft-compositor.visibility.test.ts new file mode 100644 index 000000000000..65e22d2cec91 --- /dev/null +++ b/src/channels/progress-draft-compositor.visibility.test.ts @@ -0,0 +1,57 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { createChannelProgressDraftCompositor } from "./progress-draft-compositor.js"; +import { DEFAULT_PROGRESS_DRAFT_INITIAL_DELAY_MS } from "./streaming.js"; + +function createProgress(update: () => Promise | boolean | void) { + return createChannelProgressDraftCompositor({ + entry: { streaming: { mode: "progress", progress: { label: "Working" } } }, + mode: "progress", + active: true, + seed: "test", + update, + }); +} + +describe("progress draft visibility", () => { + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + it.each([ + ["sync void", () => undefined], + ["async void", async () => undefined], + ["explicit true", async () => true], + ])("treats %s as accepted legacy-visible progress", async (_label, update) => { + vi.useFakeTimers(); + const progress = createProgress(update); + + expect(await progress.pushToolProgress("🛠️ Exec")).toBe(false); + await vi.advanceTimersByTimeAsync(DEFAULT_PROGRESS_DRAFT_INITIAL_DELAY_MS); + + expect(progress.isVisible).toBe(true); + }); + + it("keeps explicit false pending and retryable", async () => { + vi.useFakeTimers(); + const update = vi.fn().mockResolvedValueOnce(false).mockResolvedValueOnce(true); + const progress = createProgress(update); + + await progress.pushToolProgress("🛠️ Exec"); + await vi.advanceTimersByTimeAsync(DEFAULT_PROGRESS_DRAFT_INITIAL_DELAY_MS); + expect(progress.isVisible).toBe(false); + + expect(await progress.pushToolProgress("🛠️ Exec")).toBe(true); + expect(update).toHaveBeenCalledTimes(2); + expect(progress.isVisible).toBe(true); + }); + + it("does not dedupe a rejected update", async () => { + const update = vi.fn().mockResolvedValueOnce(false).mockResolvedValueOnce(true); + const progress = createProgress(update); + + expect(await progress.pushToolProgress("🛠️ Exec", { startImmediately: true })).toBe(false); + expect(await progress.pushToolProgress("🛠️ Exec", { startImmediately: true })).toBe(true); + expect(update).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/channels/progress-visibility.ts b/src/channels/progress-visibility.ts new file mode 100644 index 000000000000..0f9cd1e27ad9 --- /dev/null +++ b/src/channels/progress-visibility.ts @@ -0,0 +1,9 @@ +type ProgressVisibilityCallbackResult = boolean | void | Promise; + +/** Await progress without changing the legacy `void` acceptance contract. */ +export async function settleProgressVisibilityCallbackResult( + callbackResult: ProgressVisibilityCallbackResult, +): Promise<{ result: boolean | void; visible: boolean }> { + const result = await callbackResult; + return { result, visible: result !== false }; +} diff --git a/src/plugin-sdk/reply-runtime.contract.test.ts b/src/plugin-sdk/reply-runtime.contract.test.ts new file mode 100644 index 000000000000..5cba8e6fdd6a --- /dev/null +++ b/src/plugin-sdk/reply-runtime.contract.test.ts @@ -0,0 +1,35 @@ +import { describe, expectTypeOf, it } from "vitest"; +import type { GetReplyOptions } from "./reply-runtime.js"; + +type ProgressResult = boolean | void; +type ProgressCallback = GetReplyOptions[ + | "onToolResult" + | "onToolStart" + | "onItemEvent" + | "onPlanUpdate" + | "onApprovalEvent" + | "onCommandOutput" + | "onPatchSummary"]; +type ProgressBoundaryCallback = GetReplyOptions[ + | "onReasoningEnd" + | "onAssistantMessageStart" + | "onBlockReplyQueued" + | "onCompactionStart" + | "onCompactionEnd"]; + +describe("reply runtime public progress contracts", () => { + it("exports acceptance-aware progress callback results", () => { + expectTypeOf>().returns.toEqualTypeOf< + Promise | ProgressResult + >(); + expectTypeOf>().returns.toEqualTypeOf< + Promise | ProgressResult + >(); + expectTypeOf>().returns.toEqualTypeOf< + Promise | ProgressResult + >(); + expectTypeOf>().returns.toEqualTypeOf< + Promise | ProgressResult + >(); + }); +});