diff --git a/cli/src/agent/backends/acp/AcpMessageHandler.test.ts b/cli/src/agent/backends/acp/AcpMessageHandler.test.ts index cc47e19c..5b20d136 100644 --- a/cli/src/agent/backends/acp/AcpMessageHandler.test.ts +++ b/cli/src/agent/backends/acp/AcpMessageHandler.test.ts @@ -2545,3 +2545,391 @@ describe('AcpMessageHandler', () => { clearGeneratedImages(); }); }); + +describe('AcpMessageHandler idle text flush', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + function textMessages(messages: AgentMessage[]): Array> { + return messages.filter((message): message is Extract => + message.type === 'text' + ); + } + + it('flushes buffered text after the idle interval without a boundary event', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'hello world' } + }); + + await vi.advanceTimersByTimeAsync(999); + expect(messages).toEqual([]); + + await vi.advanceTimersByTimeAsync(1); + expect(messages).toEqual([{ type: 'text', text: 'hello world', id: expect.any(String), streamSnapshot: true, live: true }]); + + await vi.advanceTimersByTimeAsync(10000); + expect(textMessages(messages)).toHaveLength(1); + }); + + it('postpones the idle flush while updates keep arriving', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'hello ' } + }); + + await vi.advanceTimersByTimeAsync(600); + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'world' } + }); + + await vi.advanceTimersByTimeAsync(600); + expect(messages).toEqual([]); + + await vi.advanceTimersByTimeAsync(400); + expect(messages).toEqual([{ type: 'text', text: 'hello world', id: expect.any(String), streamSnapshot: true, live: true }]); + }); + + it('finalizes an idled segment under the same stream id at the boundary', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'hello' } + }); + + await vi.advanceTimersByTimeAsync(1000); + const [snapshot] = textMessages(messages); + expect(snapshot).toEqual({ + type: 'text', + text: 'hello', + id: expect.any(String), + streamSnapshot: true, + live: true + }); + + handler.flushText(); + const text = textMessages(messages); + expect(text).toHaveLength(2); + expect(text[1]).toEqual({ + type: 'text', + text: 'hello', + id: snapshot.id, + streamSnapshot: true + }); + + handler.drainBuffers(); + expect(textMessages(messages)).toHaveLength(2); + }); + + it('cancels the idle timer when a tool call closes the text segment', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'hello' } + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.toolCall, + toolCallId: 'tool-1', + title: 'Read', + rawInput: { path: 'README.md' }, + status: 'in_progress' + }); + + expect(messages.map((m) => m.type)).toEqual(['text', 'tool_call']); + + await vi.advanceTimersByTimeAsync(5000); + expect(messages).toHaveLength(2); + }); + + it('dedupe mode streams full-text snapshots under one stream id', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'Hello world.' } + }); + await vi.advanceTimersByTimeAsync(1000); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'Hello world. And more.' } + }); + await vi.advanceTimersByTimeAsync(1000); + + handler.flushText(); + + const text = textMessages(messages); + expect(text).toHaveLength(3); + expect(text[0].text).toEqual('Hello world.'); + expect(text[1].text).toEqual('Hello world. And more.'); + expect(text[2].text).toEqual('Hello world. And more.'); + expect(text[0].id).toEqual(expect.any(String)); + expect(text[1].id).toEqual(text[0].id); + expect(text[2].id).toEqual(text[0].id); + expect(text[2].streamSnapshot).toBe(true); + }); + + it('stays silent when no text is buffered', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentThoughtChunk, + content: { type: 'text', text: 'thinking' } + }); + + await vi.advanceTimersByTimeAsync(10000); + expect(messages.filter((m) => m.type === 'text')).toEqual([]); + }); + + it('is disabled when textIdleFlushMs is zero', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 0 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'hello' } + }); + + await vi.advanceTimersByTimeAsync(10000); + expect(textMessages(messages)).toHaveLength(0); + + handler.flushText(); + expect(messages).toEqual([{ type: 'text', text: 'hello' }]); + }); + + it('re-asserts the same snapshot for a repeated cumulative chunk', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'abc' } + }); + await vi.advanceTimersByTimeAsync(1000); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'abc' } + }); + await vi.advanceTimersByTimeAsync(1000); + + const text = textMessages(messages); + expect(text).toHaveLength(2); + expect(text[0].id).toEqual(expect.any(String)); + expect(text[1].id).toEqual(text[0].id); + expect(text[1].text).toEqual('abc'); + }); + + it('settles a prepend-then-append rewrite on the full answer under one stream id', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'world' } + }); + await vi.advanceTimersByTimeAsync(1000); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'hello world' } + }); + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'hello world!' } + }); + await vi.advanceTimersByTimeAsync(1000); + + handler.flushText(); + + const text = textMessages(messages); + expect(text).toHaveLength(3); + const id = text[0].id; + expect(id).toEqual(expect.any(String)); + for (const message of text) { + expect(message.id).toEqual(id); + } + expect(text[0].text).toEqual('world'); + expect(text[1].text).toEqual('hello world!'); + expect(text[2].text).toEqual('hello world!'); + expect(text.some((message) => message.text === 'hello ')).toBe(false); + }); + + it('defers an incomplete internal envelope until a boundary flush drops it', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: '{"type":"ou' } + }); + await vi.advanceTimersByTimeAsync(1000); + expect(textMessages(messages)).toEqual([]); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'tput","data":{"parentUuid":null,"sessionId":"s","userType":"x"}}' } + }); + handler.drainBuffers(); + expect(textMessages(messages)).toEqual([]); + }); + + it('still flushes a complete JSON answer at idle', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: '{"a":1}' } + }); + await vi.advanceTimersByTimeAsync(1000); + expect(textMessages(messages)).toEqual([{ type: 'text', text: '{"a":1}', id: expect.any(String), streamSnapshot: true, live: true }]); + }); + + it('keeps reasoning above the answer when a trailing thought arrives after text', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'the answer' } + }); + await vi.advanceTimersByTimeAsync(100); + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentThoughtChunk, + content: { type: 'text', text: 'trailing thought' } + }); + + await vi.advanceTimersByTimeAsync(1000); + expect(textMessages(messages)).toEqual([]); + + handler.drainBuffers(); + + const ordered = messages.filter((message) => message.type === 'text' || message.type === 'reasoning'); + const reasoningIndex = ordered.findIndex( + (message) => message.type === 'reasoning' && message.text.includes('trailing thought') + ); + const textIndex = ordered.findIndex((message) => message.type === 'text'); + expect(reasoningIndex).toBeGreaterThanOrEqual(0); + expect(textIndex).toBeGreaterThan(reasoningIndex); + expect(ordered[textIndex]).toEqual({ type: 'text', text: 'the answer' }); + }); + + it('still flushes brace-leading text that cannot be a control envelope', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: '{"name":"hapi",' } + }); + + await vi.advanceTimersByTimeAsync(1000); + expect(textMessages(messages)).toEqual([{ type: 'text', text: '{"name":"hapi",', id: expect.any(String), streamSnapshot: true, live: true }]); + }); + + it('defers a control envelope whose data key arrives before type', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: '{"data":{"parentUuid":null,"sessionId":"s","userType":"x"},' } + }); + await vi.advanceTimersByTimeAsync(1000); + expect(textMessages(messages)).toEqual([]); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: '"type":"output"}' } + }); + await vi.advanceTimersByTimeAsync(1000); + expect(textMessages(messages)).toEqual([]); + + handler.drainBuffers(); + expect(textMessages(messages)).toEqual([]); + }); + + it('keeps one stream id across an idle gap inside an open code fence', async () => { + const messages: AgentMessage[] = []; + const handler = new AcpMessageHandler((message) => messages.push(message), { + textChunkMode: 'delta', + textIdleFlushMs: 1000 + }); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: '```js\nconst a = 1;\n' } + }); + await vi.advanceTimersByTimeAsync(1000); + + await handler.handleUpdate({ + sessionUpdate: ACP_SESSION_UPDATE_TYPES.agentMessageChunk, + content: { type: 'text', text: 'console.log(a);\n```' } + }); + await vi.advanceTimersByTimeAsync(1000); + + const text = textMessages(messages); + expect(text).toHaveLength(2); + expect(text[0].id).toEqual(expect.any(String)); + expect(text[1].id).toEqual(text[0].id); + expect(text[1].text).toEqual('```js\nconst a = 1;\nconsole.log(a);\n```'); + }); +}); diff --git a/cli/src/agent/backends/acp/AcpMessageHandler.ts b/cli/src/agent/backends/acp/AcpMessageHandler.ts index 69aaec32..6db23695 100644 --- a/cli/src/agent/backends/acp/AcpMessageHandler.ts +++ b/cli/src/agent/backends/acp/AcpMessageHandler.ts @@ -43,6 +43,13 @@ type DerivedToolName = ReturnType; const REASONING_SNAPSHOT_INTERVAL_MS = 250; +// ACP gives no "message finished" signal for a turn's last text segment. HAPI +// buffers assistant text and only emits it at explicit boundaries (tool_call / +// plan / image / rate-limit) or at turn-boundary drains; when neither arrives +// the segment would sit in the buffer until the next user prompt. This idle +// timer closes that gap without touching the turn lifecycle. +const TEXT_IDLE_FLUSH_INTERVAL_MS = 1000; + /** * Extracts _meta.kind from the first diff block in a content array. * Returns null when content is not an array, is empty, or the first block @@ -396,10 +403,72 @@ function getSuffixPrefixOverlap(base: string, next: string): number { return 0; } +// Control envelopes this codebase must never surface as visible assistant text +// (see isInternalEventJson / parseRateLimitText). They arrive in pieces - delta +// mode sends one fragment per chunk and cumulative dedupe can also hold a +// partial one - so an early idle flush cannot tell a partial envelope from a +// partial answer. A partial object stays deferred only while its first key (or +// its `type` value) can still turn out to be an envelope, so property order is +// not assumed and brace-leading text that is ruled out immediately (e.g. +// `{"name":...`) still flushes at idle. +const CONTROL_ENVELOPE_KEYS = ['data', 'rate_limit_info', 'type']; +const CONTROL_ENVELOPE_TYPES = ['output', 'rate_limit_event']; + +function isCompleteJsonObject(text: string): boolean { + try { + JSON.parse(text); + return true; + } catch { + return false; + } +} + +/** + * True when `text` may be a control envelope that has not been fully received + * yet. A complete object is never a candidate: a fully received envelope is + * filtered by isInternalEventJson / the rate-limit parser instead, and a + * genuine JSON answer must be able to flush at idle. For a partial object the + * answer is decided from the first key as far as it has arrived: once the key + * (or a `type` value) can no longer become an envelope shape, the text is + * ordinary content and may be emitted. Whitespace is ignored so leading + * indentation or embedded newlines do not hide the shape. The boundary flush + * is where a split envelope is reassembled and filtered. + */ +function mayBeUnfinishedControlEnvelope(text: string): boolean { + if (isCompleteJsonObject(text)) return false; + const compact = text.trimStart().replace(/\s+/g, ''); + if (!compact.startsWith('{')) return false; + const keyMatch = /^\{"([A-Za-z_][A-Za-z0-9_-]*)/.exec(compact); + if (!keyMatch) return true; + const key = keyMatch[1]; + if (compact[keyMatch[0].length] !== '"') { + return CONTROL_ENVELOPE_KEYS.some((candidate) => candidate.startsWith(key)); + } + if (!CONTROL_ENVELOPE_KEYS.includes(key)) return false; + if (key !== 'type') return true; + const afterKey = compact.slice(keyMatch[0].length + 1); + const valueMatch = /^:"([A-Za-z_]*)/.exec(afterKey); + if (!valueMatch) return true; + const value = valueMatch[1]; + if (afterKey[valueMatch[0].length] !== '"') { + return CONTROL_ENVELOPE_TYPES.some((candidate) => candidate.startsWith(value)); + } + return CONTROL_ENVELOPE_TYPES.includes(value); +} + export class AcpMessageHandler { private readonly toolCalls = new Map(); private acceptingUpdates = true; private bufferedText = ''; + // Stable stream id for a text segment that has already been emitted by an + // idle flush. Mirrors the reasoning snapshot scheme (see flushReasoning): + // the web reducer replaces a text block in place when a repeated stream id + // arrives, so later snapshots settle the segment without duplicating or + // reordering what is already rendered. Null until the first idle emission + // of the current segment; a boundary that never saw one keeps the upstream + // message shape (no id). + private textStreamId: string | null = null; + private textIdleTimer: ReturnType | null = null; // Array buffer avoids the O(N²) string concatenation that per-token // ACP streams (OpenCode/Zen emits one chunk per generated token) would // otherwise incur — a 10k-token reasoning trace allocates 10k full-buffer @@ -415,7 +484,7 @@ export class AcpMessageHandler { constructor( onMessage: (message: AgentMessage) => void, - private readonly options: { textChunkMode?: AcpTextChunkMode; flavor?: string } = {} + private readonly options: { textChunkMode?: AcpTextChunkMode; flavor?: string; textIdleFlushMs?: number } = {} ) { this.onMessage = (message) => { if (this.acceptingUpdates) onMessage(message); @@ -427,34 +496,76 @@ export class AcpMessageHandler { deactivate(): void { this.acceptingUpdates = false; this.bufferedText = ''; + this.textStreamId = null; + this.clearTextIdleFlushTimer(); this.bufferedReasoning = []; this.resetReasoningState(); } /** - * Emits any buffered assistant text as a single message and clears the - * buffer. Callers must treat this as a text-segment boundary: it is - * invoked internally before tool_call / plan events and externally at - * turn boundaries by AcpSdkBackend. + * Emits buffered assistant text. A true segment boundary clears the buffer + * and is invoked internally before tool_call / plan events and externally + * at turn boundaries by AcpSdkBackend; `mode: 'idle'` is the idle timer's + * opportunistic flush (see refreshTextIdleFlushTimer). * * The internal-event check in `handleUpdate` only sees one chunk at a - * time, so it cannot recognise an envelope that arrived in pieces — in + * time, so it cannot recognise an envelope that arrived in pieces - in * `delta` mode (OpenCode) every chunk is a fragment and none of them - * parses as JSON on its own. This flush boundary is the first place the - * reassembled text exists, so it is the only place a split envelope can - * be caught. Re-checking here is what makes the filter complete rather - * than merely likely to fire. + * parses as JSON on its own. This flush (or the boundary) is the first + * place the reassembled text exists, so it is the only place a split + * envelope can be caught. Re-checking here is what makes the filter + * complete rather than merely likely to fire. + * + * Idle emissions stream full-text snapshots under a stable id, matching + * flushReasoning: the reducer replaces a text block in place for a + * repeated stream id, so a pause inside an open Markdown fence does not + * split the rendered answer, and a dedupe rewrite settles on the full text + * instead of emitting a reordered prefix. The boundary then emits one + * final message with the same id, or the plain upstream shape when no idle + * emission happened. + * + * Idle flushes defer, keeping the buffer open, while a reasoning block is + * still open (the boundary drain exists to keep Reasoning above the + * answer - see drainBuffers) or while the buffer may still be a control + * envelope (see mayBeUnfinishedControlEnvelope). */ - flushText(): void { - if (!this.bufferedText) { + flushText(mode: 'boundary' | 'idle' = 'boundary'): void { + const text = this.bufferedText; + if (!text) { + if (mode === 'boundary') { + this.textStreamId = null; + this.clearTextIdleFlushTimer(); + } return; } - const text = this.bufferedText; + if (mode === 'idle') { + if (this.bufferedReasoning.length > 0) return; + if (mayBeUnfinishedControlEnvelope(text)) return; + // A complete envelope can only reach this point unchecked by the + // guard; never let one become visible text. + if (isInternalEventJson(text)) return; + if (!this.textStreamId) { + this.textStreamId = randomUUID(); + } + this.onMessage({ + type: 'text', + text, + id: this.textStreamId, + streamSnapshot: true, + live: true + }); + return; + } + const id = this.textStreamId; this.bufferedText = ''; + this.textStreamId = null; + this.clearTextIdleFlushTimer(); if (isInternalEventJson(text)) { return; } - this.onMessage({ type: 'text', text }); + this.onMessage(id + ? { type: 'text', text, id, streamSnapshot: true } + : { type: 'text', text }); } /** @@ -502,7 +613,39 @@ export class AcpMessageHandler { this.flushText(); } + private clearTextIdleFlushTimer(): void { + if (this.textIdleTimer !== null) { + clearTimeout(this.textIdleTimer); + this.textIdleTimer = null; + } + } + + /** + * (Re)arms the idle text flush. Called on every incoming update and after + * each appended text chunk so the timer only fires once the stream has been + * silent for a full interval. A non-positive `textIdleFlushMs` disables the + * timer entirely. + */ + private refreshTextIdleFlushTimer(): void { + this.clearTextIdleFlushTimer(); + const idleFlushMs = this.options.textIdleFlushMs ?? TEXT_IDLE_FLUSH_INTERVAL_MS; + if (idleFlushMs <= 0 || !this.bufferedText) { + return; + } + const timer = setTimeout(() => { + this.textIdleTimer = null; + this.flushText('idle'); + }, idleFlushMs); + timer.unref?.(); + this.textIdleTimer = timer; + } + private appendTextChunk(text: string): void { + this.appendTextChunkContent(text); + this.refreshTextIdleFlushTimer(); + } + + private appendTextChunkContent(text: string): void { if (this.textChunkMode === 'delta') { if (text) { this.bufferedText += text; @@ -599,6 +742,10 @@ export class AcpMessageHandler { const updateType = asString(update.sessionUpdate); if (!updateType) return; + // Any activity postpones the idle text flush so a segment is not split + // while the agent is still working. + this.refreshTextIdleFlushTimer(); + if (updateType === ACP_SESSION_UPDATE_TYPES.agentThoughtChunk) { // Thought chunks do not participate in intra-turn ordering and // must not flush the text buffer (that would split a live text diff --git a/cli/src/copilot/copilotRemoteLauncher.ts b/cli/src/copilot/copilotRemoteLauncher.ts index ebdb0304..c0a46d4c 100644 --- a/cli/src/copilot/copilotRemoteLauncher.ts +++ b/cli/src/copilot/copilotRemoteLauncher.ts @@ -240,6 +240,11 @@ export class CopilotRemoteLauncher extends RemoteLauncherBase { switch (message.type) { case 'text': + // Interim idle snapshots; the settled text arrives at the + // segment boundary, which is what belongs in the local log. + if (message.live) { + break; + } this.messageBuffer.addMessage(message.text, 'assistant'); break; case 'reasoning': diff --git a/cli/src/cursor/cursorAcpRemoteLauncher.ts b/cli/src/cursor/cursorAcpRemoteLauncher.ts index 8ac1c978..ce0ae085 100644 --- a/cli/src/cursor/cursorAcpRemoteLauncher.ts +++ b/cli/src/cursor/cursorAcpRemoteLauncher.ts @@ -866,6 +866,11 @@ class CursorAcpRemoteLauncher extends RemoteLauncherBase { switch (message.type) { case 'text': + // Interim idle snapshots; the settled text arrives at the + // segment boundary, which is what belongs in the local log. + if (message.live) { + break; + } this.messageBuffer.addMessage(message.text, 'assistant'); break; case 'reasoning': diff --git a/cli/src/dsh/dshRemoteLauncher.ts b/cli/src/dsh/dshRemoteLauncher.ts index d68c9687..50dcd3c6 100644 --- a/cli/src/dsh/dshRemoteLauncher.ts +++ b/cli/src/dsh/dshRemoteLauncher.ts @@ -211,6 +211,9 @@ export class DshRemoteLauncher extends RemoteLauncherBase { switch (message.type) { case 'text': + // Interim idle snapshots; the settled text arrives at the + // segment boundary, which is what belongs in the local log. + if (message.live) break this.messageBuffer.addMessage(message.text, 'assistant') break case 'generated_image': diff --git a/cli/src/grok/grokRemoteLauncher.ts b/cli/src/grok/grokRemoteLauncher.ts index fe591848..175b2697 100644 --- a/cli/src/grok/grokRemoteLauncher.ts +++ b/cli/src/grok/grokRemoteLauncher.ts @@ -389,6 +389,9 @@ class GrokRemoteLauncher extends RemoteLauncherBase { switch (message.type) { case 'text': + // Interim idle snapshots; the settled text arrives at the + // segment boundary, which is what belongs in the local log. + if (message.live) break this.messageBuffer.addMessage(message.text, 'assistant') break case 'reasoning': diff --git a/cli/src/kimi/kimiRemoteLauncher.ts b/cli/src/kimi/kimiRemoteLauncher.ts index f0c4f35f..7efab7a0 100644 --- a/cli/src/kimi/kimiRemoteLauncher.ts +++ b/cli/src/kimi/kimiRemoteLauncher.ts @@ -229,6 +229,11 @@ class KimiRemoteLauncher extends RemoteLauncherBase { switch (message.type) { case 'text': + // Interim idle snapshots; the settled text arrives at the + // segment boundary, which is what belongs in the local log. + if (message.live) { + break; + } this.messageBuffer.addMessage(message.text, 'assistant'); break; case 'reasoning': diff --git a/cli/src/opencode/opencodeRemoteLauncher.test.ts b/cli/src/opencode/opencodeRemoteLauncher.test.ts index b82c9ce9..435b45ad 100644 --- a/cli/src/opencode/opencodeRemoteLauncher.test.ts +++ b/cli/src/opencode/opencodeRemoteLauncher.test.ts @@ -1,5 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { MessageQueue2 } from '@/utils/MessageQueue2'; +import { MessageBuffer } from '@/ui/ink/messageBuffer'; import type { OpencodeMode, PermissionMode } from './types'; const harness = vi.hoisted(() => ({ @@ -23,6 +24,9 @@ const harness = vi.hoisted(() => ({ stderrHandler: null as null | ((error: { type: string; message: string; raw: string }) => void), hangPrompt: false, resolvePrompt: null as null | (() => void), + // The message callback backend.prompt() receives, so tests can emit + // AgentMessages (e.g. live vs settled text) without a real backend. + promptOnMessage: null as null | ((message: unknown) => void), cancelPrompt: vi.fn(async (_sessionId: string) => {}), // Lets a test take full manual control of when a given prompt() call // resolves, instead of the fixed-one-tick setImmediate delay below — @@ -117,7 +121,8 @@ vi.mock('./utils/opencodeBackend', () => ({ harness.thoughtLevelOption = { ...harness.thoughtLevelOption, currentValue: value }; } }), - prompt: vi.fn(async (_sessionId: string, content: unknown[]) => { + prompt: vi.fn(async (_sessionId: string, content: unknown[], onMessage?: (message: unknown) => void) => { + harness.promptOnMessage = onMessage ?? null; harness.promptContents.push(content); harness.events.push('prompt:start'); harness.promptCount++; @@ -502,6 +507,7 @@ describe('opencodeRemoteLauncher inline model switch', () => { compactHarness.triggerImpl = null; compactHarness.resultImpl = null; harness.promptImpl = null; + harness.promptOnMessage = null; harness.sessionModelsMetadata = undefined; harness.cancelPromptImpl = null; harness.newSessionImpl = null; @@ -2433,6 +2439,36 @@ describe('opencodeRemoteLauncher inline model switch', () => { 'prompt:end' ]); }); + + it('skips interim idle text snapshots in the local log while still forwarding them to the hub', async () => { + const addMessage = vi.spyOn(MessageBuffer.prototype, 'addMessage'); + const { session, agentMessages } = createSessionStub([ + { message: 'first', mode: createMode() } + ]); + + const launcherPromise = opencodeRemoteLauncher(session as never); + + while (!harness.events.includes('prompt:start')) { + await new Promise((resolve) => setImmediate(resolve)); + } + + const onMessage = harness.promptOnMessage; + if (!onMessage) { + throw new Error('prompt() was not called with a message callback'); + } + onMessage({ type: 'text', text: 'partial', id: 'stream-1', streamSnapshot: true, live: true }); + onMessage({ type: 'text', text: 'partial settled', id: 'stream-1', streamSnapshot: true }); + + // The idle snapshot must reach the hub (web reducer keeps one block per + // stream id) but must not duplicate the answer in the local terminal log. + expect(agentMessages).toContainEqual({ type: 'message', message: 'partial', id: 'stream-1', streamSnapshot: true }); + expect(agentMessages).toContainEqual({ type: 'message', message: 'partial settled', id: 'stream-1', streamSnapshot: true }); + expect(addMessage).not.toHaveBeenCalledWith('partial', 'assistant'); + expect(addMessage).toHaveBeenCalledWith('partial settled', 'assistant'); + + await launcherPromise; + addMessage.mockRestore(); + }); }); describe('selectAbortStatusMessage', () => { diff --git a/cli/src/opencode/opencodeRemoteLauncher.ts b/cli/src/opencode/opencodeRemoteLauncher.ts index 90831c11..43df1a67 100644 --- a/cli/src/opencode/opencodeRemoteLauncher.ts +++ b/cli/src/opencode/opencodeRemoteLauncher.ts @@ -1295,6 +1295,11 @@ class OpencodeRemoteLauncher extends RemoteLauncherBase { switch (message.type) { case 'text': + // Interim idle snapshots; the settled text arrives at the + // segment boundary, which is what belongs in the local log. + if (message.live) { + break; + } this.messageBuffer.addMessage(message.text, 'assistant'); break; case 'reasoning':