From 1142c41ee0b49d5d00fa88ec33b2feb0ef1d71d0 Mon Sep 17 00:00:00 2001 From: alari Date: Sun, 14 Jun 2026 11:15:03 +0300 Subject: [PATCH] fix: stop OpenCode reasoning leaking into response text Some providers (e.g. Kimi via ollama-cloud) stream both reasoning and the visible answer as field=text deltas distinguished only by partID, and never emit message.part.updated. The old inReasoningPhase flag was only set from part.updated type=reasoning, so it never fired and the model's chain-of-thought was emitted as visible response text. Classify each partID via a cached REST lookup, buffering a part's deltas until its kind is known, then routing reasoning to a hidden thinking summary and text to the transcript. message.part.updated now also populates the cache so providers that send it skip the lookup. Co-Authored-By: Claude Opus 4 --- server/opencode-process.test.ts | 42 +++++++ server/opencode-process.ts | 189 ++++++++++++++++++++++++-------- 2 files changed, 188 insertions(+), 43 deletions(-) diff --git a/server/opencode-process.test.ts b/server/opencode-process.test.ts index 7af23028..fc9f88e4 100644 --- a/server/opencode-process.test.ts +++ b/server/opencode-process.test.ts @@ -305,6 +305,48 @@ describe('OpenCodeProcess', () => { expect(thinkingHandler).not.toHaveBeenCalled() }) + it('routes Kimi-style field=text deltas by partID (reasoning hidden, answer shown)', async () => { + // Kimi via OpenCode streams BOTH reasoning and the answer as field=text + // deltas, distinguished only by partID, and never sends + // message.part.updated — so the part kind is resolved via a REST lookup. + const textHandler = vi.fn() + const thinkingHandler = vi.fn() + ocp.on('text', textHandler) + ocp.on('thinking', thinkingHandler) + setSessionId(ocp, 'oc-session-1') + + // Classify 'prt_reason' as reasoning, everything else as text. + mockFetch.mockImplementation((url: string) => { + const type = url.includes('prt_reason') ? 'reasoning' : 'text' + return Promise.resolve({ ok: true, json: () => Promise.resolve({ type }) }) + }) + + for (const d of ['The user ', 'is greeting me. ', 'I should respond.']) { + callHandleSSE(ocp, { + type: 'message.part.delta', + properties: { sessionID: 'oc-session-1', messageID: 'msg_1', partID: 'prt_reason', field: 'text', delta: d }, + }) + } + // Deltas are buffered pending classification — nothing emitted yet. + expect(textHandler).not.toHaveBeenCalled() + expect(thinkingHandler).not.toHaveBeenCalled() + + // Once the lookup resolves, reasoning becomes a thinking summary, never text. + await vi.waitFor(() => expect(thinkingHandler).toHaveBeenCalledTimes(1)) + expect(textHandler).not.toHaveBeenCalled() + + for (const d of ['Hello', '! How can I help?']) { + callHandleSSE(ocp, { + type: 'message.part.delta', + properties: { sessionID: 'oc-session-1', messageID: 'msg_1', partID: 'prt_answer', field: 'text', delta: d }, + }) + } + await vi.waitFor(() => expect(textHandler).toHaveBeenCalled()) + const shown = textHandler.mock.calls.map(c => c[0] as string).join('') + expect(shown).toBe('Hello! How can I help?') + expect(shown).not.toContain('greeting') + }) + it('maps running tool parts to tool_active events', () => { const toolActiveHandler = vi.fn() ocp.on('tool_active', toolActiveHandler) diff --git a/server/opencode-process.ts b/server/opencode-process.ts index 116089d2..b6ce5dda 100644 --- a/server/opencode-process.ts +++ b/server/opencode-process.ts @@ -32,6 +32,8 @@ import { summarizeToolInput } from './tool-labels.js' /** A part within an OpenCode message (text, reasoning, tool, step markers). */ interface OpenCodeMessagePart { + /** Stable part ID — matches the `partID` carried on message.part.delta events. */ + id?: string type: 'text' | 'reasoning' | 'tool' | 'step-start' | 'step-finish' /** Text/reasoning content (field name is 'text', not 'content'). */ text?: string @@ -443,6 +445,19 @@ export class OpenCodeProcess extends EventEmitter implement * buffer instead of being emitted as visible text. */ private inReasoningPhase = false + /** + * Cache of message-part kinds keyed by partID. OpenCode's `message.part.delta` + * stream uses `field=text` for BOTH reasoning and text parts — the only + * discriminator is the `partID`. Some providers (Kimi/ollama-cloud) never emit + * `message.part.updated`/`message.updated`, so the part type must be resolved + * via a REST lookup (classifyPart) and remembered here to route subsequent + * deltas without re-fetching. + */ + private partKinds = new Map() + /** Buffered `field=text` deltas per partID, held until the part kind is known. */ + private partDeltaBuffers = new Map() + /** partIDs with an in-flight classifyPart REST lookup (dedup guard). */ + private partClassifyInflight = new Set() constructor(workingDir: string, opts?: Partial) { super() @@ -711,6 +726,97 @@ export class OpenCodeProcess extends EventEmitter implement return sessionID === this.opencodeSessionId } + /** + * Emit a visible-text delta, buffering the leading deltas first so a + * user-echo prefix (some providers echo the user message back) can be + * detected and stripped before display. + */ + private emitTextDelta(delta: string): void { + if (!this.deltaBufferFlushed && this.lastUserInput) { + this.deltaBuffer += delta + if (this.deltaBuffer.length >= this.lastUserInput.length) { + this.deltaBufferFlushed = true + if (this.deltaBuffer.startsWith(this.lastUserInput)) { + const remainder = this.deltaBuffer.slice(this.lastUserInput.length) + if (remainder) this.emit('text', remainder) + } else { + this.emit('text', this.deltaBuffer) + } + this.deltaBuffer = '' + } + // Still buffering — don't emit yet + } else { + this.emit('text', delta) + } + } + + /** + * Accumulate a reasoning delta and emit a one-line thinking summary once + * enough content has arrived. Reasoning is never shown in the transcript. + */ + private appendReasoningDelta(delta: string): void { + this.reasoningBuffer += delta + if (this.reasoningBuffer.length > 20 && !this.emittedReasoningSummary) { + this.emittedReasoningSummary = true + const match = this.reasoningBuffer.match(/^(.+?[.!?\n])/) + const summary = match && match[1].length <= 120 + ? match[1].replace(/\n/g, ' ').trim() + : this.reasoningBuffer.slice(0, 80).trim() + this.emit('thinking', summary) + } + } + + /** + * Record the resolved kind for a partID and flush any deltas buffered while + * its type was unknown — reasoning deltas become a thinking summary, text + * deltas are emitted, anything else is dropped. + */ + private recordPartKind(partID: string, kind: 'reasoning' | 'text' | 'other'): void { + if (this.partKinds.get(partID) === kind && !this.partDeltaBuffers.has(partID)) return + this.partKinds.set(partID, kind) + const buf = this.partDeltaBuffers.get(partID) + if (!buf) return + this.partDeltaBuffers.delete(partID) + if (kind === 'reasoning') { + for (const d of buf) this.appendReasoningDelta(d) + } else if (kind === 'text') { + for (const d of buf) this.emitTextDelta(d) + } + // 'other' (tool/step) → discard buffered text + } + + /** + * Resolve a part's kind via REST when the SSE stream gives no type signal. + * OpenCode's delta stream uses `field=text` for both reasoning and text + * parts, and some providers (e.g. Kimi/ollama-cloud) never emit + * message.part.updated, so the part must be fetched to know whether its + * deltas are visible response text or hidden reasoning. Defaults to showing + * the content on any failure — better to display than to silently hide. + */ + private async classifyPart(messageID: string, partID: string): Promise { + if (this.partKinds.has(partID) || this.partClassifyInflight.has(partID)) return + this.partClassifyInflight.add(partID) + let kind: 'reasoning' | 'text' | 'other' = 'text' + try { + const baseUrl = `http://localhost:${serverState.port}` + const res = await fetch( + `${baseUrl}/session/${this.opencodeSessionId}/message/${messageID}/part/${partID}`, + { + headers: { ...authHeaders(), 'x-opencode-directory': this.workingDir }, + signal: AbortSignal.timeout(10_000), + }, + ) + if (res.ok) { + const part = await res.json() as { type?: string } + kind = part.type === 'reasoning' ? 'reasoning' : part.type === 'text' ? 'text' : 'other' + } + } catch { + // Network/timeout — fall back to showing the content as text. + } + this.partClassifyInflight.delete(partID) + this.recordPartKind(partID, kind) + } + /** Flush any buffered text deltas that haven't been emitted yet (e.g. turn ended before buffer threshold). */ private flushDeltaBuffer(): void { if (!this.deltaBufferFlushed && this.deltaBuffer) { @@ -767,55 +873,43 @@ export class OpenCodeProcess extends EventEmitter implement } if (field === 'text' && delta) { this.receivedDeltas = true + const partID = properties.partID as string | undefined - // Some providers (e.g. Kimi via OpenCode) send reasoning content as - // field=text deltas. When inReasoningPhase is set (by a preceding - // part.updated type=reasoning event), route to reasoning buffer. + // OpenCode streams BOTH reasoning and visible-text parts as + // `field=text` deltas; the only discriminator is the partID. Route + // based on the known part kind. Legacy `inReasoningPhase` (set by + // part.updated type=reasoning on versions that emit it) still wins. if (this.inReasoningPhase) { - this.reasoningBuffer += delta - if (this.reasoningBuffer.length > 20 && !this.emittedReasoningSummary) { - this.emittedReasoningSummary = true - const match = this.reasoningBuffer.match(/^(.+?[.!?\n])/) - const summary = match && match[1].length <= 120 - ? match[1].replace(/\n/g, ' ').trim() - : this.reasoningBuffer.slice(0, 80).trim() - this.emit('thinking', summary) - } + this.appendReasoningDelta(delta) break } - - // Buffer initial deltas to detect and strip user echo prefix. - // Some providers echo the user message at the start of the assistant - // response, which causes duplicate display. - if (!this.deltaBufferFlushed && this.lastUserInput) { - this.deltaBuffer += delta - if (this.deltaBuffer.length >= this.lastUserInput.length) { - this.deltaBufferFlushed = true - if (this.deltaBuffer.startsWith(this.lastUserInput)) { - const remainder = this.deltaBuffer.slice(this.lastUserInput.length) - if (remainder) this.emit('text', remainder) - } else { - this.emit('text', this.deltaBuffer) - } - this.deltaBuffer = '' - } - // Still buffering — don't emit yet - } else { - this.emit('text', delta) + const kind = partID ? this.partKinds.get(partID) : undefined + if (kind === 'reasoning') { + this.appendReasoningDelta(delta) + break } - } else if (field === 'reasoning' && delta) { - // Accumulate reasoning deltas and emit a thinking summary once we - // have enough content, so the UI shows a thinking indicator during - // streaming (not only when message.part.updated arrives later). - this.reasoningBuffer += delta - if (this.reasoningBuffer.length > 20 && !this.emittedReasoningSummary) { - this.emittedReasoningSummary = true - const match = this.reasoningBuffer.match(/^(.+?[.!?\n])/) - const summary = match && match[1].length <= 120 - ? match[1].replace(/\n/g, ' ').trim() - : this.reasoningBuffer.slice(0, 80).trim() - this.emit('thinking', summary) + if (kind === 'other') { + // tool/step parts shouldn't stream visible text — ignore. + break + } + if (kind === 'text' || !partID) { + this.emitTextDelta(delta) + break } + + // Unknown partID: some providers (e.g. Kimi/ollama-cloud) never emit + // message.part.updated, so the part type isn't known from the stream. + // Buffer this part's deltas and resolve its kind via a REST lookup; + // the buffer is flushed (or dropped, if reasoning) once classified. + const buf = this.partDeltaBuffers.get(partID) + if (buf) buf.push(delta) + else this.partDeltaBuffers.set(partID, [delta]) + const messageID = properties.messageID as string | undefined + if (messageID) void this.classifyPart(messageID, partID) + } else if (field === 'reasoning' && delta) { + // Some providers send reasoning on a dedicated `field=reasoning` + // stream — always hidden from the transcript. + this.appendReasoningDelta(delta) } break } @@ -845,6 +939,9 @@ export class OpenCodeProcess extends EventEmitter implement if (this.inReasoningPhase) { this.inReasoningPhase = false } + // Record the kind so streaming deltas for this partID route + // correctly without a REST lookup (and flush any buffered ones). + if (part.id) this.recordPartKind(part.id, 'text') // Text may arrive via message.part.delta (streaming) or as full // content here (OpenCode >=1.4 message.updated). Only emit if we // haven't already streamed it via delta events or emitted it from @@ -866,6 +963,9 @@ export class OpenCodeProcess extends EventEmitter implement // are reasoning content, not visible text. Set the phase flag so // the delta handler routes them to the reasoning buffer. this.inReasoningPhase = true + // Record the kind so streaming deltas for this partID are routed + // to the reasoning buffer (and flush any buffered ones). + if (part.id) this.recordPartKind(part.id, 'reasoning') // OpenCode uses 'text' field, not 'content'. Reasoning may be // empty or encrypted (e.g. OpenAI models). Only emit if present. const content = part.text || '' @@ -1369,6 +1469,9 @@ export class OpenCodeProcess extends EventEmitter implement this.reasoningBuffer = '' this.emittedReasoningSummary = false this.inReasoningPhase = false + this.partKinds.clear() + this.partDeltaBuffers.clear() + this.partClassifyInflight.clear() this.startTurnWatchdog() const baseUrl = `http://localhost:${serverState.port}`