/** * Tests channel streaming helper lifecycle and event forwarding. */ import { afterEach, describe, expect, it, vi } from "vitest"; import { buildChannelProgressDraftLine, createChannelProgressDraftGate, DEFAULT_PROGRESS_DRAFT_LABELS, formatChannelProgressDraftLine, formatChannelProgressDraftLineForEntry, formatChannelProgressDraftText, getChannelStreamingConfigObject, isChannelProgressDraftWorkToolName, isPotentialTruncatedFinal, mergeChannelProgressDraftLine, resolveChannelPreviewStreamMode, resolveChannelProgressDraftMaxLineChars, resolveChannelProgressDraftLabel, resolveChannelProgressDraftMaxLines, resolveChannelProgressDraftRender, resolveChannelStreamingBlockCoalesce, resolveChannelStreamingBlockEnabled, resolveChannelStreamingChunkMode, resolveChannelStreamingNativeTransport, resolveChannelStreamingProgressCommentary, resolveChannelStreamingPreviewCommandText, resolveChannelStreamingPreviewChunk, resolveChannelStreamingSuppressDefaultToolProgressMessages, resolveChannelStreamingPreviewToolProgress, resolveTranscriptBackedChannelFinalText, selectLongerFinalText, } from "./channel-streaming.js"; describe("channel-streaming", () => { afterEach(() => { vi.useRealTimers(); }); it("reads canonical nested streaming config first", () => { const entry = { streaming: { chunkMode: "newline", nativeTransport: true, block: { enabled: true, coalesce: { minChars: 40, maxChars: 80, idleMs: 250 }, }, preview: { chunk: { minChars: 10, maxChars: 20, breakPreference: "sentence" }, toolProgress: false, commandText: "status", }, }, chunkMode: "length", blockStreaming: false, nativeStreaming: false, blockStreamingCoalesce: { minChars: 5, maxChars: 15, idleMs: 100 }, draftChunk: { minChars: 2, maxChars: 4, breakPreference: "paragraph" }, } as const; expect(getChannelStreamingConfigObject(entry)).toEqual(entry.streaming); expect(resolveChannelStreamingChunkMode(entry)).toBe("newline"); expect(resolveChannelStreamingNativeTransport(entry)).toBe(true); expect(resolveChannelStreamingBlockEnabled(entry)).toBe(true); expect(resolveChannelStreamingBlockCoalesce(entry)).toEqual({ minChars: 40, maxChars: 80, idleMs: 250, }); expect(resolveChannelStreamingPreviewChunk(entry)).toEqual({ minChars: 10, maxChars: 20, breakPreference: "sentence", }); expect(resolveChannelStreamingPreviewToolProgress(entry)).toBe(false); expect(resolveChannelStreamingPreviewCommandText(entry)).toBe("status"); }); it("keeps progress-only tool progress config out of normal preview modes", () => { expect( resolveChannelStreamingPreviewToolProgress({ streaming: { mode: "partial", progress: { toolProgress: false } }, }), ).toBe(true); expect( resolveChannelStreamingPreviewToolProgress({ streaming: { mode: "block", preview: { toolProgress: true }, progress: { toolProgress: false }, }, }), ).toBe(true); expect( resolveChannelStreamingPreviewToolProgress({ streaming: { mode: "progress", preview: { toolProgress: true }, progress: { toolProgress: false }, }, }), ).toBe(false); }); it("enables commentary progress only for progress-mode drafts", () => { expect( resolveChannelStreamingProgressCommentary({ streaming: { mode: "progress", progress: { commentary: true } }, }), ).toBe(true); expect( resolveChannelStreamingProgressCommentary({ streaming: { mode: "partial", progress: { commentary: true } }, }), ).toBe(false); expect( resolveChannelStreamingProgressCommentary({ streaming: { mode: "progress" }, }), ).toBe(false); }); it("falls back to legacy flat fields when the canonical object is absent", () => { const entry = { chunkMode: "newline", blockStreaming: true, nativeStreaming: true, blockStreamingCoalesce: { minChars: 120, maxChars: 240, idleMs: 500 }, draftChunk: { minChars: 8, maxChars: 16, breakPreference: "newline" }, } as const; expect(getChannelStreamingConfigObject(entry)).toBeUndefined(); expect(resolveChannelStreamingChunkMode(entry)).toBe("newline"); expect(resolveChannelStreamingNativeTransport(entry)).toBe(true); expect(resolveChannelStreamingBlockEnabled(entry)).toBe(true); expect(resolveChannelStreamingBlockCoalesce(entry)).toEqual({ minChars: 120, maxChars: 240, idleMs: 500, }); expect(resolveChannelStreamingPreviewChunk(entry)).toEqual({ minChars: 8, maxChars: 16, breakPreference: "newline", }); expect(resolveChannelStreamingPreviewToolProgress(entry)).toBe(true); }); it("preserves progress as a first-class preview mode", () => { expect(resolveChannelPreviewStreamMode({ streaming: "progress" }, "off")).toBe("progress"); expect(resolveChannelPreviewStreamMode({ streaming: { mode: "progress" } }, "off")).toBe( "progress", ); }); it("keeps block preview mode separate from block delivery", () => { expect(resolveChannelStreamingBlockEnabled({ streaming: "block" })).toBeUndefined(); expect(resolveChannelStreamingBlockEnabled({ streaming: { mode: "block" } })).toBeUndefined(); expect( resolveChannelStreamingBlockEnabled({ streaming: { mode: "block", block: { enabled: true } }, }), ).toBe(true); expect(resolveChannelStreamingBlockEnabled({ streaming: "block", blockStreaming: false })).toBe( false, ); }); it("selects a longer transcript candidate for ellipsis-truncated finals", async () => { const fullAnswer = "Here is the complete final answer with enough stable prefix text before the ellipsis and enough continuation text after it."; const truncatedFinal = "Here is the complete final answer with enough stable prefix text before the ellipsis..."; expect(isPotentialTruncatedFinal(truncatedFinal)).toBe(true); expect( selectLongerFinalText({ finalText: truncatedFinal, candidateTexts: ["short", fullAnswer], }), ).toBe(fullAnswer); await expect( resolveTranscriptBackedChannelFinalText({ finalText: truncatedFinal, resolveCandidateText: async () => fullAnswer, }), ).resolves.toBe(fullAnswer); }); it("keeps intentional ellipsis finals when candidates do not prove truncation", async () => { const finalText = "Here is the complete final answer with enough stable prefix text before an intentional pause..."; const candidateText = "Here is the complete final answer with enough stable prefix text before an intentional pause... then punctuation"; expect( selectLongerFinalText({ finalText, candidateTexts: [candidateText], }), ).toBeUndefined(); await expect( resolveTranscriptBackedChannelFinalText({ finalText, resolveCandidateText: async () => candidateText, }), ).resolves.toBe(finalText); }); it("suppresses standalone tool progress for active preview drafts", () => { expect( resolveChannelStreamingSuppressDefaultToolProgressMessages({ streaming: { mode: "progress", progress: { toolProgress: false } }, }), ).toBe(true); expect( resolveChannelStreamingSuppressDefaultToolProgressMessages( { streaming: { mode: "partial", preview: { toolProgress: false } } }, { draftStreamActive: true }, ), ).toBe(true); expect( resolveChannelStreamingSuppressDefaultToolProgressMessages( { streaming: { mode: "partial", preview: { toolProgress: false } } }, { draftStreamActive: true, previewToolProgressEnabled: true }, ), ).toBe(true); expect( resolveChannelStreamingSuppressDefaultToolProgressMessages( { streaming: { mode: "progress" } }, { draftStreamActive: false }, ), ).toBe(false); }); it("uses auto progress labels when no explicit label is configured", () => { expect(DEFAULT_PROGRESS_DRAFT_LABELS[0]).toBe("Working"); expect(resolveChannelProgressDraftLabel({ random: () => 0 })).toBe( DEFAULT_PROGRESS_DRAFT_LABELS[0], ); expect(resolveChannelProgressDraftLabel({ random: () => 0.99 })).toBe( DEFAULT_PROGRESS_DRAFT_LABELS.at(-1), ); expect( resolveChannelProgressDraftLabel({ entry: { streaming: { progress: { label: " AUTO " } } }, random: () => 0, }), ).toBe(DEFAULT_PROGRESS_DRAFT_LABELS[0]); }); it("separates progress labels from detail lines with a blank line", () => { const entry = { streaming: { progress: { label: "Working" } } }; expect( formatChannelProgressDraftText({ entry, lines: ["🛠️ pgrep -fl Discord || true (agent)", "Discord is installed."], }), ).toBe("Working\n\n🛠️ pgrep -fl Discord || true (agent)\n• Discord is installed."); }); it("supports explicit progress labels and custom label sets", () => { expect( resolveChannelProgressDraftLabel({ entry: { streaming: { progress: { label: "Crunching" } } }, }), ).toBe("Crunching"); expect( resolveChannelProgressDraftLabel({ entry: { streaming: { progress: { labels: ["Pearling"] } } }, random: () => 0.5, }), ).toBe("Pearling"); expect( resolveChannelProgressDraftLabel({ entry: { streaming: { progress: { label: false } } }, }), ).toBeUndefined(); }); it("formats bounded progress draft text", () => { const entry = { streaming: { progress: { label: "Shelling", maxLines: 2, maxLineChars: 80, render: "rich" } }, }; expect(resolveChannelProgressDraftMaxLines(entry)).toBe(2); expect(resolveChannelProgressDraftMaxLineChars(entry)).toBe(80); expect(resolveChannelProgressDraftRender(entry)).toBe("rich"); expect( formatChannelProgressDraftText({ entry, lines: [" tool: read ", "patch applied", "tests done"], formatLine: (line) => `\`${line}\``, }), ).toBe("• `patch applied`\n• `tests done`"); expect( formatChannelProgressDraftText({ entry, lines: ["🛠️ Exec", "plain update"], }), ).toBe("🛠️ Exec\n• plain update"); expect( formatChannelProgressDraftText({ entry: { streaming: { progress: { label: false } } }, lines: [ { kind: "item", text: "_Checking source data before summarizing._", label: "Commentary", prefix: false, }, ], }), ).toBe("_Checking source data before summarizing._"); }); it("renders progress labels as rolling lines", () => { const entry = { streaming: { progress: { label: "Shelling", maxLines: 3 } } }; expect( formatChannelProgressDraftText({ entry, lines: ["🛠️ Exec", "📖 Read", "🩹 Patch"], }), ).toBe("🛠️ Exec\n📖 Read\n🩹 Patch"); }); it("renders structured progress lines with compact details", () => { const line = buildChannelProgressDraftLine({ event: "patch", summary: "1 modified", modified: ["extensions/discord/src/monitor/message-handler.draft-preview.ts"], }); expect( formatChannelProgressDraftText({ entry: { streaming: { progress: { label: false } } }, lines: line ? [line] : [], }), ).toBe("🩹 1 modified; extensions/discord/src/monitor/message-handler.draft-preview.ts"); }); it("bounds progress draft line length to reduce edit reflow", () => { expect( formatChannelProgressDraftText({ entry: { streaming: { progress: { label: "Shelling" } } }, lines: ["x".repeat(160)], formatLine: (line) => `\`${line}\``, }), ).toBe(`Shelling\n\n• \`${"x".repeat(119)}…\``); }); it("honors configured progress draft line length and cuts prose on word boundaries", () => { expect( formatChannelProgressDraftText({ entry: { streaming: { progress: { label: "Shelling", maxLineChars: 64 } } }, lines: [ "I'm checking whether the generated video exists or if the generator bailed while writing output.", ], }), ).toBe("Shelling\n\n• I'm checking whether the generated video exists or if the…"); }); it("falls back to plain commentary when compaction drops the closing italic marker", () => { expect( formatChannelProgressDraftText({ entry: { streaming: { progress: { label: false, maxLineChars: 32 } } }, lines: [ { kind: "item", text: `_${"x".repeat(80)}_`, label: "Commentary", prefix: false, }, ], }), ).toBe(`${"x".repeat(30)}…`); }); it("keeps compacted raw progress lines from leaking unmatched markdown backticks", () => { const line = buildChannelProgressDraftLine( { event: "tool", name: "exec", args: { command: "node scripts/check-something-with-a-very-long-path /tmp/openclaw/some/really/deep/path/that/keeps/going/and/going/index.ts --flag value", }, }, { detailMode: "raw" }, ); const text = formatChannelProgressDraftText({ entry: { streaming: { progress: { label: "Shelling" } } }, lines: line ? [line] : [], }); expect(text).toBe( "Shelling\n\n🛠️ run node script…e…y/deep/path/that/keeps/going/and/going/index.ts --flag value", ); expect(text.match(/`/g) ?? []).toHaveLength(0); }); it("formats progress draft lines with shared tool display labels", () => { const progressLine = buildChannelProgressDraftLine({ event: "tool", name: "write", args: { path: "/tmp/demo/index.html" }, }); if (!progressLine) { throw new Error("expected tool progress draft line"); } expect(progressLine.kind).toBe("tool"); expect(progressLine.icon).toBe("✍️"); expect(progressLine.label).toBe("Write"); expect(progressLine.detail).toBe("to /tmp/demo/index.html"); expect(progressLine.text).toBe("✍️ Write: to /tmp/demo/index.html"); expect(progressLine.toolName).toBe("write"); expect( formatChannelProgressDraftLine({ event: "tool", name: "write", args: { path: "/tmp/demo/index.html" }, }), ).toBe("✍️ Write: to /tmp/demo/index.html"); expect( formatChannelProgressDraftLine({ event: "item", itemKind: "tool", name: "write", meta: "/tmp/demo/style.css", }), ).toBe("✍️ Write: /tmp/demo/style.css"); expect( formatChannelProgressDraftLine({ event: "item", itemKind: "status", title: "Fast", summary: "💨Fast: auto-off(75s>=60s)", }), ).toBe("💨Fast: auto-off(75s>=60s)"); expect( formatChannelProgressDraftLine({ event: "patch", modified: ["/tmp/demo/index.html", "/tmp/demo/style.css"], }), ).toBe("🩹 Apply Patch: /tmp/demo/{index.html, style.css}"); expect( formatChannelProgressDraftLine( { event: "tool", name: "exec", args: { command: "pnpm test -- --watch=false" }, }, { detailMode: "raw" }, ), ).toBe("🛠️ run tests, `pnpm test -- --watch=false`"); expect( formatChannelProgressDraftLine({ event: "tool", name: "bash", args: { command: "sed -n '1,80p' extensions/discord/src/draft-stream.ts" }, }), ).toBe("🛠️ print lines 1-80 from extensions/discord/src/draft-stream.ts"); expect( formatChannelProgressDraftLine({ event: "tool", name: "web_search", args: { search_query: [{ q: "Codex OAuth API key" }], response_length: "short" }, }), ).toBe('🔎 Web Search: for "Codex OAuth API key"'); expect( formatChannelProgressDraftLine({ event: "item", itemKind: "command", name: "exec", progressText: "raw command output", }), ).toBe("🛠️ raw command output"); expect( formatChannelProgressDraftLine( { event: "item", itemKind: "command", name: "exec", progressText: "raw command output", }, { commandText: "status" }, ), ).toBe("🛠️ Exec"); expect( formatChannelProgressDraftLine( { event: "tool", name: "exec", args: { command: "pnpm test" }, }, { detailMode: "raw", commandText: "status" }, ), ).toBe("🛠️ Exec"); expect( formatChannelProgressDraftLineForEntry( { streaming: { preview: { commandText: "status" } } }, { event: "item", itemKind: "command", name: "exec", progressText: "raw command output", }, ), ).toBe("🛠️ Exec"); expect( formatChannelProgressDraftLine({ event: "item", itemKind: "analysis", title: "Reasoning", }), ).toBeUndefined(); expect( formatChannelProgressDraftLine({ event: "item", itemKind: "analysis", title: "Reasoning", progressText: "Reading the code path", }), ).toBe("Reading the code path"); }); it("updates keyed progress lines in place", () => { const first = buildChannelProgressDraftLine({ event: "item", itemId: "preamble-1", itemKind: "preamble", title: "Preamble", progressText: "Checking the", }); const second = buildChannelProgressDraftLine({ event: "item", itemId: "preamble-1", itemKind: "preamble", title: "Preamble", progressText: "Checking the app-server stream", }); if (!first || !second) { throw new Error("expected preamble progress lines"); } const initialLines: Array = ["🛠️ Exec"]; const lines = mergeChannelProgressDraftLine(initialLines, first, { maxLines: 4 }); const updated = mergeChannelProgressDraftLine(lines, second, { maxLines: 4 }); expect(updated).toHaveLength(2); expect(updated.at(-1)).toMatchObject({ id: "preamble-1", text: "Checking the app-server stream", }); expect( formatChannelProgressDraftText({ lines: updated, entry: { streaming: { progress: { label: false } } }, }), ).toBe("🛠️ Exec\n• Checking the app-server stream"); }); it("keeps public command progress ids while replacing by command correlation", () => { const toolLine = buildChannelProgressDraftLine({ event: "tool", itemId: "tool:call-1", toolCallId: "call-1", name: "bash", phase: "start", }); const commandLine = buildChannelProgressDraftLine({ event: "command-output", itemId: "tool:call-1-output", toolCallId: "call-1", name: "bash", phase: "end", exitCode: 0, }); const itemLine = buildChannelProgressDraftLine({ event: "item", itemId: "tool:call-1", toolCallId: "call-1", itemKind: "command", name: "bash", phase: "update", progressText: "install dependencies", }); expect(toolLine).toMatchObject({ id: "tool:call-1", kind: "tool", toolName: "bash" }); expect(itemLine).toMatchObject({ id: "tool:call-1", kind: "item", toolName: "bash" }); expect(commandLine).toMatchObject({ id: "tool:call-1-output", kind: "command-output", status: "completed", toolName: "bash", }); if (!toolLine || !itemLine || !commandLine) { throw new Error("expected command progress lines"); } const updated = [itemLine, commandLine].reduce( (lines, line) => mergeChannelProgressDraftLine(lines, line, { maxLines: 4 }), mergeChannelProgressDraftLine([], toolLine, { maxLines: 4 }), ); expect(updated).toHaveLength(1); expect(updated[0]).toMatchObject({ id: "tool:call-1-output", kind: "command-output", detail: "install dependencies", status: "completed", text: "🛠️ install dependencies", }); expect( formatChannelProgressDraftText({ lines: updated, entry: { streaming: { progress: { label: false } } }, }), ).toBe("🛠️ install dependencies"); const recoveredItemLine = buildChannelProgressDraftLine({ event: "item", itemId: "command-2", itemKind: "command", name: "bash", phase: "end", status: "failed", progressText: "install dependencies failed", }); const recoveredCommandLine = buildChannelProgressDraftLine({ event: "command-output", itemId: "command-2", toolCallId: "call-2", name: "bash", phase: "end", exitCode: 0, }); if (!recoveredItemLine || !recoveredCommandLine) { throw new Error("expected recovered command progress lines"); } const recoveredUpdated = mergeChannelProgressDraftLine( [recoveredItemLine], recoveredCommandLine, { maxLines: 4 }, ); expect(recoveredUpdated).toMatchObject([ { id: "command-2", kind: "command-output", status: "completed", text: "🛠️ Bash", }, ]); expect(recoveredUpdated[0]).not.toHaveProperty("detail"); }); it("starts progress drafts after five seconds or a second work event", async () => { vi.useFakeTimers(); const onStart = vi.fn(async () => {}); const gate = createChannelProgressDraftGate({ onStart }); await expect(gate.noteWork()).resolves.toBe(false); expect(onStart).not.toHaveBeenCalled(); await vi.advanceTimersByTimeAsync(4_999); expect(onStart).not.toHaveBeenCalled(); await vi.advanceTimersByTimeAsync(1); expect(onStart).toHaveBeenCalledTimes(1); expect(gate.hasStarted).toBe(true); }); it("starts progress drafts immediately on the second work event", async () => { vi.useFakeTimers(); const onStart = vi.fn(async () => {}); const gate = createChannelProgressDraftGate({ onStart }); await gate.noteWork(); await expect(gate.noteWork()).resolves.toBe(true); expect(onStart).toHaveBeenCalledTimes(1); await vi.advanceTimersByTimeAsync(5_000); expect(onStart).toHaveBeenCalledTimes(1); }); it("does not report started when delayed progress startup rejects", async () => { vi.useFakeTimers(); const error = new Error("draft unavailable"); const onStart = vi .fn<() => Promise>() .mockRejectedValueOnce(error) .mockResolvedValueOnce(undefined); const onStartError = vi.fn(); const gate = createChannelProgressDraftGate({ onStart, onStartError }); await expect(gate.noteWork()).resolves.toBe(false); await vi.advanceTimersByTimeAsync(5_000); expect(onStart).toHaveBeenCalledTimes(1); expect(gate.hasStarted).toBe(false); expect(onStartError).toHaveBeenCalledWith(error); await expect(gate.noteWork()).resolves.toBe(true); expect(onStart).toHaveBeenCalledTimes(2); expect(gate.hasStarted).toBe(true); }); it("keeps concurrent progress startup single-flight until onStart resolves", async () => { vi.useFakeTimers(); let resolveStart: (() => void) | undefined; const onStart = vi.fn( () => new Promise((resolve) => { resolveStart = resolve; }), ); const gate = createChannelProgressDraftGate({ onStart }); await gate.noteWork(); const firstStart = gate.noteWork(); const secondStart = gate.startNow(); await Promise.resolve(); expect(onStart).toHaveBeenCalledTimes(1); expect(gate.hasStarted).toBe(true); resolveStart?.(); await expect(firstStart).resolves.toBe(true); await expect(secondStart).resolves.toBeUndefined(); expect(onStart).toHaveBeenCalledTimes(1); expect(gate.hasStarted).toBe(true); }); it("does not report active when cancel wins the startup race", async () => { vi.useFakeTimers(); let resolveStart: (() => void) | undefined; const onStart = vi.fn( () => new Promise((resolve) => { resolveStart = resolve; }), ); const gate = createChannelProgressDraftGate({ onStart }); await gate.noteWork(); const startResult = gate.noteWork(); await Promise.resolve(); expect(onStart).toHaveBeenCalledTimes(1); gate.cancel(); resolveStart?.(); await expect(startResult).resolves.toBe(false); expect(gate.hasStarted).toBe(false); }); it("joins explicit startup before applying the first-work delay", async () => { vi.useFakeTimers(); let resolveStart: (() => void) | undefined; const onStart = vi.fn( () => new Promise((resolve) => { resolveStart = resolve; }), ); const gate = createChannelProgressDraftGate({ onStart }); const explicitStart = gate.startNow(); await Promise.resolve(); const workDuringStart = gate.noteWork(); expect(onStart).toHaveBeenCalledTimes(1); expect(gate.hasStarted).toBe(true); resolveStart?.(); await expect(explicitStart).resolves.toBeUndefined(); await expect(workDuringStart).resolves.toBe(true); expect(onStart).toHaveBeenCalledTimes(1); expect(gate.hasStarted).toBe(true); }); it("ignores message-like tools for progress draft work", () => { expect(isChannelProgressDraftWorkToolName("message")).toBe(false); expect(isChannelProgressDraftWorkToolName("react")).toBe(false); expect(isChannelProgressDraftWorkToolName("web_search")).toBe(true); expect(isChannelProgressDraftWorkToolName("exec")).toBe(true); }); });