mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-27 12:56:01 -06:00
fix(slack): stream native task rows as deltas (#125168)
Slack appends task_update details/output per update for the same id (only title/status replace); the native progress stream re-sent every row each snapshot, so bash rows accumulated "completedcompleted…". Reconcile now emits only changed rows with append-only field deltas, keeps status words out of details, and puts the once-emitted result (file delta or failing exit) in output.
This commit is contained in:
Ayaan Zaidi
@@ -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"}}
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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 () => {
|
||||
|
||||
@@ -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",
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
@@ -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<string, SlackNativeTaskRow>;
|
||||
};
|
||||
|
||||
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<string, { title: string; status: SlackPlanTaskStatus }>();
|
||||
}): { chunks: AnyChunk[] | undefined; snapshot: SlackNativeStreamSnapshot } {
|
||||
const nextTasks = new Map<string, SlackNativeTaskRow>();
|
||||
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: {
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user