fix(progress): preserve callback acceptance results (#120171)

* fix(progress): preserve callback acceptance results

* fix(progress): require transport acknowledgements

* fix(progress): preserve direct acceptance outcomes

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

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

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;

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

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<void>;
 };
 
@@ -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<string, unknown> = {
         id,
@@ -316,6 +316,7 @@ export function createClickClackAgentProgressPublisher(params: {
         line,
       });
       seenLines.add(id);
+      return false;
     },
     async finalize() {
       if (!started || cleared) {

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") {

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();
+  });
+});

extensions/discord/src/monitor/message-handler.process-progress.ts

@@ -88,11 +88,17 @@ export function createDiscordMessageProgressRuntime(params: {
   const buildProgressSummaryLine = () => `-# ${progressReceipt.buildSummaryLine()}`;
 
   const replyOptions: Partial<ReplyOptions> = {
-    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;
     },
   };
 

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?.({

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<boolean | void> = [];
+
+    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],

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> | false | void;
+    }) => Promise<boolean | void> | boolean | void;
     onNarrationUpdate?: (payload: { text: string }) => Promise<void> | void;
     onProgressNarratorLifecycle?: (lifecycle: {
       beginTurn: () => void;

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: () => ({

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;
                 },
               }
             : {}),

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;
     },
   };
 }

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();
+  });
+});

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<void>;
   stop: () => Promise<string | undefined>;
   discardPending: () => Promise<void>;
   eventId: () => string | undefined;

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

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;
             },
           },
         }),

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,

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;
                   }
                 },
               }

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<void> {
+  async onPartialReply(payload: { text?: string }): Promise<boolean> {
     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 保证串行调用） */

extensions/signal/src/monitor/event-handler.inbound-context.test.ts

@@ -28,9 +28,9 @@ type DispatchInboundMessageMockParams = {
     allowProgressCallbacksWhenSourceDeliverySuppressed?: boolean;
     allowToolLifecycleWhenProgressHidden?: boolean;
     onReplyStart?: () => void | Promise<void>;
-    onToolStart?: (payload: { name?: string }) => void | Promise<void>;
-    onCompactionStart?: () => void | Promise<void>;
-    onCompactionEnd?: () => void | Promise<void>;
+    onToolStart?: (payload: { name?: string }) => boolean | void | Promise<boolean | void>;
+    onCompactionStart?: () => boolean | void | Promise<boolean | void>;
+    onCompactionEnd?: () => boolean | void | Promise<boolean | void>;
   };
 };
 

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;
                   },
                 }
               : {}),

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<ReturnType<typeof buildSlackProgressStreamStartChunks>>,
-    chunkKey: string,
-  ) => {
+  ): Promise<boolean> => {
     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<ReturnType<typeof buildSlackProgressStreamUpdateChunks>>,
-    chunkKey: string,
-  ) => {
+  ): Promise<boolean> => {
     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<boolean> => {
     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,
     });
   };

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);

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);
         },
       },
     });

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 {

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,

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);

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;
           },
         }
       : {}),

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<void>;
+  onPartialReply?: (payload: PartialReplyPayload) => boolean | void | Promise<boolean | void>;
   onAssistantMessageStart?: () => void | Promise<void>;
   onBlockReply?: (payload: BlockReplyPayload, context?: BlockReplyContext) => void | Promise<void>;
   onBlockReplyFlush?: (context: BlockReplyFlushContext) => void | Promise<void>;

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<void>;
+  callback: () => unknown;
   label: string;
   log: CallbackLogger;
 }): void {
   try {
     const result = params.callback();
-    if (isPromiseLike<void>(result)) {
+    if (isPromiseLike<unknown>(result)) {
       void Promise.resolve(result).catch((error: unknown) => {
         params.log.warn(`${params.label} callback failed: ${String(error)}`);
       });

src/agents/embedded-agent-subscribe.types.ts

@@ -66,7 +66,7 @@ export type SubscribeEmbeddedAgentSessionParams = {
   onBlockReplyFlush?: (context: BlockReplyFlushContext) => void | Promise<void>;
   blockReplyBreak?: "text_end" | "message_end";
   blockReplyChunking?: BlockReplyChunking;
-  onPartialReply?: (payload: PartialReplyPayload) => void | Promise<void>;
+  onPartialReply?: (payload: PartialReplyPayload) => boolean | void | Promise<boolean | void>;
   onAssistantMessageStart?: () => void | Promise<void>;
   onExecutionPhase?: (info: {
     phase: "tool_execution_started";

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> | void;
-  onReasoningStream?: (payload: ReasoningStreamPayload) => Promise<void> | void;
+  onPartialReply?: (
+    payload: PartialReplyPayload,
+  ) => Promise<ProgressCallbackResult> | ProgressCallbackResult;
+  onReasoningStream?: (
+    payload: ReasoningStreamPayload,
+  ) => Promise<ProgressCallbackResult> | ProgressCallbackResult;
   onReasoningProgress?: (payload: ReasoningProgressPayload) => Promise<void> | void;
   streamReasoningInNonStreamModes?: boolean;
   /** Called when a thinking/reasoning block ends. */
-  onReasoningEnd?: () => Promise<void> | void;
+  onReasoningEnd?: () => Promise<ProgressCallbackResult> | ProgressCallbackResult;
   /** Called when a new assistant message starts (e.g., after tool call or thinking block). */
-  onAssistantMessageStart?: () => Promise<void> | void;
+  onAssistantMessageStart?: () => Promise<ProgressCallbackResult> | 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> | void;
+  onBlockReplyQueued?: (
+    payload: ReplyPayload,
+    context?: BlockReplyContext,
+  ) => Promise<ProgressCallbackResult> | ProgressCallbackResult;
   onBlockReply?: (payload: ReplyPayload, context?: BlockReplyContext) => Promise<void> | void;
-  onToolResult?: (payload: ReplyPayload) => Promise<void> | void;
+  onToolResult?: (
+    payload: ReplyPayload,
+  ) => Promise<ProgressCallbackResult> | 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<string, unknown>;
     detailMode?: "explain" | "raw";
-  }) => Promise<void> | void;
+  }) => Promise<ProgressCallbackResult> | 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> | void;
+  }) => Promise<ProgressCallbackResult> | 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> | void;
+  }) => Promise<ProgressCallbackResult> | 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> | void;
+  }) => Promise<ProgressCallbackResult> | ProgressCallbackResult;
   /** Called when context auto-compaction starts (allows UX feedback during the pause). */
