diff --git a/extensions/slack/src/__traces__/progress-native-unified.trace.jsonl b/extensions/slack/src/__traces__/progress-native-unified.trace.jsonl index ca74c17e5e9b..3e3e22cbb5d1 100644 --- a/extensions/slack/src/__traces__/progress-native-unified.trace.jsonl +++ b/extensions/slack/src/__traces__/progress-native-unified.trace.jsonl @@ -5,11 +5,11 @@ {"seq":5,"at":1500,"dir":"out","kind":"users.info","data":{"payload":{"user":"U0TRACE"},"result":{"team_id":"T0TRACE"},"target":"U0TRACE"}} {"seq":6,"at":1500,"dir":"out","kind":"chat.startStream","data":{"payload":{"channel":"C0TRACE","chunks":[{"title":"I’m checking the native Slack stream before applying the focused patch.","type":"plan_update"},{"details":"to src/native-card.ts (22 chars)","id":"write_1_fdf08c92","status":"in_progress","title":"Write","type":"task_update"}],"recipient_team_id":"T0TRACE","recipient_user_id":"U0TRACE","task_display_mode":"plan","thread_ts":"ts#1"},"result":{"ts":"ts#2"},"target":"C0TRACE"}} {"seq":7,"at":2000,"dir":"in","kind":"partial","data":{"text":"I’m checking the native Slack stream before applying the focused patch. I’m applying it now."}} -{"seq":8,"at":2000,"dir":"out","kind":"chat.appendStream","data":{"payload":{"channel":"C0TRACE","chunks":[{"title":"I’m checking the native Slack stream before applying the focused patch. I’m applying it now.","type":"plan_update"},{"details":"to src/native-card.ts (22 chars)","id":"write_1_fdf08c92","status":"in_progress","title":"Write","type":"task_update"}],"ts":"ts#2"},"result":{"ok":true},"target":"ts#2"}} +{"seq":8,"at":2000,"dir":"out","kind":"chat.appendStream","data":{"payload":{"channel":"C0TRACE","chunks":[{"title":"I’m checking the native Slack stream before applying the focused patch. I’m applying it now.","type":"plan_update"}],"ts":"ts#2"},"result":{"ok":true},"target":"ts#2"}} {"seq":9,"at":2000,"dir":"in","kind":"tool-progress","data":{"name":"write","phase":"result"}} -{"seq":10,"at":2000,"dir":"out","kind":"chat.appendStream","data":{"payload":{"channel":"C0TRACE","chunks":[{"title":"I’m checking the native Slack stream before applying the focused patch. I’m applying it now.","type":"plan_update"},{"details":"src/native-card.ts","id":"write_1_fdf08c92","status":"complete","title":"Write","type":"task_update"}],"ts":"ts#2"},"result":{"ok":true},"target":"ts#2"}} +{"seq":10,"at":2000,"dir":"out","kind":"chat.appendStream","data":{"payload":{"channel":"C0TRACE","chunks":[{"id":"write_1_fdf08c92","status":"complete","title":"Write","type":"task_update"}],"ts":"ts#2"},"result":{"ok":true},"target":"ts#2"}} {"seq":11,"at":2000,"dir":"in","kind":"final","data":{"text":"The unified native Slack turn is complete."}} -{"seq":12,"at":2000,"dir":"out","kind":"chat.appendStream","data":{"payload":{"channel":"C0TRACE","chunks":[{"title":"I’m checking the native Slack stream before applying the focused patch. I’m applying it now.","type":"plan_update"},{"details":"src/native-card.ts","id":"write_1_fdf08c92","output":"+2 −0","sources":[{"text":"Open in OpenClaw","type":"url_source","url":"https://team.openclaw.ai/openclaw/chat/trace-agent/slack/channel/c0trace"}],"status":"complete","title":"Write","type":"task_update"}],"ts":"ts#2"},"result":{"ok":true},"target":"ts#2"}} +{"seq":12,"at":2000,"dir":"out","kind":"chat.appendStream","data":{"payload":{"channel":"C0TRACE","chunks":[{"id":"write_1_fdf08c92","output":"+2 −0","sources":[{"text":"Open in OpenClaw","type":"url_source","url":"https://team.openclaw.ai/openclaw/chat/trace-agent/slack/channel/c0trace"}],"status":"complete","title":"Write","type":"task_update"}],"ts":"ts#2"},"result":{"ok":true},"target":"ts#2"}} {"seq":13,"at":2000,"dir":"in","kind":"idle"} {"seq":14,"at":2000,"dir":"out","kind":"assistant.threads.setStatus","data":{"payload":{"channel_id":"C0TRACE","status":"","thread_ts":"ts#1"},"result":{"ok":true},"target":"C0TRACE/ts#1"}} {"seq":15,"at":2000,"dir":"out","kind":"chat.stopStream","data":{"payload":{"channel":"C0TRACE","chunks":[{"text":"\nThe unified native Slack turn is complete.","type":"markdown_text"}],"ts":"ts#2"},"result":{"ok":true},"target":"ts#2"}} diff --git a/extensions/slack/src/monitor/message-handler/dispatch-progress.ts b/extensions/slack/src/monitor/message-handler/dispatch-progress.ts index 6c03341ee919..506488bb6147 100644 --- a/extensions/slack/src/monitor/message-handler/dispatch-progress.ts +++ b/extensions/slack/src/monitor/message-handler/dispatch-progress.ts @@ -20,7 +20,8 @@ import { SLACK_EDIT_TEXT_MAX_BYTES, SLACK_TEXT_LIMIT } from "../../limits.js"; import { buildSlackProgressStreamCompletionChunks, reconcileSlackNativeTaskChunks, - type SlackNativeTaskSnapshot, + EMPTY_SLACK_NATIVE_STREAM_SNAPSHOT, + type SlackNativeStreamSnapshot, } from "../../progress-blocks.js"; import { applyAppendOnlyStreamUpdate } from "../../stream-mode.js"; import { appendSlackStream, stopSlackStream } from "../../streaming.js"; @@ -103,16 +104,16 @@ export function createSlackProgressRuntime(runtimeParams: { previewStreamingEnabled, }); let previewToolProgressSuppressed = false; - // 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(); + // Plan title and task rows already delivered to the native stream; the + // reconciler diffs each snapshot against it and terminalizes ids that drop + // out (plan shrinks, tool-line <-> plan source switches). + let nativeStreamSnapshot: SlackNativeStreamSnapshot = EMPTY_SLACK_NATIVE_STREAM_SNAPSHOT; let appendRenderedText = ""; let appendSourceText = ""; let nativeProgressCompletionSent = false; // Terminal status of the turn's final payload; completion retries and // queued rotation must not repaint an errored turn as complete. let nativeProgressTerminalStatus: "complete" | "error" = "complete"; - let nativeProgressChunkKey: string | undefined; let nativeNarrationRenderedText = ""; let nativeNarrationSourceText = ""; // Native streaming appends; overlapping updates would re-append identical @@ -218,7 +219,7 @@ export function createSlackProgressRuntime(runtimeParams: { const snapshot = progressDraft.getSnapshot(); const progressLines = resolveNativeProgressLines(snapshot); const narrationUpdate = resolveNarrationUpdate(resolveNativeProgressNarration(snapshot)); - const hasRetirableNativeTasks = [...nativeTaskState.values()].some( + const hasRetirableNativeTasks = [...nativeStreamSnapshot.tasks.values()].some( (task) => task.status !== "complete" && task.status !== "error", ); if ( @@ -238,12 +239,10 @@ export function createSlackProgressRuntime(runtimeParams: { return false; } const reconciled = reconcileSlackNativeTaskChunks({ - previousTasks: nativeTaskState, + previous: nativeStreamSnapshot, chunks: buildNativeProgressChunks(snapshot), }); - const chunkKey = JSON.stringify(reconciled.chunks ?? []); - const taskChunksChanged = chunkKey !== nativeProgressChunkKey; - const chunks = taskChunksChanged ? reconciled.chunks : undefined; + const chunks = reconciled.chunks; if (!chunks?.length && !narrationUpdate.delta) { return false; } @@ -268,9 +267,8 @@ export function createSlackProgressRuntime(runtimeParams: { nativeNarrationRenderedText = narrationUpdate.next.rendered; nativeNarrationSourceText = narrationUpdate.next.source; } - if (taskChunksChanged) { - nativeProgressChunkKey = chunkKey; - nativeTaskState = reconciled.tasks; + if (chunks?.length) { + nativeStreamSnapshot = reconciled.snapshot; } return true; } catch (err) { @@ -397,7 +395,7 @@ export function createSlackProgressRuntime(runtimeParams: { const snapshot = progressDraft.getSnapshot(); const lines = resolveNativeProgressLines(snapshot); const sessionUrl = progressCard.resolveSessionUrl(); - const hasRetirableNativeTasks = [...nativeTaskState.values()].some( + const hasRetirableNativeTasks = [...nativeStreamSnapshot.tasks.values()].some( (task) => task.status !== "complete" && task.status !== "error", ); if ( @@ -410,7 +408,7 @@ export function createSlackProgressRuntime(runtimeParams: { return undefined; } return reconcileSlackNativeTaskChunks({ - previousTasks: nativeTaskState, + previous: nativeStreamSnapshot, chunks: buildSlackProgressStreamCompletionChunks({ title: resolveNativeProgressTitle(snapshot) ?? @@ -597,10 +595,9 @@ export function createSlackProgressRuntime(runtimeParams: { await dropDetachedProgressCards(); } resetProgressTurnState(); - nativeTaskState = new Map(); + nativeStreamSnapshot = EMPTY_SLACK_NATIVE_STREAM_SNAPSHOT; nativeProgressCompletionSent = false; nativeProgressTerminalStatus = "complete"; - nativeProgressChunkKey = undefined; progressCard.reset(); // A re-armed turn is a new visible reply: it must not dedupe against or // inherit delivery state from the settled turn (mirrors queued admission). 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 601a0a80e2c9..73033c1265ba 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 @@ -2611,13 +2611,13 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { planUpdate("tool one"), taskUpdate(contentTaskId("item"), "tool one", "in_progress"), ]); + // Rows already on the stream are not resent: Slack appends task text per + // update, so each append carries only the changed rows. expectNativeProgressAppend(0, [ planUpdate("tool two"), - taskUpdate(contentTaskId("item"), "tool one", "in_progress"), taskUpdate(contentTaskId("item"), "tool two", "in_progress"), ]); expectNativeProgressAppend(2, [ - planUpdate("tool three"), taskUpdate(contentTaskId("item"), "tool one", "complete"), taskUpdate(contentTaskId("item"), "tool two", "complete"), taskUpdate(contentTaskId("item"), "tool three", "complete"), @@ -2638,10 +2638,7 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { planUpdate("slow tool"), taskUpdate(contentTaskId("item"), "slow tool", "in_progress"), ]); - expectNativeProgressAppend(0, [ - planUpdate("slow tool"), - taskUpdate(contentTaskId("item"), "slow tool", "complete"), - ]); + expectNativeProgressAppend(0, [taskUpdate(contentTaskId("item"), "slow tool", "complete")]); expect(startSlackStreamMock.mock.invocationCallOrder[0]).toBeLessThan( appendSlackStreamMock.mock.invocationCallOrder[0] ?? 0, ); @@ -3341,7 +3338,7 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { expect(createSlackDraftStreamMock).not.toHaveBeenCalled(); expectNativeProgressStart([planUpdate("bash"), taskUpdate(taskId, "bash", "in_progress")]); - expectNativeProgressAppend(0, [planUpdate("bash"), taskUpdate(taskId, "bash", "complete")]); + expectNativeProgressAppend(0, [taskUpdate(taskId, "bash", "complete")]); expect(startSlackStreamMock.mock.invocationCallOrder[0]).toBeLessThan( appendSlackStreamMock.mock.invocationCallOrder[0] ?? 0, ); @@ -3402,9 +3399,7 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { expect.stringMatching(/^command_call_1_[a-f0-9]{8}$/), ]); expect(taskUpdates.at(0)?.id).toEqual(expect.stringMatching(/^command_call_1_[a-f0-9]{8}$/)); - expect(taskUpdates).toContainEqual( - taskUpdate(taskUpdates.at(0)?.id, "bash", "complete", { details: "completed" }), - ); + expect(taskUpdates).toContainEqual(taskUpdate(taskUpdates.at(0)?.id, "bash", "complete")); expect(deliverRepliesMock).not.toHaveBeenCalled(); expectNativeStreamText(`\n${FINAL_REPLY_TEXT}`); }); @@ -3681,10 +3676,7 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { planUpdate("failing tool"), taskUpdate(contentTaskId("item"), "failing tool", "in_progress"), ]); - expectNativeProgressAppend(0, [ - planUpdate("failing tool"), - taskUpdate(contentTaskId("item"), "failing tool", "error"), - ]); + expectNativeProgressAppend(0, [taskUpdate(contentTaskId("item"), "failing tool", "error")]); expect(deliverRepliesMock).toHaveBeenCalledTimes(1); expect(finalizeSlackPreviewEditMock).not.toHaveBeenCalled(); const deliverParams = requireRecord( @@ -3710,7 +3702,6 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { expect(deliverRepliesMock).not.toHaveBeenCalled(); expectMockCallArgFields(stopSlackStreamMock, 0, "native progress stream stop", { chunks: [ - planUpdate("tool three"), taskUpdate(contentTaskId("item"), "tool one", "complete"), taskUpdate(contentTaskId("item"), "tool two", "complete"), taskUpdate(contentTaskId("item"), "tool three", "complete"), @@ -3735,7 +3726,6 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { taskUpdate(contentTaskId("item"), "tool one", "in_progress"), ]); expectNativeProgressAppend(2, [ - planUpdate("Shelling"), taskUpdate(contentTaskId("item"), "tool one", "complete"), taskUpdate(contentTaskId("item"), "tool two", "complete"), taskUpdate(contentTaskId("item"), "tool three", "complete"), @@ -3766,10 +3756,8 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { planUpdate("Shelling"), taskUpdate(taskId, "bash", "in_progress", { details: "12345…uvwxyz" }), ]); - expectNativeProgressAppend(0, [ - planUpdate("Shelling"), - taskUpdate(taskId, "bash", "complete", { details: "12345…uvwxyz" }), - ]); + // Slack appends `details` per task_update; the unchanged command is not resent. + expectNativeProgressAppend(0, [taskUpdate(taskId, "bash", "complete")]); }); it("preserves patch item identity in native Slack progress task updates", async () => { @@ -3793,10 +3781,7 @@ describe("dispatchPreparedSlackMessage preview fallback", () => { planUpdate("updated Slack progress tests"), taskUpdate(taskId, "updated Slack progress tests", "in_progress"), ]); - expectNativeProgressAppend(0, [ - planUpdate("updated Slack progress tests"), - taskUpdate(taskId, "updated Slack progress tests", "complete"), - ]); + expectNativeProgressAppend(0, [taskUpdate(taskId, "updated Slack progress tests", "complete")]); }); it("preserves text Slack progress lines after a draft boundary status update", async () => { diff --git a/extensions/slack/src/progress-blocks.test.ts b/extensions/slack/src/progress-blocks.test.ts index 2a540b63b4b9..2e1777b75cdc 100644 --- a/extensions/slack/src/progress-blocks.test.ts +++ b/extensions/slack/src/progress-blocks.test.ts @@ -1,9 +1,11 @@ // Slack tests cover progress blocks plugin behavior. +import type { ChannelProgressDraftLine } from "openclaw/plugin-sdk/channel-outbound"; import { describe, expect, it } from "vitest"; import { buildSlackProgressCardBlocks, buildSlackProgressStreamCompletionChunks, buildSlackProgressStreamChunks, + EMPTY_SLACK_NATIVE_STREAM_SNAPSHOT, reconcileSlackNativeTaskChunks, } from "./progress-blocks.js"; @@ -209,7 +211,7 @@ describe("native Slack progress stream chunks", () => { it("terminalizes orphaned rows when a plan snapshot shrinks", () => { const first = reconcileSlackNativeTaskChunks({ - previousTasks: new Map(), + previous: EMPTY_SLACK_NATIVE_STREAM_SNAPSHOT, chunks: buildSlackProgressStreamChunks({ title: "Implementation", lines: [], @@ -221,7 +223,7 @@ describe("native Slack progress stream chunks", () => { }), }); const shrunk = reconcileSlackNativeTaskChunks({ - previousTasks: first.tasks, + previous: first.snapshot, chunks: buildSlackProgressStreamChunks({ title: "Implementation", lines: [], @@ -230,7 +232,6 @@ describe("native Slack progress stream chunks", () => { }); expect(shrunk.chunks).toEqual([ - planUpdate("Implementation"), taskUpdate("plan_step_1", "Inspect code", "in_progress"), taskUpdate("plan_step_2", "Patch code", "complete"), taskUpdate("plan_step_3", "Run tests", "complete"), @@ -239,13 +240,13 @@ describe("native Slack progress stream chunks", () => { it("terminalizes tool-line tasks when the source switches to a typed plan", () => { const lineChunks = reconcileSlackNativeTaskChunks({ - previousTasks: new Map(), + previous: EMPTY_SLACK_NATIVE_STREAM_SNAPSHOT, chunks: buildSlackProgressStreamChunks({ lines: [itemLine("run tests", "Running tests")], }), }); const planChunks = reconcileSlackNativeTaskChunks({ - previousTasks: lineChunks.tasks, + previous: lineChunks.snapshot, chunks: buildSlackProgressStreamChunks({ title: "Implementation", lines: [], @@ -261,19 +262,21 @@ describe("native Slack progress stream chunks", () => { it("keeps content-derived task ids stable when a rolling line window shifts", () => { const first = reconcileSlackNativeTaskChunks({ - previousTasks: new Map(), + previous: EMPTY_SLACK_NATIVE_STREAM_SNAPSHOT, chunks: buildSlackProgressStreamChunks({ lines: [itemLine("first task"), itemLine("shared task")], }), }); const shifted = reconcileSlackNativeTaskChunks({ - previousTasks: first.tasks, + previous: first.snapshot, chunks: buildSlackProgressStreamChunks({ lines: [itemLine("shared task"), itemLine("new task")], }), }); - const firstShared = [...first.tasks].find(([, task]) => task.title === "shared task"); - const shiftedShared = [...shifted.tasks].find(([, task]) => task.title === "shared task"); + const firstShared = [...first.snapshot.tasks].find(([, task]) => task.title === "shared task"); + const shiftedShared = [...shifted.snapshot.tasks].find( + ([, task]) => task.title === "shared task", + ); expect(firstShared?.[0]).toBeDefined(); expect(shiftedShared?.[0]).toBe(firstShared?.[0]); @@ -318,18 +321,65 @@ describe("native Slack progress stream chunks", () => { ); }); - it("keeps chunks untouched when no previous tasks are orphaned", () => { - const chunks = buildSlackProgressStreamChunks({ - title: "Implementation", - lines: [], - plan: [{ step: "Inspect code", status: "in_progress" }], + it("emits nothing when the snapshot matches what the stream already holds", () => { + const build = () => + buildSlackProgressStreamChunks({ + title: "Implementation", + lines: [], + plan: [{ step: "Inspect code", status: "in_progress" }], + }); + const first = reconcileSlackNativeTaskChunks({ + previous: EMPTY_SLACK_NATIVE_STREAM_SNAPSHOT, + chunks: build(), }); - const reconciled = reconcileSlackNativeTaskChunks({ - previousTasks: new Map([["plan_step_1", { title: "Inspect code", status: "in_progress" }]]), - chunks, + const repeated = reconcileSlackNativeTaskChunks({ previous: first.snapshot, chunks: build() }); + + expect(first.chunks).toEqual(build()); + expect(repeated.chunks).toBeUndefined(); + expect(repeated.snapshot).toEqual(first.snapshot); + }); + + it("streams task details and output as append-only deltas", () => { + // Slack concatenates details/output per task_update for the same id, so a + // resent field must carry only the unsent suffix. + const line = (status: string): ChannelProgressDraftLine => ({ + id: "call-1", + kind: "command-output", + label: "Bash", + detail: "pnpm test", + status, + text: `🛠️ Bash: pnpm test · ${status}`, + toolName: "bash", + }); + const first = reconcileSlackNativeTaskChunks({ + previous: EMPTY_SLACK_NATIVE_STREAM_SNAPSHOT, + chunks: buildSlackProgressStreamChunks({ title: "Shelling", lines: [line("running")] }), + }); + const repeated = reconcileSlackNativeTaskChunks({ + previous: first.snapshot, + chunks: buildSlackProgressStreamChunks({ title: "Shelling", lines: [line("running")] }), + }); + const failed = reconcileSlackNativeTaskChunks({ + previous: repeated.snapshot, + chunks: buildSlackProgressStreamChunks({ title: "Shelling", lines: [line("exit 1")] }), + }); + const finished = reconcileSlackNativeTaskChunks({ + previous: failed.snapshot, + chunks: buildSlackProgressStreamCompletionChunks({ + title: "Shelling", + lines: [line("exit 1")], + diffStat: { files: 2, added: 5, removed: 2 }, + }), }); - expect(reconciled.chunks).toEqual(chunks); + const taskId = expect.stringMatching(/^call_1_[a-f0-9]{8}$/u); + expect(first.chunks).toEqual([ + planUpdate("Shelling"), + taskUpdate(taskId, "bash", "in_progress", { details: "pnpm test" }), + ]); + expect(repeated.chunks).toBeUndefined(); + expect(failed.chunks).toEqual([taskUpdate(taskId, "bash", "error", { output: "exit 1" })]); + expect(finished.chunks).toEqual([taskUpdate(taskId, "bash", "error", { output: " · +5 −2" })]); }); it("starts native Slack progress with plan/task chunks instead of a static blocks plan", () => { @@ -410,7 +460,8 @@ describe("native Slack progress stream chunks", () => { details: "command finished", }), taskUpdate(contentTaskId("exec"), "exec", "error", { - details: "command failed · exit 1", + details: "command failed", + output: "exit 1", }), ]); }); @@ -564,11 +615,11 @@ describe("native Slack progress stream chunks", () => { ? running[1].id : undefined; expect(running?.[1]).toMatchObject({ id: expect.stringMatching(/^call_2_[a-f0-9]{8}$/u) }); - expect(completed?.[1]).toMatchObject({ + expect(completed?.[1]).toEqual({ + type: "task_update", id: runningTaskId, status: "complete", title: "bash", - details: "completed", }); }); @@ -608,10 +659,11 @@ describe("native Slack progress stream chunks", () => { ], }), ).toEqual([ - planUpdate("Exec — command failed · exit 1"), + planUpdate("Exec — command failed"), taskUpdate(contentTaskId("item"), "tool one", "complete"), taskUpdate(contentTaskId("command_output"), "Exec", "error", { - details: "command failed · exit 1", + details: "command failed", + output: "exit 1", }), ]); }); diff --git a/extensions/slack/src/progress-blocks.ts b/extensions/slack/src/progress-blocks.ts index 03a9834f0022..54e8b978f63a 100644 --- a/extensions/slack/src/progress-blocks.ts +++ b/extensions/slack/src/progress-blocks.ts @@ -12,6 +12,7 @@ import { SLACK_MAX_BLOCKS } from "./blocks-input.js"; import { normalizeSlackOutboundText } from "./format.js"; import { escapeSlackMrkdwn } from "./monitor/mrkdwn.js"; import { SLACK_SESSION_LINK_ACTION_ID } from "./reply-action-ids.js"; +import { applyAppendOnlyStreamUpdate } from "./stream-mode.js"; import { truncateSlackText } from "./truncate.js"; const SLACK_PROGRESS_FIELD_MAX = 1800; @@ -121,17 +122,25 @@ function lineTaskTitle(line: ChannelProgressDraftLine): string { return compactTitle(label); } +// Native task rows stream `details`/`output` append-only (see +// reconcileSlackNativeTaskChunks), so `details` holds only the stable tool +// detail and `output` the once-emitted result: a file delta or the failing +// terminal status. Non-terminal status words are already the row icon. function lineTaskDetails(line: ChannelProgressDraftLine, maxLineChars: number): string | undefined { - const detail = (lineDetailParts(line).join(" · ") || line.status?.trim()) + const detail = line.detail ?.replace(SLACK_PROGRESS_LINE_DELTA_RE, "") .replace(/\s+·\s*$/u, "") .trim(); - return detail ? compactDetail(detail, maxLineChars) : undefined; + return detail && detail !== line.status?.trim() ? compactDetail(detail, maxLineChars) : undefined; } function lineTaskOutput(line: ChannelProgressDraftLine): string | undefined { - const match = SLACK_PROGRESS_LINE_DELTA_RE.exec(lineDetailParts(line).join(" · ")); - return match ? `+${match[1]} −${match[2]}` : undefined; + const match = line.detail ? SLACK_PROGRESS_LINE_DELTA_RE.exec(line.detail) : null; + if (match) { + return `+${match[1]} −${match[2]}`; + } + const status = line.status?.replace(/\s+/g, " ").trim(); + return status && lineTaskStatus(line) === "error" ? status : undefined; } function lineTaskStatus(line: ChannelProgressDraftLine): SlackPlanTaskStatus { @@ -313,7 +322,7 @@ export function buildSlackProgressStreamChunks(params: { chunk.output = task.output; } if (index === finalTaskIndex && diffOutput) { - chunk.output = diffOutput; + chunk.output = [task.output, diffOutput].filter(Boolean).join(" · "); } if (index === finalTaskIndex && params.sessionUrl) { chunk.sources = buildSessionSources(params.sessionUrl); @@ -447,57 +456,138 @@ export function buildSlackProgressCardBlocks(params: { return blocks.slice(0, SLACK_MAX_BLOCKS); } -export type SlackNativeTaskSnapshot = ReadonlyMap< - string, - { title: string; status: SlackPlanTaskStatus } ->; +type SlackNativeStreamField = { rendered: string; source: string }; + +type SlackNativeTaskRow = { + title: string; + status: SlackPlanTaskStatus; + details?: SlackNativeStreamField; + output?: SlackNativeStreamField; + sourcesSent?: boolean; +}; + +/** Task rows and plan title already delivered to one native Slack stream. */ +export type SlackNativeStreamSnapshot = { + planTitle?: string; + tasks: ReadonlyMap; +}; + +export const EMPTY_SLACK_NATIVE_STREAM_SNAPSHOT: SlackNativeStreamSnapshot = { tasks: new Map() }; + +const SLACK_TASK_FIELD_SEPARATOR = " · "; + +// Slack appends `details`/`output` text per task_update for the same id +// (verified live 2026-08-17: two chunks with "detA"/"detB" rendered "detAdetB"); +// title/status replace. Each field therefore streams as append-only text: send +// only the unsent suffix, and join a divergent value with a separator. +function resolveTaskFieldDelta( + previous: SlackNativeStreamField | undefined, + incoming: string | undefined, +): { field: SlackNativeStreamField | undefined; delta: string | undefined } { + if (!incoming) { + return { field: previous, delta: undefined }; + } + // A restatement already visible in the row (write start "to a.ts (22 chars)" + // -> patch result "a.ts") adds nothing; joining it would only repeat the file. + if (previous?.rendered.includes(incoming)) { + return { field: { rendered: previous.rendered, source: incoming }, delta: undefined }; + } + const next = applyAppendOnlyStreamUpdate({ + incoming, + rendered: previous?.rendered ?? "", + source: previous?.source ?? "", + separator: SLACK_TASK_FIELD_SEPARATOR, + }); + const delta = next.changed ? next.rendered.slice(previous?.rendered.length ?? 0) : undefined; + return { field: { rendered: next.rendered, source: next.source }, delta }; +} /** - * Slack native streams key task rows by persistent id with no removal chunk. - * When the task source switches representation (tool lines <-> typed plan) or - * a snapshot drops ids, previously emitted non-terminal rows must receive a - * final update or they linger in_progress forever. + * Turns a full task snapshot into the delta Slack must receive. Native streams + * key rows by persistent id with no removal chunk: unchanged rows are omitted, + * changed rows carry only their unsent field text, and rows that dropped out + * (plan shrinks, tool-line <-> plan source switches) get a final complete + * update or they linger in_progress forever. */ export function reconcileSlackNativeTaskChunks(params: { - previousTasks: SlackNativeTaskSnapshot; + previous: SlackNativeStreamSnapshot; chunks: AnyChunk[] | undefined; -}): { chunks: AnyChunk[] | undefined; tasks: SlackNativeTaskSnapshot } { - const nextTasks = new Map(); +}): { chunks: AnyChunk[] | undefined; snapshot: SlackNativeStreamSnapshot } { + const nextTasks = new Map(); + let planTitle = params.previous.planTitle; + const emitted: AnyChunk[] = []; for (const chunk of params.chunks ?? []) { - if (chunk.type === "task_update") { - nextTasks.set(chunk.id, { - title: chunk.title, - status: chunk.status as SlackPlanTaskStatus, - }); + if (chunk.type === "plan_update") { + if (chunk.title !== planTitle) { + planTitle = chunk.title; + emitted.push(chunk); + } + continue; } - } - const orphaned = [...params.previousTasks].filter( - ([id, task]) => !nextTasks.has(id) && task.status !== "complete" && task.status !== "error", - ); - const terminalized = orphaned.map(([id, task]) => { - const entry = { title: task.title, status: "complete" as const }; - nextTasks.set(id, entry); - return { - type: "task_update" as const, - id, - title: task.title, - status: "complete" as const, + if (chunk.type !== "task_update") { + emitted.push(chunk); + continue; + } + const previousRow = params.previous.tasks.get(chunk.id); + const status = chunk.status as SlackPlanTaskStatus; + const details = resolveTaskFieldDelta(previousRow?.details, chunk.details); + const output = resolveTaskFieldDelta(previousRow?.output, chunk.output); + // The session source is a per-turn constant; deliver it once. + const sourcesChanged = Boolean(chunk.sources) && !previousRow?.sourcesSent; + const row: SlackNativeTaskRow = { title: chunk.title, status }; + if (details.field) { + row.details = details.field; + } + if (output.field) { + row.output = output.field; + } + if (sourcesChanged || previousRow?.sourcesSent) { + row.sourcesSent = true; + } + nextTasks.set(chunk.id, row); + const rowChanged = + !previousRow || + previousRow.title !== chunk.title || + previousRow.status !== status || + Boolean(details.delta) || + Boolean(output.delta) || + sourcesChanged; + if (!rowChanged) { + continue; + } + const update: TaskUpdateChunk = { + type: "task_update", + id: chunk.id, + title: chunk.title, + status, }; - }); - // Carry forward already-terminal rows so a later reappearance diffs correctly. - for (const [id, task] of params.previousTasks) { - if (!nextTasks.has(id)) { - nextTasks.set(id, task); + if (details.delta) { + update.details = details.delta; } + if (output.delta) { + update.output = output.delta; + } + if (sourcesChanged) { + update.sources = chunk.sources; + } + emitted.push(update); } - // An explicitly cleared source still needs its previous rows retired even - // when the current build produced no chunks of its own. - const chunks = params.chunks?.length - ? [...params.chunks, ...terminalized] - : terminalized.length - ? terminalized - : params.chunks; - return { chunks, tasks: nextTasks }; + for (const [id, row] of params.previous.tasks) { + if (nextTasks.has(id)) { + continue; + } + // Carry forward already-terminal rows so a later reappearance diffs correctly. + if (row.status === "complete" || row.status === "error") { + nextTasks.set(id, row); + continue; + } + nextTasks.set(id, { ...row, status: "complete" }); + emitted.push({ type: "task_update", id, title: row.title, status: "complete" }); + } + return { + chunks: emitted.length > 0 ? emitted : undefined, + snapshot: { ...(planTitle ? { planTitle } : {}), tasks: nextTasks }, + }; } export function buildSlackProgressStreamCompletionChunks(params: { diff --git a/extensions/slack/src/stream-mode.ts b/extensions/slack/src/stream-mode.ts index d48fc069d679..5b7ad866aaca 100644 --- a/extensions/slack/src/stream-mode.ts +++ b/extensions/slack/src/stream-mode.ts @@ -25,6 +25,8 @@ export function applyAppendOnlyStreamUpdate(params: { incoming: string; rendered: string; source: string; + /** Joins a divergent incoming value onto the already-rendered text. */ + separator?: string; }): { rendered: string; source: string; changed: boolean } { const incoming = params.incoming.trimEnd(); if (!incoming) { @@ -53,7 +55,7 @@ export function applyAppendOnlyStreamUpdate(params: { return { rendered: params.rendered, source: params.source, changed: false }; } - const separator = params.rendered.endsWith("\n") ? "" : "\n"; + const separator = params.separator ?? (params.rendered.endsWith("\n") ? "" : "\n"); return { rendered: `${params.rendered}${separator}${incoming}`, source: incoming,