From a1b41d5ddb057fe0956b1d53c19fdbf4d33430d9 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Fri, 31 Jul 2026 17:17:07 -0700 Subject: [PATCH] refactor(channels): share Matrix and Slack progress draft composition (#117097) --- .../monitor/handler-draft-controller.ts | 177 ++++++------------ .../matrix/src/matrix/monitor/handler.ts | 2 +- .../message-handler/dispatch-progress.ts | 41 +--- .../dispatch.preview-fallback.test.ts | 2 +- .../progress-draft-compositor.test.ts | 18 ++ src/channels/progress-draft-compositor.ts | 24 +-- 6 files changed, 92 insertions(+), 172 deletions(-) diff --git a/extensions/matrix/src/matrix/monitor/handler-draft-controller.ts b/extensions/matrix/src/matrix/monitor/handler-draft-controller.ts index 960a7433629a..9b252bfc9658 100644 --- a/extensions/matrix/src/matrix/monitor/handler-draft-controller.ts +++ b/extensions/matrix/src/matrix/monitor/handler-draft-controller.ts @@ -1,14 +1,9 @@ import { type AgentPlanStep, buildChannelProgressDraftLineForEntry, - type ChannelProgressDraftLine, - createChannelProgressDraftGate, + createChannelProgressDraftCompositor, formatChannelProgressDraftLine, formatChannelProgressDraftText, - isChannelProgressDraftWorkToolName, - mergeChannelProgressDraftLine, - normalizeChannelProgressDraftLineIdentity, - resolveChannelProgressDraftMaxLines, } from "openclaw/plugin-sdk/channel-outbound"; import type { GetReplyOptions } from "openclaw/plugin-sdk/reply-runtime"; import type { CoreConfig, MatrixConfig, MatrixStreamingMode, ReplyToMode } from "../../types.js"; @@ -80,124 +75,41 @@ export async function createMatrixDraftController(params: { const pendingDraftBoundaries: PendingDraftBoundary[] = []; const latestQueuedDraftBoundaryOffsets = new Map(); let currentDraftReplyToId = draftReplyToId; - let previewToolProgressSuppressed = false; - let previewToolProgressLines: Array = []; - let latestPlan: AgentPlanStep[] | undefined; - let latestPlanExplanation: string | undefined; + let previewPlan: AgentPlanStep[] | undefined; + let previewPlanExplanation: string | undefined; + let previewPlanSuppressed = false; const progressConfigEntry = accountConfig ?? cfg.channels?.matrix; const progressSeed = `${accountId}:${roomId}`; - // Set after the first final payload consumes or discards the draft event - // so subsequent finals go through normal delivery. - - const renderProgressDraft = () => { - if (!draftStream) { - return; - } - const previewText = formatChannelProgressDraftText({ + const renderPreviewPlan = (): string => + formatChannelProgressDraftText({ entry: progressConfigEntry, - lines: previewToolProgressLines, + lines: [...progressDraft.getSnapshot().lines], seed: progressSeed, formatLine: formatMatrixToolProgressMarkdownCode, bullet: "-", - narration: latestPlanExplanation, - plan: latestPlan, + narration: previewPlanExplanation, + plan: previewPlan, }); - if (!previewText) { - return; - } - draftStream.update(previewText); - }; - const progressDraftGate = createChannelProgressDraftGate({ - onStart: renderProgressDraft, + const progressDraft = createChannelProgressDraftCompositor({ + entry: progressConfigEntry, + mode: streaming === "quiet" ? "partial" : streaming, + active: Boolean(draftStream), + seed: progressSeed, + formatLine: formatMatrixToolProgressMarkdownCode, + update: (text) => { + const previewText = + !progressDraftStreaming && (previewPlan || previewPlanExplanation) + ? renderPreviewPlan() + : text.replace(/^• /gmu, "- "); + draftStream?.update(previewText); + }, }); - const pushPreviewToolProgress = async ( - line?: string | ChannelProgressDraftLine, - options?: { toolName?: string }, - ) => { - if (!draftStream) { - return; - } - if (options?.toolName !== undefined && !isChannelProgressDraftWorkToolName(options.toolName)) { - return; - } - const normalized = normalizeChannelProgressDraftLineIdentity(line); - const progressLine = typeof line === "object" && line !== undefined ? line : normalized; - if (!progressDraftStreaming) { - if (!shouldStreamPreviewToolProgress || previewToolProgressSuppressed || !normalized) { - return; - } - const nextLines = mergeChannelProgressDraftLine(previewToolProgressLines, progressLine, { - maxLines: resolveChannelProgressDraftMaxLines(progressConfigEntry), - }); - if (nextLines === previewToolProgressLines) { - return; - } - previewToolProgressLines = nextLines; - draftStream.update( - formatChannelProgressDraftText({ - entry: progressConfigEntry, - lines: previewToolProgressLines, - seed: progressSeed, - formatLine: formatMatrixToolProgressMarkdownCode, - bullet: "-", - narration: latestPlanExplanation, - plan: latestPlan, - }), - ); - return; - } - if (shouldStreamPreviewToolProgress && !previewToolProgressSuppressed && normalized) { - previewToolProgressLines = mergeChannelProgressDraftLine( - previewToolProgressLines, - progressLine, - { - maxLines: resolveChannelProgressDraftMaxLines(progressConfigEntry), - }, - ); - } - const alreadyStarted = progressDraftGate.hasStarted; - const progressActive = await progressDraftGate.noteWork(); - if ((alreadyStarted || progressActive) && progressDraftGate.hasStarted) { - renderProgressDraft(); - } - }; - - const pushPlanProgress = async (steps?: AgentPlanStep[], explanation?: string) => { - latestPlan = steps?.length ? steps.map((entry) => ({ ...entry })) : undefined; - latestPlanExplanation = explanation?.replace(/\s+/g, " ").trim() || undefined; - if (!draftStream || previewToolProgressSuppressed) { - return; - } - if (!progressDraftStreaming) { - renderProgressDraft(); - return; - } - const alreadyStarted = progressDraftGate.hasStarted; - await progressDraftGate.startNow(); - if (alreadyStarted && progressDraftGate.hasStarted) { - // An empty-render clear keeps the prior draft visible on purpose: - // deleting mid-turn drops the edit anchor, and zero-step snapshots - // only arrive from label:false configs with retracting producers. - renderProgressDraft(); - } - }; - - const suppressPreviewToolProgressForAnswerText = (text: string | undefined) => { - if (!text?.trim()) { - return; - } - previewToolProgressSuppressed = true; - previewToolProgressLines = []; - latestPlan = undefined; - latestPlanExplanation = undefined; - }; - const resetPreviewToolProgress = () => { - previewToolProgressSuppressed = false; - previewToolProgressLines = []; - latestPlan = undefined; - latestPlanExplanation = undefined; + previewPlan = undefined; + previewPlanExplanation = undefined; + previewPlanSuppressed = false; + progressDraft.reset(); }; const buildPreviewToolProgressReplyOptions = (): Partial => { @@ -214,7 +126,7 @@ export async function createMatrixDraftController(params: { ...options, onToolStart: async (payload) => { const toolName = payload.name?.trim(); - await pushPreviewToolProgress( + await progressDraft.pushToolProgress( buildChannelProgressDraftLineForEntry( progressConfigEntry, { @@ -231,7 +143,7 @@ export async function createMatrixDraftController(params: { ); }, onItemEvent: async (payload) => { - await pushPreviewToolProgress( + await progressDraft.pushToolProgress( buildChannelProgressDraftLineForEntry(progressConfigEntry, { event: "item", itemId: payload.itemId, @@ -251,13 +163,27 @@ export async function createMatrixDraftController(params: { if (payload.phase !== "update") { return; } - await pushPlanProgress(payload.steps, payload.explanation); + if (progressDraftStreaming) { + await progressDraft.pushPlanProgress(payload.steps, { explanation: payload.explanation }); + return; + } + if (!draftStream || previewPlanSuppressed) { + return; + } + previewPlan = payload.steps?.length + ? payload.steps.map((step) => ({ ...step })) + : undefined; + previewPlanExplanation = payload.explanation?.replace(/\s+/g, " ").trim() || undefined; + const text = renderPreviewPlan(); + if (text) { + draftStream.update(text); + } }, onApprovalEvent: async (payload) => { if (payload.phase !== "requested") { return; } - await pushPreviewToolProgress( + await progressDraft.pushToolProgress( formatChannelProgressDraftLine({ event: "approval", phase: payload.phase, @@ -272,7 +198,7 @@ export async function createMatrixDraftController(params: { if (payload.phase !== "end") { return; } - await pushPreviewToolProgress( + await progressDraft.pushToolProgress( buildChannelProgressDraftLineForEntry(progressConfigEntry, { event: "command-output", itemId: payload.itemId, @@ -289,7 +215,7 @@ export async function createMatrixDraftController(params: { if (payload.phase !== "end") { return; } - await pushPreviewToolProgress( + await progressDraft.pushToolProgress( buildChannelProgressDraftLineForEntry(progressConfigEntry, { event: "patch", itemId: payload.itemId, @@ -376,13 +302,13 @@ export async function createMatrixDraftController(params: { pendingDraftBoundaries.length = 0; latestQueuedDraftBoundaryOffsets.clear(); currentDraftReplyToId = draftReplyToId; - progressDraftGate.reset(); + progressDraft.beginNewTurn({ force: true }); resetPreviewToolProgress(); }; return { draftStream, - progressDraftGate, + cancelProgressDraft: () => progressDraft.cancel(), buildPreviewToolProgressReplyOptions, queueDraftBlockBoundary, advanceDraftBlockBoundary, @@ -409,7 +335,12 @@ export async function createMatrixDraftController(params: { return; } latestDraftFullText = text; - suppressPreviewToolProgressForAnswerText(text); + if (text.trim()) { + previewPlanSuppressed = true; + previewPlan = undefined; + previewPlanExplanation = undefined; + progressDraft.suppress(); + } updateDraftFromLatestFullText(); }, }; diff --git a/extensions/matrix/src/matrix/monitor/handler.ts b/extensions/matrix/src/matrix/monitor/handler.ts index d969fe9b321c..162101a79332 100644 --- a/extensions/matrix/src/matrix/monitor/handler.ts +++ b/extensions/matrix/src/matrix/monitor/handler.ts @@ -526,7 +526,7 @@ export function createMatrixRoomMessageHandler(params: MatrixMonitorHandlerParam }, dispatcherOptions: { ...turnDispatcherOptions, - onSettled: () => draftController.progressDraftGate.cancel(), + onSettled: () => draftController.cancelProgressDraft(), }, replyOptions: { skillFilter: roomConfig?.skills, diff --git a/extensions/slack/src/monitor/message-handler/dispatch-progress.ts b/extensions/slack/src/monitor/message-handler/dispatch-progress.ts index e3d44416554c..56616d192112 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch-progress.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch-progress.ts @@ -4,8 +4,6 @@ import { createChannelProgressReceiptTracker, formatChannelProgressDraftText, isChannelProgressDraftWorkToolName, - mergeChannelProgressDraftLine, - resolveChannelProgressDraftMaxLines, resolveChannelProgressDraftMaxLineChars, resolveChannelProgressDraftRender, resolveChannelStreamingPreviewToolProgress, @@ -110,7 +108,6 @@ export function createSlackProgressRuntime(runtimeParams: { previewStreamingEnabled, }); let previewToolProgressSuppressed = false; - let legacyPreviewToolProgressLines: ChannelProgressDraftLine[] = []; // Last task rows emitted to the native stream; reconciliation terminalizes // ids that drop out (plan shrinks, tool-line <-> plan source switches). let nativeTaskState: SlackNativeTaskSnapshot = new Map(); @@ -321,7 +318,7 @@ export function createSlackProgressRuntime(runtimeParams: { const progressDraft = createChannelProgressDraftCompositor({ entry: account.config, mode: slackStreaming.mode, - active: progressDraftActive && streamMode === "status_final", + active: progressDraftActive, seed: progressSeed, formatLine: escapeSlackMrkdwn, reasoningLinePrefix: "🧠 ", @@ -437,7 +434,7 @@ export function createSlackProgressRuntime(runtimeParams: { } const text = formatChannelProgressDraftText({ entry: account.config, - lines: legacyPreviewToolProgressLines, + lines: [...progressDraft.getSnapshot().lines], seed: progressSeed, formatLine: escapeSlackMrkdwn, narration: explanation, @@ -468,31 +465,10 @@ export function createSlackProgressRuntime(runtimeParams: { await progressDraft.pushToolProgress(line, options); return; } - if ( - !line || - !normalized || - !draftStream || - !previewToolProgressEnabled || - previewToolProgressSuppressed - ) { + if (!line || !normalized || !draftStream || !previewToolProgressEnabled) { return; } - const nextLines = mergeChannelProgressDraftLine(legacyPreviewToolProgressLines, line, { - maxLines: resolveChannelProgressDraftMaxLines(account.config), - }); - if (nextLines === legacyPreviewToolProgressLines) { - return; - } - legacyPreviewToolProgressLines = nextLines; - draftStream.update( - formatChannelProgressDraftText({ - entry: account.config, - lines: legacyPreviewToolProgressLines, - seed: progressSeed, - formatLine: escapeSlackMrkdwn, - }), - ); - hasStreamedMessage = true; + await progressDraft.pushToolProgress(line, options); }; const updateDraftFromPartial = (text?: string) => { @@ -503,7 +479,7 @@ export function createSlackProgressRuntime(runtimeParams: { if (streamMode === "append") { previewToolProgressSuppressed = true; - legacyPreviewToolProgressLines = []; + progressDraft.suppress(); const next = applyAppendOnlyStreamUpdate({ incoming: trimmed, rendered: appendRenderedText, @@ -524,7 +500,7 @@ export function createSlackProgressRuntime(runtimeParams: { } previewToolProgressSuppressed = true; - legacyPreviewToolProgressLines = []; + progressDraft.suppress(); draftStream?.update(trimmed); hasStreamedMessage = true; }; @@ -551,6 +527,8 @@ export function createSlackProgressRuntime(runtimeParams: { text: normalized, label: "Reasoning", }); + // Tool admission closes reasoning bursts; restore this still-open preview lane. + progressDraft.mergeReasoningProgress(normalized, { snapshot: true }); return; } progressReceipt.noteReasoning(); @@ -564,9 +542,8 @@ export function createSlackProgressRuntime(runtimeParams: { appendSourceText = ""; }; const resetDraftProgressState = () => { - progressDraft.resetReasoningProgress(); previewToolProgressSuppressed = false; - legacyPreviewToolProgressLines = []; + progressDraft.reset(); }; const beginNewProgressTurn = async (options?: { force?: boolean }) => { const completionChunks = diff --git a/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts b/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts index 0e844b3b4645..3ccc09f05768 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch.preview-fallback.test.ts @@ -1942,7 +1942,7 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { ); expect(draftStream.update).toHaveBeenCalledWith( - "Shelling\n• ran <!here> <@U123> \\*bold\\* \\`code\\` & done", + "Shelling\n\n• ran <!here> <@U123> \\*bold\\* \\`code\\` & done", ); }); diff --git a/src/channels/progress-draft-compositor.test.ts b/src/channels/progress-draft-compositor.test.ts index 8bf60cc2bc9d..942454d408ec 100644 --- a/src/channels/progress-draft-compositor.test.ts +++ b/src/channels/progress-draft-compositor.test.ts @@ -59,6 +59,24 @@ describe("createChannelProgressDraftCompositor", () => { expect(update).toHaveBeenLastCalledWith("🛠️ Next", expect.anything()); }); + it("publishes partial-preview tool lines without enabling progress-only plans", async () => { + const update = vi.fn(); + const progress = createChannelProgressDraftCompositor({ + entry: { streaming: { mode: "partial", progress: { label: false } } }, + mode: "partial", + active: true, + seed: "preview", + update, + }); + + await progress.pushToolProgress("Inspecting files"); + expect(update).toHaveBeenLastCalledWith("• Inspecting files", { + lines: ["Inspecting files"], + }); + expect(await progress.pushPlanProgress([{ step: "Patch", status: "in_progress" }])).toBe(false); + expect(update).toHaveBeenCalledTimes(1); + }); + it("returns detached structured state for channel-native renderers", async () => { const progress = createChannelProgressDraftCompositor({ entry: { streaming: { mode: "progress", progress: { label: false } } }, diff --git a/src/channels/progress-draft-compositor.ts b/src/channels/progress-draft-compositor.ts index 7fd684c5b0f6..babbe3205ab6 100644 --- a/src/channels/progress-draft-compositor.ts +++ b/src/channels/progress-draft-compositor.ts @@ -264,10 +264,7 @@ export function createChannelProgressDraftCompositor(params: { planExplanation = ""; }; - const render = async (options?: { flush?: boolean }): Promise => { - if (!params.active || params.mode !== "progress" || finalReplyStarted || finalReplyDelivered) { - return false; - } + const publish = async (options?: { flush?: boolean }): Promise => { const text = formatDraftText(); const linesChanged = params.updateOnLineChange === true && lines !== lastRenderedLines; if (!text || (text === lastRenderedText && !linesChanged)) { @@ -279,6 +276,13 @@ export function createChannelProgressDraftCompositor(params: { return true; }; + const render = async (options?: { flush?: boolean }): Promise => { + if (!params.active || params.mode !== "progress" || finalReplyStarted || finalReplyDelivered) { + return false; + } + return await publish(options); + }; + const schedulePreambleExpiryRefresh = () => { clearPreambleExpiryTimer(); if ( @@ -409,17 +413,7 @@ export function createChannelProgressDraftCompositor(params: { } lines = nextLines; if (params.mode !== "progress") { - if (!shouldStoreLine) { - return false; - } - const text = formatDraftText(); - if (!text || text === lastRenderedText) { - return false; - } - lastRenderedText = text; - lastRenderedLines = lines; - await params.update(text, { lines: [...lines] }); - return true; + return shouldStoreLine ? await publish() : false; } if (options?.startImmediately || params.shouldStartNow?.(line)) { const alreadyStarted = gate.hasStarted;