-  onCompactionStart?: () => Promise<void> | void;
+  onCompactionStart?: () => Promise<ProgressCallbackResult> | ProgressCallbackResult;
   /** Called when context auto-compaction completes. */
-  onCompactionEnd?: () => Promise<void> | void;
+  onCompactionEnd?: () => Promise<ProgressCallbackResult> | 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;

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<void>) => Promise<void>;
+  schedule: (deliver: () => Promise<unknown>) => Promise<void>;
 };
 
 export function createAgentEventDeliveryStartOrder(): AgentEventDeliveryStartOrder {
@@ -18,7 +18,7 @@ export function createAgentEventDeliveryStartOrder(): AgentEventDeliveryStartOrd
         releaseStart = resolve;
       });
       await previousStart;
-      let delivery: Promise<void>;
+      let delivery: Promise<unknown>;
       try {
         delivery = deliver();
       } finally {
@@ -33,7 +33,7 @@ export function createAgentEventBridge<T>(params: {
   runId: string;
   suppressed?: boolean;
   read: (evt: AgentEventPayload) => T | undefined;
-  deliver?: (payload: T) => Promise<void>;
+  deliver?: (payload: T) => Promise<unknown>;
   startOrder?: AgentEventDeliveryStartOrder;
 }) {
   const deliver = params.deliver;
@@ -44,7 +44,7 @@ export function createAgentEventBridge<T>(params: {
     };
   }
   let unsubscribed = false;
-  let delivery = Promise.resolve();
+  let delivery: Promise<unknown> = Promise.resolve();
   const rawUnsubscribe = onAgentEvent((evt) => {
     if (evt.runId !== params.runId) {
       return;

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,
-                  }),
+                  });
+                },
               ),
             ]);
           },

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<void>;
+  deliver?: (text: string) => Promise<boolean | void>;
   startOrder?: AgentEventDeliveryStartOrder;
 }) {
   let lastText: string | undefined;
@@ -431,7 +431,7 @@ type RunCliAgentWithLifecycleParams = {
    */
   onActivity?: () => void;
   preserveProgressCallbackStartOrder?: boolean;
-  onAssistantText?: (text: string) => Promise<void>;
+  onAssistantText?: (text: string) => Promise<boolean | void>;
   onReasoningText?: (payload: ReasoningTextPayload) => Promise<void>;
   onReasoningProgress?: (payload: ReasoningProgressPayload) => Promise<void>;
   onToolEvent?: (payload: CliToolEventPayload) => Promise<void>;

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,

src/auto-reply/reply/agent-runner-presentation.ts

@@ -24,8 +24,8 @@ type AgentTurnPresentation = {
   normalizeStreamingText: (payload: ReplyPayload) => { text?: string; skip: boolean };
   startPresentationWhileTyping: (
     typingPromise: Promise<void>,
-    startPresentation: () => void | Promise<void>,
-  ) => Promise<void>;
+    startPresentation: () => boolean | void | Promise<boolean | void>,
+  ) => Promise<boolean | void>;
   blockReplyHandler: ReturnType<typeof createBlockReplyDeliveryHandler> | undefined;
 };
 
@@ -93,9 +93,9 @@ export function createAgentTurnPresentation(params: {
 
   const startPresentationWhileTyping = async (
     typingPromise: Promise<void>,
-    startPresentation: () => void | Promise<void>,
+    startPresentation: () => boolean | void | Promise<boolean | void>,
   ) => {
-    let presentationPromise: void | Promise<void>;
+    let presentationPromise: boolean | void | Promise<boolean | void>;
     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;

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<NonNullable<typeof opts.onPartialReply>>[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;

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> | void;
+  onPartialReply?: (payload: { text?: string }) => Promise<boolean | void> | boolean | void;
   onAssistantMessageStart?: () => Promise<void> | void;
   onReasoningStream?: (payload: { text?: string }) => Promise<void> | 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();
       }

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" };
     };
 

src/auto-reply/reply/dispatch-from-config.prepare-execution.ts

@@ -265,7 +265,7 @@ export async function prepareDispatchExecution(state: ChooseDispatchRouteReadySt
       releaseStart: () => releaseStart?.(),
     };
   };
-  const wrapProgressCallback = <Args extends unknown[], Result extends false | void>(
+  const wrapProgressCallback = <Args extends unknown[], Result extends boolean | void>(
     callback: ((...args: Args) => Promise<Result> | Result) | undefined,
     options?: {
       allowWhenToolSummariesHidden?: boolean;

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?.({

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);
   });
 

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> | boolean | void,
+  ): Promise<boolean | void> => {
+    let completed = false;
+    let result: boolean | void = false;
+    await enqueueProgress(async () => {
+      result = await deliver();
+      completed = true;
+    });
+    return completed ? result : false;
+  };
   const wrap = <T>(callback: ((value: T) => unknown) | undefined, allowed = progressAllowed) =>
     callback
       ? (value: T) =>
@@ -134,6 +146,19 @@ export async function executeFollowupTurn(params: {
             }
           })
       : undefined;
+  const wrapVisibility = <T>(
+    callback: ((value: T) => Promise<boolean | void> | 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;
       });
     },
   };

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> | void;
+  update: (
+    text: string,
+    options?: ChannelProgressDraftUpdateOptions,
+  ) => Promise<boolean | void> | boolean | void;
   deleteCurrent?: () => Promise<void> | void;
   tryNativeUpdate?: (text: string) => Promise<boolean> | 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<typeof setTimeout> | 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<boolean> => {
@@ -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;
     },
   };
 }

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> | 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);
+  });
+});

src/channels/progress-visibility.ts

@@ -0,0 +1,9 @@
+type ProgressVisibilityCallbackResult = boolean | void | Promise<boolean | void>;
+
+/** 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 };
+}

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<Exclude<ProgressCallback, undefined>>().returns.toEqualTypeOf<
+      Promise<ProgressResult> | ProgressResult
+    >();
+    expectTypeOf<Exclude<GetReplyOptions["onPartialReply"], undefined>>().returns.toEqualTypeOf<
+      Promise<ProgressResult> | ProgressResult
+    >();
+    expectTypeOf<Exclude<GetReplyOptions["onReasoningStream"], undefined>>().returns.toEqualTypeOf<
+      Promise<ProgressResult> | ProgressResult
+    >();
+    expectTypeOf<Exclude<ProgressBoundaryCallback, undefined>>().returns.toEqualTypeOf<
+      Promise<ProgressResult> | ProgressResult
+    >();
+  });
+});