From fb86901af7ffed7f8017df9682586415aebfd403 Mon Sep 17 00:00:00 2001 From: sadfun Date: Sat, 22 Aug 2026 23:36:03 +0200 Subject: [PATCH 1/2] Replace thinking drafts with final replies --- src/channels/draft-stream.ts | 51 ++++++++++--------- test/draft-stream.test.ts | 95 ++++++++++++++++++++++++++++++++++++ 2 files changed, 120 insertions(+), 26 deletions(-) create mode 100644 test/draft-stream.test.ts diff --git a/src/channels/draft-stream.ts b/src/channels/draft-stream.ts index 75c3f0b..cc8fb6c 100644 --- a/src/channels/draft-stream.ts +++ b/src/channels/draft-stream.ts @@ -7,9 +7,9 @@ import { DraftThrottle } from "./draft-throttle.js"; import { formatThinkingBlock, splitMessageText } from "./progress.js"; /** - * Channel-agnostic streaming reply: one editable progress message whose edits - * are throttled to one per interval, followed by the final answer as separate - * messages. Connectors supply the transport (post/update) and the + * Channel-agnostic streaming reply: one editable message that starts as + * progress and is replaced by the final answer. Edits are throttled to one + * per interval. Connectors supply the transport (post/update) and the * channel-specific text rendering; all draft/lifecycle state lives here. */ export abstract class DraftReplyStream implements OutboundStream { @@ -144,22 +144,26 @@ export abstract class DraftReplyStream implements OutboundStream { } const chunks = text.length === 0 ? [] : splitMessageText(this.renderFinal(text), this.#textLimit); - // Freeze the progress message without the streaming cursor. The answer - // itself arrives as separate messages below: a silent edit of the - // thinking message never notifies anyone, and a failed edit must not - // take the answer down with it. + const [first, ...remaining] = chunks; + let chunksToPost = chunks; if (this.#messageId !== undefined) { - await this.update( - this.#messageId, - this.renderProgress(formatThinkingBlock(this.#progress)).slice(0, this.#textLimit), - ).catch((error: unknown) => { - this.logger.debug(`${this.#channelLabel} progress freeze failed`, { - error: errorMessage(error), - }); - }); + const replacement = + first ?? this.renderProgress(formatThinkingBlock(this.#progress)).slice(0, this.#textLimit); + if (replacement !== this.#lastPublishedText) { + try { + await this.update(this.#messageId, replacement); + chunksToPost = remaining; + } catch (error) { + this.logger.debug(`${this.#channelLabel} final replacement failed`, { + error: errorMessage(error), + }); + } + } else { + chunksToPost = remaining; + } } let undelivered = 0; - for (const chunk of chunks) { + for (const chunk of chunksToPost) { try { await this.post(chunk); } catch (error) { @@ -197,16 +201,11 @@ export abstract class DraftReplyStream implements OutboundStream { } private preview(): string { - // Rendering can expand the progress block past the message limit, so - // clamp it before budgeting the final-text tail. - const progress = this.renderProgress(formatThinkingBlock(this.#progress)).slice( - 0, - this.#textLimit - 3, - ); - if (this.#finalText.length === 0) return `${progress}\n\n▌`; - // available === 0 must not fall through to slice(-0), which is slice(0). - const available = Math.max(0, this.#textLimit - progress.length - 3); + if (this.#finalText.length === 0) { + return this.renderProgress(formatThinkingBlock(this.#progress)).slice(0, this.#textLimit); + } + const available = Math.max(0, this.#textLimit - 1); const finalText = available === 0 ? "" : this.renderFinal(this.#finalText).slice(-available); - return `${progress}\n\n${finalText}▌`; + return `${finalText}▌`; } } diff --git a/test/draft-stream.test.ts b/test/draft-stream.test.ts new file mode 100644 index 0000000..21e96c9 --- /dev/null +++ b/test/draft-stream.test.ts @@ -0,0 +1,95 @@ +import { afterEach, describe, expect, it, jest } from "bun:test"; +import { DraftReplyStream } from "../src/channels/draft-stream.js"; +import { Logger } from "../src/shared/logger.js"; + +class RecordingDraftStream extends DraftReplyStream { + public readonly posts: string[] = []; + public readonly updates: string[] = []; + public failUpdates = false; + + public constructor(limit = 100) { + super(new Logger("error"), "Test", limit); + } + + protected async postInitial(content: string): Promise { + this.posts.push(content); + return "message-1"; + } + + protected async post(content: string): Promise { + this.posts.push(content); + } + + protected async update(_messageId: string, content: string): Promise { + if (this.failUpdates) throw new Error("update failed"); + this.updates.push(content); + } + + protected renderProgress(block: string): string { + return block; + } + + protected renderFinal(text: string): string { + return text; + } +} + +afterEach(() => { + jest.useRealTimers(); +}); + +describe("DraftReplyStream", () => { + it("posts one thinking block and replaces it with the final answer", async () => { + const stream = new RecordingDraftStream(); + + await stream.start({ summary: "Thinking…", actions: [], plan: [] }); + expect(stream.posts).toEqual(["▌ Thinking…"]); + + await stream.complete("Finished"); + expect(stream.updates).toEqual(["Finished"]); + expect(stream.posts).toEqual(["▌ Thinking…"]); + }); + + it("replaces thinking with answer text while streaming", async () => { + jest.useFakeTimers(); + const stream = new RecordingDraftStream(); + + await stream.start(); + stream.appendFinal("Working answer"); + jest.advanceTimersByTime(1_500); + await Promise.resolve(); + + expect(stream.updates).toEqual(["Working answer▌"]); + }); + + it("posts only overflow chunks after replacing the thinking message", async () => { + const stream = new RecordingDraftStream(10); + + await stream.start(); + await stream.complete("1234567890abcdefghij"); + + expect(stream.updates).toEqual(["1234567890"]); + expect(stream.posts.slice(1)).toEqual(["abcdefghij"]); + }); + + it("posts the complete answer when replacing the thinking message fails", async () => { + const stream = new RecordingDraftStream(); + + await stream.start(); + stream.failUpdates = true; + await stream.complete("Finished"); + + expect(stream.updates).toEqual([]); + expect(stream.posts.at(-1)).toBe("Finished"); + }); + + it("publishes the latest progress when a turn ends without an answer", async () => { + const stream = new RecordingDraftStream(); + + await stream.start({ summary: "Thinking…", actions: [], plan: [] }); + stream.setProgress({ summary: "Waiting for input", actions: [], plan: [] }); + await stream.complete(""); + + expect(stream.updates).toEqual(["▌ Waiting for input"]); + }); +}); From 243bafae60e698238ecc6106f767c8bfe40aca09 Mon Sep 17 00:00:00 2001 From: sadfun Date: Sat, 22 Aug 2026 23:43:12 +0200 Subject: [PATCH 2/2] address greptile review feedback (greploop iteration 1) --- src/channels/draft-stream.ts | 17 ++++++----------- test/draft-stream.test.ts | 17 ++++++----------- 2 files changed, 12 insertions(+), 22 deletions(-) diff --git a/src/channels/draft-stream.ts b/src/channels/draft-stream.ts index cc8fb6c..96f860a 100644 --- a/src/channels/draft-stream.ts +++ b/src/channels/draft-stream.ts @@ -15,7 +15,6 @@ import { formatThinkingBlock, splitMessageText } from "./progress.js"; export abstract class DraftReplyStream implements OutboundStream { static readonly #draftIntervalMs = 1_500; #progress: ProgressSnapshot = { actions: [], plan: [] }; - #finalText = ""; #messageId: string | undefined; #starting: Promise | undefined; #lastPublishedText = ""; @@ -116,10 +115,11 @@ export abstract class DraftReplyStream implements OutboundStream { this.scheduleDraft(); } - public appendFinal(delta: string): void { - if (this.#lifecycle.closed) return; - this.#finalText += delta; - this.scheduleDraft(); + public appendFinal(_delta: string): void { + // Keep the editable message as an unambiguous progress placeholder until + // complete() can replace it with the authoritative final text. If a + // streamed preview were published first and the final edit failed, the + // delivery fallback would leave that stale preview beside the full answer. } public async complete( @@ -201,11 +201,6 @@ export abstract class DraftReplyStream implements OutboundStream { } private preview(): string { - if (this.#finalText.length === 0) { - return this.renderProgress(formatThinkingBlock(this.#progress)).slice(0, this.#textLimit); - } - const available = Math.max(0, this.#textLimit - 1); - const finalText = available === 0 ? "" : this.renderFinal(this.#finalText).slice(-available); - return `${finalText}▌`; + return this.renderProgress(formatThinkingBlock(this.#progress)).slice(0, this.#textLimit); } } diff --git a/test/draft-stream.test.ts b/test/draft-stream.test.ts index 21e96c9..dc0b6d2 100644 --- a/test/draft-stream.test.ts +++ b/test/draft-stream.test.ts @@ -1,4 +1,4 @@ -import { afterEach, describe, expect, it, jest } from "bun:test"; +import { describe, expect, it } from "bun:test"; import { DraftReplyStream } from "../src/channels/draft-stream.js"; import { Logger } from "../src/shared/logger.js"; @@ -34,10 +34,6 @@ class RecordingDraftStream extends DraftReplyStream { } } -afterEach(() => { - jest.useRealTimers(); -}); - describe("DraftReplyStream", () => { it("posts one thinking block and replaces it with the final answer", async () => { const stream = new RecordingDraftStream(); @@ -50,16 +46,14 @@ describe("DraftReplyStream", () => { expect(stream.posts).toEqual(["▌ Thinking…"]); }); - it("replaces thinking with answer text while streaming", async () => { - jest.useFakeTimers(); + it("keeps the thinking placeholder until the complete answer is available", async () => { const stream = new RecordingDraftStream(); await stream.start(); stream.appendFinal("Working answer"); - jest.advanceTimersByTime(1_500); - await Promise.resolve(); - expect(stream.updates).toEqual(["Working answer▌"]); + expect(stream.posts).toEqual(["▌ Thinking…"]); + expect(stream.updates).toEqual([]); }); it("posts only overflow chunks after replacing the thinking message", async () => { @@ -76,11 +70,12 @@ describe("DraftReplyStream", () => { const stream = new RecordingDraftStream(); await stream.start(); + stream.appendFinal("Partial preview"); stream.failUpdates = true; await stream.complete("Finished"); expect(stream.updates).toEqual([]); - expect(stream.posts.at(-1)).toBe("Finished"); + expect(stream.posts).toEqual(["▌ Thinking…", "Finished"]); }); it("publishes the latest progress when a turn ends without an answer", async () => {