From 0fead1c035201704509f70fcc2278708615933ef Mon Sep 17 00:00:00 2001 From: nordicnode Date: Mon, 31 Aug 2026 10:49:27 -0700 Subject: [PATCH 1/3] fix(agent-runtime): release an unclosed think-tag open as text at flush MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An explicit open committed the rest of the step to the thinking box unconditionally, so when a model wrote the tag as prose (docs, quoted templates, a broken chat template) the visible answer landed in the thinking box and the user saw an empty or short reply. Give the explicit open the same treatment the implicit head already has: hold until a close settles the block as reasoning, and release the hold as text when the step ends without one — an answer is delayed, never swallowed. The budget stays implicit-head-only: a genuine think block can legitimately run long, and only its close (or flush) settles it. Refs #1155 --- .../util/__tests__/think-tag-stream.test.ts | 46 +++++++++- .../src/util/think-tag-stream.ts | 89 ++++++++++++------- 2 files changed, 101 insertions(+), 34 deletions(-) diff --git a/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts b/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts index 702f5fe3f7..6dcb5f825c 100644 --- a/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts +++ b/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts @@ -59,10 +59,50 @@ describe('ThinkTagStream — paired tags', () => { expect(joined(run(['ends with <']), 'text')).toBe('ends with <') }) - it('treats an unclosed open tag as reasoning through end of stream', () => { + it('releases an unclosed open tag as text at flush', () => { + // The open may be a truncated thought or the tag quoted as prose — both + // look identical until the step ends, so the hold resolves as text either + // way: an answer is delayed, never swallowed (issue #1155, bug 2). const out = run(['truncated thou', 'ght']) - expect(joined(out, 'reasoning')).toBe('truncated thought') - expect(joined(out, 'text')).toBe('') + expect(joined(out, 'reasoning')).toBe('') + expect(joined(out, 'text')).toBe('truncated thought') + }) + + it('keeps an answer quoted around a prose open tag as text', () => { + // Docs quoting the tag: the answer before it already streamed as text, + // the text after it must not vanish into the thinking box. + const out = run([ + 'Write ', + 'like this in your docs. The answer continues here.', + ]) + expect(joined(out, 'reasoning')).toBe('') + expect(joined(out, 'text')).toBe( + 'Write like this in your docs. The answer continues here.', + ) + }) + + it('holds an explicit open only until its close arrives', () => { + const stream = new ThinkTagStream() + expect(stream.push('Answer part one. plan it')).toEqual([ + { type: 'text', text: 'Answer part one. ' }, + ]) + expect(stream.push('Here is the answer.')).toEqual([ + { type: 'reasoning', text: 'plan it' }, + { type: 'text', text: 'Here is the answer.' }, + ]) + }) + + it('releases an explicit open on a native reasoning chunk as text', () => { + const stream = new ThinkTagStream() + expect(stream.push('A quoted open')).toEqual([ + { type: 'text', text: 'A ' }, + ]) + expect(stream.disarmImplicitOpen()).toEqual([ + { type: 'text', text: ' quoted open' }, + ]) + expect(stream.push(' and the answer continues')).toEqual([ + { type: 'text', text: ' and the answer continues' }, + ]) }) }) diff --git a/packages/agent-runtime/src/util/think-tag-stream.ts b/packages/agent-runtime/src/util/think-tag-stream.ts index 86b37af3da..6663af4280 100644 --- a/packages/agent-runtime/src/util/think-tag-stream.ts +++ b/packages/agent-runtime/src/util/think-tag-stream.ts @@ -22,8 +22,15 @@ * * 1. `…` — paired tags. Content between them is reasoning. * Unambiguous, free, always on. - * 2. A bare `` that never closes (a truncated thought). Everything - * after it is reasoning. + * 2. A bare `` that never closes. Two different events produce this: + * a thought truncated mid-stream, or the tag written as PROSE — docs that + * mention the tag, a quoted template, a lane with a broken chat template. + * The two are indistinguishable while the deltas arrive, so everything + * after the open is HELD rather than emitted (same mechanism as rule 3). + * A close settles the hold as reasoning; a step that ends without one + * releases it as text, so an answer is delayed but never swallowed. + * (History still stores the raw text either way — this is a + * display/reclassification concern, not context loss.) * 3. An orphan `` with no open tag — the DeepSeek shape above, where * the open tag was consumed by the chat template's prefill. The text * BEFORE it is reasoning, but by the time the marker arrives that text has @@ -153,27 +160,34 @@ export function historyLeaksThinkTags( export class ThinkTagStream { /** Trailing bytes withheld because they may be the start of a tag. */ private partial = '' - /** Leading content withheld while `implicitOpen` is still undecided. */ + /** Content withheld while the classification of an open block is still + * undecided: the implicit head, or everything since an explicit open. */ private held = '' private implicitOpen: boolean private inThinkBlock: boolean + /** True while `held` is waiting for the close that settles its + * classification. Armed by construction (implicitOpen) and by every open + * tag; cleared by the close ({@link confirmOpenHold}) or by giving up + * ({@link abandonOpenHold}) — budget, a native reasoning chunk, or flush. */ + private holdingForOpen: boolean constructor(options: ThinkTagStreamOptions = {}) { this.implicitOpen = options.implicitOpen ?? false this.inThinkBlock = this.implicitOpen + this.holdingForOpen = this.implicitOpen } /** - * Give up on `implicitOpen` and release anything held as text. + * Give up on the hold and release anything held as text. * - * Called when the step turns out not to be leaking after all. The strongest - * such signal is a native reasoning chunk: a lane that populates - * `reasoning_content` is by definition not putting the thought in `content`, - * so whatever is in `content` is the answer. + * Called when the step turns out not to be thinking in `content` after all. + * The strongest such signal is a native reasoning chunk: a lane that + * populates `reasoning_content` is by definition not putting the thought in + * `content`, so whatever is in `content` is the answer. */ disarmImplicitOpen(): ThinkStreamSegment[] { - if (!this.implicitOpen) return [] - return this.abandonImplicitOpen() + if (!this.holdingForOpen) return [] + return this.abandonOpenHold() } push(chunk: string): ThinkStreamSegment[] { @@ -189,10 +203,11 @@ export class ThinkTagStream { this.addReasoning(segments, buffer.slice(0, closeIdx)) buffer = buffer.slice(closeIdx + CLOSE_TAG.length) this.inThinkBlock = false - // The close the implicit block was waiting for: everything held is - // confirmed reasoning. It can only happen once — a later orphan close - // is an ordinary stray marker and is stripped below. - this.confirmImplicitOpen(segments) + // The close the hold was waiting for: everything held is confirmed + // reasoning — whether the block was opened implicitly (a leaked chain + // of thought) or explicitly. A later orphan close is an ordinary + // stray marker and is stripped below. + this.confirmOpenHold(segments) continue } @@ -203,6 +218,11 @@ export class ThinkTagStream { this.addText(segments, buffer.slice(0, openIdx)) buffer = buffer.slice(openIdx + OPEN_TAG.length) this.inThinkBlock = true + // An open tag could be a block the model is thinking in, or the tag + // quoted as prose. Both look identical until a close (or the end of + // the step) settles it, so hold from here rather than committing the + // rest of the step to the thinking box. + this.holdingForOpen = true continue } // Orphan close with nothing to close: drop the marker so it cannot reach @@ -220,23 +240,23 @@ export class ThinkTagStream { } /** Emit everything withheld. A partial tag that never completed was always - * just text, and content held for an orphan close that never came is the - * answer — releasing both here is what makes the speculation lossless. */ + * just text, and content held for a close that never came is the answer — + * releasing both here is what makes the hold lossless. An unclosed open is + * treated exactly like an orphan close: the marker is scaffolding, the + * text around it is the answer. */ flush(): ThinkStreamSegment[] { const segments: ThinkStreamSegment[] = [] + if (this.holdingForOpen) segments.push(...this.abandonOpenHold()) const trailing = this.partial this.partial = '' - if (trailing) { - if (this.inThinkBlock) this.addReasoning(segments, trailing) - else this.addText(segments, trailing) - } - if (this.implicitOpen) segments.push(...this.abandonImplicitOpen()) + if (trailing) this.addText(segments, trailing) return segments } - /** The orphan close arrived: what was held was reasoning after all. */ - private confirmImplicitOpen(segments: ThinkStreamSegment[]): void { - if (!this.implicitOpen) return + /** The close arrived: what was held was reasoning after all. */ + private confirmOpenHold(segments: ThinkStreamSegment[]): void { + if (!this.holdingForOpen) return + this.holdingForOpen = false this.implicitOpen = false const held = this.held this.held = '' @@ -244,7 +264,8 @@ export class ThinkTagStream { } /** No close is coming: what was held was the answer. */ - private abandonImplicitOpen(): ThinkStreamSegment[] { + private abandonOpenHold(): ThinkStreamSegment[] { + this.holdingForOpen = false this.implicitOpen = false this.inThinkBlock = false const held = this.held @@ -259,16 +280,22 @@ export class ThinkTagStream { // A nested/duplicated open tag inside a block is scaffolding, never thought. const cleaned = text.split(OPEN_TAG).join('') if (!cleaned) return - if (!this.implicitOpen) { + if (!this.holdingForOpen) { push(segments, 'reasoning', cleaned) return } - // Still undecided: this is reasoning only if an orphan close confirms it, - // so hold rather than send. Past the budget the step is answering, not - // thinking, and the hold is released as text. + // Undecided: reasoning only if a close confirms it, so hold rather than + // send. The budget bounds only the implicit-head speculation — a held + // chain of thought from a leaking lane runs well past it, so past the + // budget the step is answering, not thinking, and the hold is released as + // text. An explicit open gets no budget: a genuine think block can + // legitimately run long, and only its close (or flush) settles it. this.held += cleaned - if (this.held.length >= IMPLICIT_OPEN_BUDGET_CHARS) { - segments.push(...this.abandonImplicitOpen()) + if ( + this.implicitOpen && + this.held.length >= IMPLICIT_OPEN_BUDGET_CHARS + ) { + segments.push(...this.abandonOpenHold()) } } From 9284bbedb3debfd5135311318298c8fc02b2705d Mon Sep 17 00:00:00 2001 From: nordicnode Date: Tue, 1 Sep 2026 12:03:51 -0700 Subject: [PATCH 2/3] Bound the explicit think-open hold: commit as reasoning past EXPLICIT_OPEN_HOLD_CHARS so paired traces stream live (review feedback on #1173) --- .../util/__tests__/think-tag-stream.test.ts | 49 ++++++++++-- .../src/util/think-tag-stream.ts | 78 ++++++++++++------- 2 files changed, 93 insertions(+), 34 deletions(-) diff --git a/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts b/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts index 6dcb5f825c..f3c70d3fdf 100644 --- a/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts +++ b/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from 'bun:test' import { historyLeaksThinkTags, + EXPLICIT_OPEN_HOLD_CHARS, IMPLICIT_OPEN_BUDGET_CHARS, stripThinkScaffolding, ThinkTagStream, @@ -91,7 +92,36 @@ describe('ThinkTagStream — paired tags', () => { { type: 'text', text: 'Here is the answer.' }, ]) }) + it('commits the hold as reasoning past the bound and streams the rest live', () => { + // A genuine long trace (R1-style) must not buffer until its close: past + // the hold bound the paired-block reading wins and the block streams. + const stream = new ThinkTagStream() + const long = 'x'.repeat(EXPLICIT_OPEN_HOLD_CHARS) + expect(stream.push(`${long}`)).toEqual([ + { type: 'reasoning', text: long }, + ]) + expect(stream.push(' still going')).toEqual([ + { type: 'reasoning', text: ' still going' }, + ]) + // Committed: a step end releases only the trailing partial, as reasoning. + expect(stream.push(' { + const out = run([`${'y'.repeat(EXPLICIT_OPEN_HOLD_CHARS - 1)}`]) + expect(joined(out, 'text')).toBe('y'.repeat(EXPLICIT_OPEN_HOLD_CHARS - 1)) + expect(joined(out, 'reasoning')).toBe('') + }) + + it('does not commit an implicit head past the explicit bound; text release wins', () => { + const stream = new ThinkTagStream({ implicitOpen: true }) + const long = 'x'.repeat(IMPLICIT_OPEN_BUDGET_CHARS) + expect(stream.push(long)).toEqual([{ type: 'text', text: long }]) + expect(stream.push('tail')).toEqual([ + { type: 'text', text: 'tail' }, + ]) + }) it('releases an explicit open on a native reasoning chunk as text', () => { const stream = new ThinkTagStream() expect(stream.push('A quoted open')).toEqual([ @@ -105,7 +135,6 @@ describe('ThinkTagStream — paired tags', () => { ]) }) }) - describe('ThinkTagStream — orphan close, not armed', () => { // The default for every non-leaking model: the prose is not reclassified // (it already streamed), but the bare marker must never reach a transcript. @@ -126,7 +155,10 @@ describe('ThinkTagStream — orphan close, not armed', () => { describe('ThinkTagStream — orphan close, armed', () => { it('reclassifies the head as reasoning once the marker lands', () => { const out = run( - ['Ключевая зацепка: the bundle knows the type.', 'Do that.Real answer.'], + [ + 'Ключевая зацепка: the bundle knows the type.', + 'Do that.Real answer.', + ], { implicitOpen: true }, ) expect(out).toEqual([ @@ -173,7 +205,9 @@ describe('ThinkTagStream — orphan close, armed', () => { { type: 'text', text: ' and more' }, ]) // Disarmed: a later marker is stripped, not treated as a close. - expect(stream.push('tail')).toEqual([{ type: 'text', text: 'tail' }]) + expect(stream.push('tail')).toEqual([ + { type: 'text', text: 'tail' }, + ]) }) it('disarms on a native reasoning chunk and releases the head as text', () => { @@ -223,7 +257,10 @@ describe('historyLeaksThinkTags', () => { expect( historyLeaksThinkTags([ { role: 'user', content: [{ type: 'text', text: 'why ?' }] }, - { role: 'assistant', content: [{ type: 'reasoning', text: '' }] }, + { + role: 'assistant', + content: [{ type: 'reasoning', text: '' }], + }, ]), ).toBe(false) }) @@ -238,7 +275,9 @@ describe('stripThinkScaffolding', () => { it('leaves surrounding whitespace alone, unlike stripThinkTags', () => { expect(stripThinkScaffolding(' spaced ')).toBe(' spaced ') - expect(stripThinkScaffolding('a\n\nx\n\nb')).toBe('a\n\n\n\nb') + expect(stripThinkScaffolding('a\n\nx\n\nb')).toBe( + 'a\n\n\n\nb', + ) }) }) diff --git a/packages/agent-runtime/src/util/think-tag-stream.ts b/packages/agent-runtime/src/util/think-tag-stream.ts index 6663af4280..9de797b0b4 100644 --- a/packages/agent-runtime/src/util/think-tag-stream.ts +++ b/packages/agent-runtime/src/util/think-tag-stream.ts @@ -27,10 +27,13 @@ * mention the tag, a quoted template, a lane with a broken chat template. * The two are indistinguishable while the deltas arrive, so everything * after the open is HELD rather than emitted (same mechanism as rule 3). - * A close settles the hold as reasoning; a step that ends without one - * releases it as text, so an answer is delayed but never swallowed. - * (History still stores the raw text either way — this is a - * display/reclassification concern, not context loss.) + * A close settles the hold as reasoning. Past + * {@link EXPLICIT_OPEN_HOLD_CHARS} without one, the hold COMMITS as + * reasoning and the rest of the block streams live — a genuine chain of + * thought runs arbitrarily long, and buffering one until its close would + * freeze the thinking box for the whole trace. A step that ends still + * undecided (never closed, never crossed the bound) releases the hold as + * text, so an answer is delayed but never swallowed. * 3. An orphan `` with no open tag — the DeepSeek shape above, where * the open tag was consumed by the chat template's prefill. The text * BEFORE it is reasoning, but by the time the marker arrives that text has @@ -68,13 +71,22 @@ export interface ThinkTagStreamOptions { } /** - * How much leading content to hold while waiting for an orphan ``. + * How much content an open block may hold while its classification is + * undecided, before the hold gives up. * - * A leaked chain of thought runs well past this, so the cap is not there to - * fit one — it bounds the wrong case. If the marker has not arrived by here - * the step is answering, not thinking, and the buffer is released as text. + * For the implicit head ({@link ThinkTagStreamOptions.implicitOpen}) the cap + * bounds the wrong case: a leaked chain of thought runs well past it, so if + * the orphan `` has not arrived by here the step is answering, not + * thinking, and the buffer is released as text. + * + * For an explicit `` open the same magnitude is a latency cap, not a + * size estimate: past it the hold commits as reasoning ({@link + * EXPLICIT_OPEN_HOLD_CHARS}) rather than releasing as text, because a genuine + * block legitimately runs long — the wrong case to bound is the waiting, not + * the trace. */ export const IMPLICIT_OPEN_BUDGET_CHARS = 4000 +export const EXPLICIT_OPEN_HOLD_CHARS = IMPLICIT_OPEN_BUDGET_CHARS /** * Remove think scaffolding from a fragment, leaving everything else — including @@ -167,8 +179,9 @@ export class ThinkTagStream { private inThinkBlock: boolean /** True while `held` is waiting for the close that settles its * classification. Armed by construction (implicitOpen) and by every open - * tag; cleared by the close ({@link confirmOpenHold}) or by giving up - * ({@link abandonOpenHold}) — budget, a native reasoning chunk, or flush. */ + * tag; settled by the close or by the explicit hold bound + * ({@link confirmOpenHold}), or given up ({@link abandonOpenHold}) — + * budget, a native reasoning chunk, or flush. */ private holdingForOpen: boolean constructor(options: ThinkTagStreamOptions = {}) { @@ -240,20 +253,26 @@ export class ThinkTagStream { } /** Emit everything withheld. A partial tag that never completed was always - * just text, and content held for a close that never came is the answer — - * releasing both here is what makes the hold lossless. An unclosed open is - * treated exactly like an orphan close: the marker is scaffolding, the - * text around it is the answer. */ + * just text — except inside a block that committed as reasoning, where a + * truncated close belongs with the trace it would have ended. Content + * held for a close that never came is the answer: releasing it here is + * what makes the hold lossless. An unclosed open is treated exactly like + * an orphan close — the marker is scaffolding, the text around it is the + * answer. */ flush(): ThinkStreamSegment[] { const segments: ThinkStreamSegment[] = [] if (this.holdingForOpen) segments.push(...this.abandonOpenHold()) const trailing = this.partial this.partial = '' - if (trailing) this.addText(segments, trailing) + if (trailing) { + if (this.inThinkBlock) this.addReasoning(segments, trailing) + else this.addText(segments, trailing) + } return segments } - /** The close arrived: what was held was reasoning after all. */ + /** The close arrived — or the hold bound crossed without one: what was + * held was reasoning after all, and the rest of the block streams live. */ private confirmOpenHold(segments: ThinkStreamSegment[]): void { if (!this.holdingForOpen) return this.holdingForOpen = false @@ -273,10 +292,7 @@ export class ThinkTagStream { return held ? [{ type: 'text', text: held }] : [] } - private addReasoning( - segments: ThinkStreamSegment[], - text: string, - ): void { + private addReasoning(segments: ThinkStreamSegment[], text: string): void { // A nested/duplicated open tag inside a block is scaffolding, never thought. const cleaned = text.split(OPEN_TAG).join('') if (!cleaned) return @@ -285,17 +301,21 @@ export class ThinkTagStream { return } // Undecided: reasoning only if a close confirms it, so hold rather than - // send. The budget bounds only the implicit-head speculation — a held - // chain of thought from a leaking lane runs well past it, so past the - // budget the step is answering, not thinking, and the hold is released as - // text. An explicit open gets no budget: a genuine think block can - // legitimately run long, and only its close (or flush) settles it. + // send. Two give-up points settle it, in opposite directions. The + // implicit-head budget releases as TEXT — a held chain of thought from a + // leaking lane runs well past it, so past the budget the step is + // answering, not thinking. The explicit-open hold COMMITS as reasoning — + // a genuine block runs arbitrarily long, and buffering one until its + // close would freeze the thinking box for the whole trace, so past the + // bound the paired-block reading wins and the rest streams live. this.held += cleaned - if ( - this.implicitOpen && - this.held.length >= IMPLICIT_OPEN_BUDGET_CHARS - ) { + if (this.implicitOpen && this.held.length >= IMPLICIT_OPEN_BUDGET_CHARS) { segments.push(...this.abandonOpenHold()) + } else if ( + !this.implicitOpen && + this.held.length >= EXPLICIT_OPEN_HOLD_CHARS + ) { + this.confirmOpenHold(segments) } } From 1877b55b296603eebf4adfe95b6a4fb5609a4ea6 Mon Sep 17 00:00:00 2001 From: nordicnode Date: Tue, 1 Sep 2026 12:29:12 -0700 Subject: [PATCH 3/3] Gate the explicit think-open hold on history instead of bounding it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Clean lanes (every model that pairs its tags) stream rule-1 blocks live with zero buffering — the review's latency concern. Lanes whose last assistant turn left an open unclosed arm holdExplicitOpens, so a prose-quoted tag delays the answer instead of swallowing it, released as text at the shared IMPLICIT_OPEN_BUDGET_CHARS or at flush. Replaces the commit-past-bound design from 9284bbedb, which re-swallowed long prose answers past the bound. --- .../agent-runtime/src/tools/stream-parser.ts | 40 +++-- .../util/__tests__/think-tag-stream.test.ts | 137 +++++++++++++---- .../src/util/think-tag-stream.ts | 145 +++++++++++------- 3 files changed, 216 insertions(+), 106 deletions(-) diff --git a/packages/agent-runtime/src/tools/stream-parser.ts b/packages/agent-runtime/src/tools/stream-parser.ts index edd0e60bde..69bccde3b3 100644 --- a/packages/agent-runtime/src/tools/stream-parser.ts +++ b/packages/agent-runtime/src/tools/stream-parser.ts @@ -2,10 +2,7 @@ import { toolNames } from '@codebuff/common/tools/constants' import { buildArray } from '@codebuff/common/util/array' import { STREAM_RECOVERY_EVENT } from '@codebuff/common/util/axiom-only-log' import { AbortError } from '@codebuff/common/util/error' -import { - assistantMessage, - userMessage, -} from '@codebuff/common/util/messages' +import { assistantMessage, userMessage } from '@codebuff/common/util/messages' import { generateCompactId } from '@codebuff/common/util/string' import { processStreamWithTools } from '../tool-stream-parser' @@ -18,11 +15,11 @@ import { } from './tool-executor' import { withSystemTags } from '../util/messages' import { + historyHasUnclosedOpen, historyLeaksThinkTags, stripThinkScaffolding, ThinkTagStream, } from '../util/think-tag-stream' - import type { CustomToolCall, ExecuteToolCallParams } from './tool-executor' import type { ThinkStreamSegment } from '../util/think-tag-stream' import type { AgentTemplate } from '../templates/types' @@ -156,9 +153,7 @@ export async function processStream( > & ParamsExcluding< typeof processStreamWithTools, - | 'processors' - | 'defaultProcessor' - | 'executeXmlToolCall' + 'processors' | 'defaultProcessor' | 'executeXmlToolCall' >, ) { const { @@ -180,10 +175,11 @@ export async function processStream( // Reasoning that a lane failed to put in its native field arrives here as // ordinary text, tags and all. Reclassify it before it reaches a surface, so // the thinking box is the only place a chain of thought is ever rendered. - // See util/think-tag-stream.ts for the three shapes and why the implicit-open - // rule is armed from the history rather than from a model id. + // See util/think-tag-stream.ts for the three shapes and why the hold rules + // are armed from the history rather than from a model id. const thinkTagStream = new ThinkTagStream({ implicitOpen: historyLeaksThinkTags(agentState.messageHistory), + holdExplicitOpens: historyHasUnclosedOpen(agentState.messageHistory), }) const emitThinkSegments = (segments: ThinkStreamSegment[]): void => { for (const segment of segments) { @@ -205,7 +201,8 @@ export async function processStream( const toolResults: ToolMessage[] = [] const toolResultsToAddToMessageHistory: ToolMessage[] = [] const toolCalls: (CodebuffToolCall | CustomToolCall)[] = [] - const toolCallsToAddToMessageHistory: (CodebuffToolCall | CustomToolCall)[] = [] + const toolCallsToAddToMessageHistory: (CodebuffToolCall | CustomToolCall)[] = + [] const assistantMessages: Message[] = [] // Inline agents replace the parent's history with their result. Track which // current-step messages they inherited so finalization does not append them @@ -255,7 +252,7 @@ export async function processStream( function createToolExecutionCallback(toolName: string, isXmlMode: boolean) { const responseHandler = createResponseHandler() return { - onTagStart: () => { }, + onTagStart: () => {}, onTagEnd: async (_: string, input: Record) => { if (signal.aborted) { return @@ -266,10 +263,10 @@ export async function processStream( // Check if this is an agent tool call that should be transformed to spawn_agents const transformed = !isNativeTool ? tryTransformAgentToolCall({ - toolName, - input, - spawnableAgents: agentTemplate.spawnableAgents, - }) + toolName, + input, + spawnableAgents: agentTemplate.spawnableAgents, + }) : null const isSpawnCall = Boolean(transformed) || @@ -629,17 +626,18 @@ export async function processStream( const completedToolCallIds = new Set( toolResultsToAddToMessageHistory.map((r) => r.toolCallId), ) - const filteredToolCalls = - toolCallsToAddToMessageHistory.filter((tc) => - completedToolCallIds.has(tc.toolCallId), - ) + const filteredToolCalls = toolCallsToAddToMessageHistory.filter((tc) => + completedToolCallIds.has(tc.toolCallId), + ) agentState.messageHistory = buildArray([ ...agentState.messageHistory, ...assistantMessages.filter( (message) => !claimedByInlineAgent.has(message), ), - ...filteredToolCalls.map((toolCall) => assistantMessage({ ...toolCall, type: 'tool-call' })), + ...filteredToolCalls.map((toolCall) => + assistantMessage({ ...toolCall, type: 'tool-call' }), + ), ...toolResultsToAddToMessageHistory, ...errorMessages, ]) diff --git a/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts b/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts index f3c70d3fdf..f62a8eacd9 100644 --- a/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts +++ b/packages/agent-runtime/src/util/__tests__/think-tag-stream.test.ts @@ -1,8 +1,8 @@ import { describe, expect, it } from 'bun:test' import { + historyHasUnclosedOpen, historyLeaksThinkTags, - EXPLICIT_OPEN_HOLD_CHARS, IMPLICIT_OPEN_BUDGET_CHARS, stripThinkScaffolding, ThinkTagStream, @@ -13,7 +13,7 @@ import type { ThinkStreamSegment } from '../think-tag-stream' /** Feed the deltas one at a time, then flush — the shape a real stream has. */ function run( deltas: string[], - options?: { implicitOpen?: boolean }, + options?: { implicitOpen?: boolean; holdExplicitOpens?: boolean }, ): ThinkStreamSegment[] { const stream = new ThinkTagStream(options) const out: ThinkStreamSegment[] = [] @@ -60,30 +60,30 @@ describe('ThinkTagStream — paired tags', () => { expect(joined(run(['ends with <']), 'text')).toBe('ends with <') }) - it('releases an unclosed open tag as text at flush', () => { - // The open may be a truncated thought or the tag quoted as prose — both - // look identical until the step ends, so the hold resolves as text either - // way: an answer is delayed, never swallowed (issue #1155, bug 2). + it('streams an unclosed open as reasoning on a clean lane, live like rule 1', () => { + // Default lane: a bare open is near-certainly a real block (truncated + // thought). Streaming it live keeps the zero-latency main behavior — + // the review's concern was exactly this trace freezing until a close. const out = run(['truncated thou', 'ght']) - expect(joined(out, 'reasoning')).toBe('') - expect(joined(out, 'text')).toBe('truncated thought') + expect(joined(out, 'reasoning')).toBe('truncated thought') + expect(joined(out, 'text')).toBe('') }) - it('keeps an answer quoted around a prose open tag as text', () => { - // Docs quoting the tag: the answer before it already streamed as text, - // the text after it must not vanish into the thinking box. - const out = run([ - 'Write ', - 'like this in your docs. The answer continues here.', - ]) + it('keeps an answer quoted around a prose open tag as text when armed', () => { + // History-proven lane (an open was left unclosed before): the hold is + // armed, so a quoted tag cannot swallow the answer into the thinking box. + const out = run( + ['Write ', 'like this in your docs. The answer continues here.'], + { holdExplicitOpens: true }, + ) expect(joined(out, 'reasoning')).toBe('') expect(joined(out, 'text')).toBe( 'Write like this in your docs. The answer continues here.', ) }) - it('holds an explicit open only until its close arrives', () => { - const stream = new ThinkTagStream() + it('holds an explicit open only until its close arrives, when armed', () => { + const stream = new ThinkTagStream({ holdExplicitOpens: true }) expect(stream.push('Answer part one. plan it')).toEqual([ { type: 'text', text: 'Answer part one. ' }, ]) @@ -92,29 +92,51 @@ describe('ThinkTagStream — paired tags', () => { { type: 'text', text: 'Here is the answer.' }, ]) }) - it('commits the hold as reasoning past the bound and streams the rest live', () => { - // A genuine long trace (R1-style) must not buffer until its close: past - // the hold bound the paired-block reading wins and the block streams. + + it('releases an armed hold past the budget as text, and the rest streams live', () => { + // Same give-up as the implicit head: past the budget the step is + // answering, not thinking — a quoted tag with a long answer after it + // must not be swallowed (issue #1155, bug 2). + const stream = new ThinkTagStream({ holdExplicitOpens: true }) + const long = 'x'.repeat(IMPLICIT_OPEN_BUDGET_CHARS) + expect(stream.push(`Answer ${long}`)).toEqual([ + { type: 'text', text: 'Answer ' }, + { type: 'text', text: long }, + ]) + expect(stream.push(' still answering')).toEqual([ + { type: 'text', text: ' still answering' }, + ]) + // Disarmed: a later marker is stripped, not treated as a close. + expect(stream.push('tail')).toEqual([ + { type: 'text', text: 'tail' }, + ]) + }) + + it('streams a long well-formed trace per-delta on a clean lane, never buffered', () => { + // The review's measurement: a DeepSeek-R1-style trace must not wait for + // its close. One push in, everything so far is already out. const stream = new ThinkTagStream() - const long = 'x'.repeat(EXPLICIT_OPEN_HOLD_CHARS) + const long = 'x'.repeat(5000) expect(stream.push(`${long}`)).toEqual([ { type: 'reasoning', text: long }, ]) expect(stream.push(' still going')).toEqual([ { type: 'reasoning', text: ' still going' }, ]) - // Committed: a step end releases only the trailing partial, as reasoning. - expect(stream.push('answer')).toEqual([ + { type: 'text', text: 'answer' }, + ]) }) - it('keeps the answer as text when an unclosed explicit open stays under the bound', () => { - const out = run([`${'y'.repeat(EXPLICIT_OPEN_HOLD_CHARS - 1)}`]) - expect(joined(out, 'text')).toBe('y'.repeat(EXPLICIT_OPEN_HOLD_CHARS - 1)) - expect(joined(out, 'reasoning')).toBe('') + it('flushes a clean-lane unclosed block tail as reasoning', () => { + const stream = new ThinkTagStream() + expect(stream.push('thought { + it('keeps the implicit head release as text at the shared budget', () => { const stream = new ThinkTagStream({ implicitOpen: true }) const long = 'x'.repeat(IMPLICIT_OPEN_BUDGET_CHARS) expect(stream.push(long)).toEqual([{ type: 'text', text: long }]) @@ -122,8 +144,9 @@ describe('ThinkTagStream — paired tags', () => { { type: 'text', text: 'tail' }, ]) }) - it('releases an explicit open on a native reasoning chunk as text', () => { - const stream = new ThinkTagStream() + + it('releases an explicit open on a native reasoning chunk as text, when armed', () => { + const stream = new ThinkTagStream({ holdExplicitOpens: true }) expect(stream.push('A quoted open')).toEqual([ { type: 'text', text: 'A ' }, ]) @@ -294,3 +317,55 @@ describe('historyLeaksThinkTags — head window', () => { ).toBe(false) }) }) + +describe('historyHasUnclosedOpen', () => { + const assistant = (text: string) => ({ + role: 'assistant', + content: [{ type: 'text', text }], + }) + + it('is false for a clean or properly paired last turn', () => { + expect(historyHasUnclosedOpen([])).toBe(false) + expect(historyHasUnclosedOpen([assistant('xanswer')])).toBe( + false, + ) + }) + + it('is true when the last assistant turn leaves an open unclosed', () => { + expect( + historyHasUnclosedOpen([assistant('answertruncated thought')]), + ).toBe(true) + }) + + it('is false when both tags are quoted in prose', () => { + // Docs quoting the pair close after they open — not evidence of a leak. + expect( + historyHasUnclosedOpen([ + assistant('use and to mark reasoning'), + ]), + ).toBe(false) + }) + + it('heals: only the last assistant turn counts', () => { + // A one-off quoted open arms exactly the next step; a clean reply after + // it disarms, so later genuine traces are never held. + expect( + historyHasUnclosedOpen([ + assistant('answerquoted once'), + assistant('xclean answer'), + ]), + ).toBe(false) + }) + + it('ignores user messages and reasoning parts', () => { + expect( + historyHasUnclosedOpen([ + { role: 'user', content: [{ type: 'text', text: 'why ?' }] }, + { + role: 'assistant', + content: [{ type: 'reasoning', text: 'native' }], + }, + ]), + ).toBe(false) + }) +}) diff --git a/packages/agent-runtime/src/util/think-tag-stream.ts b/packages/agent-runtime/src/util/think-tag-stream.ts index 9de797b0b4..8b2ebb71de 100644 --- a/packages/agent-runtime/src/util/think-tag-stream.ts +++ b/packages/agent-runtime/src/util/think-tag-stream.ts @@ -25,15 +25,17 @@ * 2. A bare `` that never closes. Two different events produce this: * a thought truncated mid-stream, or the tag written as PROSE — docs that * mention the tag, a quoted template, a lane with a broken chat template. - * The two are indistinguishable while the deltas arrive, so everything - * after the open is HELD rather than emitted (same mechanism as rule 3). - * A close settles the hold as reasoning. Past - * {@link EXPLICIT_OPEN_HOLD_CHARS} without one, the hold COMMITS as - * reasoning and the rest of the block streams live — a genuine chain of - * thought runs arbitrarily long, and buffering one until its close would - * freeze the thinking box for the whole trace. A step that ends still - * undecided (never closed, never crossed the bound) releases the hold as - * text, so an answer is delayed but never swallowed. + * The two are indistinguishable while the deltas arrive. On a lane whose + * history shows no unclosed open — the majority, every model that pairs + * its tags — the block streams live as reasoning, exactly like rule 1: + * buffering a genuine chain of thought until its close would freeze the + * thinking box for the whole trace. When {@link historyHasUnclosedOpen} + * proves the lane leaves opens unclosed, + * {@link ThinkTagStreamOptions.holdExplicitOpens} holds from the open + * instead: a close settles it as reasoning, and the budget or the end of + * the step releases it as text, so an answer is delayed but never + * swallowed. The cost of arming from history is one step of lag on a + * lane that leaks for the first time — the same accepted gap rule 3 has. * 3. An orphan `` with no open tag — the DeepSeek shape above, where * the open tag was consumed by the chat template's prefill. The text * BEFORE it is reasoning, but by the time the marker arrives that text has @@ -68,25 +70,32 @@ export interface ThinkTagStreamOptions { * buffer as text. */ implicitOpen?: boolean + + /** + * Hold from every explicit `` open until a close, the + * {@link IMPLICIT_OPEN_BUDGET_CHARS}, or the end of the step settles it. + * + * Off — the default — keeps rule 1 free: a paired block streams live with + * zero buffering, and the only cost is that a prose-quoted open on a lane + * with no history swallows that one step into the thinking box. Arm it from + * {@link historyHasUnclosedOpen}, never unconditionally: on a lane proven + * to leave opens unclosed the hold is protection, but on a clean lane it + * would delay every genuine trace for nothing. + */ + holdExplicitOpens?: boolean } /** - * How much content an open block may hold while its classification is - * undecided, before the hold gives up. - * - * For the implicit head ({@link ThinkTagStreamOptions.implicitOpen}) the cap - * bounds the wrong case: a leaked chain of thought runs well past it, so if - * the orphan `` has not arrived by here the step is answering, not - * thinking, and the buffer is released as text. + * How much leading content to hold while waiting for a settling marker. * - * For an explicit `` open the same magnitude is a latency cap, not a - * size estimate: past it the hold commits as reasoning ({@link - * EXPLICIT_OPEN_HOLD_CHARS}) rather than releasing as text, because a genuine - * block legitimately runs long — the wrong case to bound is the waiting, not - * the trace. + * A leaked chain of thought runs well past this, so the cap is not there to + * fit one — it bounds the wrong case. If the close has not arrived by here + * the step is answering, not thinking, and the buffer is released as text. + * It bounds the implicit head ({@link ThinkTagStreamOptions.implicitOpen}) + * and, when armed, every explicit open + * ({@link ThinkTagStreamOptions.holdExplicitOpens}) alike. */ export const IMPLICIT_OPEN_BUDGET_CHARS = 4000 -export const EXPLICIT_OPEN_HOLD_CHARS = IMPLICIT_OPEN_BUDGET_CHARS /** * Remove think scaffolding from a fragment, leaving everything else — including @@ -161,6 +170,38 @@ export function historyLeaksThinkTags( } return false } +/** + * True when the most recent assistant turn ended inside an explicit think + * block — its text contains an open with no close after it. + * + * The arming signal for {@link ThinkTagStreamOptions.holdExplicitOpens}, + * symmetric with {@link historyLeaksThinkTags}: the same lane decision the + * history is kept for. A lane that pairs its tags cleanly never arms, so + * rule 1 stays free; a lane proven to leave an open unclosed arms the next + * step so a prose-quoted tag cannot swallow the answer. + */ +export function historyHasUnclosedOpen( + messages: readonly { role: string; content: unknown }[], +): boolean { + for (let i = messages.length - 1; i >= 0; i--) { + const message = messages[i] + if (message.role !== 'assistant') continue + if (!Array.isArray(message.content)) return false + let text = '' + for (const part of message.content) { + if ( + part && + typeof part === 'object' && + (part as { type?: unknown }).type === 'text' && + typeof (part as { text?: unknown }).text === 'string' + ) { + text += (part as { text: string }).text + } + } + return text.lastIndexOf(OPEN_TAG) > text.lastIndexOf(CLOSE_TAG) + } + return false +} /** * Incremental classifier over one step's content stream. @@ -172,26 +213,31 @@ export function historyLeaksThinkTags( export class ThinkTagStream { /** Trailing bytes withheld because they may be the start of a tag. */ private partial = '' - /** Content withheld while the classification of an open block is still - * undecided: the implicit head, or everything since an explicit open. */ + /** Content withheld while the classification of the open block is still + * undecided: the implicit head, or everything since an explicit open on an + * armed lane. */ private held = '' private implicitOpen: boolean private inThinkBlock: boolean /** True while `held` is waiting for the close that settles its - * classification. Armed by construction (implicitOpen) and by every open - * tag; settled by the close or by the explicit hold bound - * ({@link confirmOpenHold}), or given up ({@link abandonOpenHold}) — - * budget, a native reasoning chunk, or flush. */ + * classification. Armed by construction (implicitOpen) and by an explicit + * open when {@link ThinkTagStreamOptions.holdExplicitOpens} is set; + * cleared by the close ({@link confirmOpenHold}) or by giving up + * ({@link abandonOpenHold}) — budget, a native reasoning chunk, or flush. */ private holdingForOpen: boolean + private holdExplicitOpens: boolean constructor(options: ThinkTagStreamOptions = {}) { this.implicitOpen = options.implicitOpen ?? false + this.holdExplicitOpens = options.holdExplicitOpens ?? false this.inThinkBlock = this.implicitOpen this.holdingForOpen = this.implicitOpen } /** - * Give up on the hold and release anything held as text. + * Give up on the hold — the implicit head's, or an explicit open's when + * {@link ThinkTagStreamOptions.holdExplicitOpens} armed it — and release + * anything held as text. * * Called when the step turns out not to be thinking in `content` after all. * The strongest such signal is a native reasoning chunk: a lane that @@ -232,10 +278,11 @@ export class ThinkTagStream { buffer = buffer.slice(openIdx + OPEN_TAG.length) this.inThinkBlock = true // An open tag could be a block the model is thinking in, or the tag - // quoted as prose. Both look identical until a close (or the end of - // the step) settles it, so hold from here rather than committing the - // rest of the step to the thinking box. - this.holdingForOpen = true + // quoted as prose. On a clean lane it is near-certainly the former — + // rule 1 streams it live, free. Only a lane the history has proven + // to leave opens unclosed holds from here, so the answer behind a + // quoted tag is delayed rather than swallowed. + if (this.holdExplicitOpens) this.holdingForOpen = true continue } // Orphan close with nothing to close: drop the marker so it cannot reach @@ -253,12 +300,12 @@ export class ThinkTagStream { } /** Emit everything withheld. A partial tag that never completed was always - * just text — except inside a block that committed as reasoning, where a - * truncated close belongs with the trace it would have ended. Content - * held for a close that never came is the answer: releasing it here is - * what makes the hold lossless. An unclosed open is treated exactly like - * an orphan close — the marker is scaffolding, the text around it is the - * answer. */ + * just text — except at the tail of an unclosed block on a clean lane, + * where it belongs with the reasoning it would have ended. Content held + * for a close that never came is the answer: releasing it here is what + * makes the hold lossless. An unclosed open on an armed lane is treated + * exactly like an orphan close — the marker is scaffolding, the text + * around it is the answer. */ flush(): ThinkStreamSegment[] { const segments: ThinkStreamSegment[] = [] if (this.holdingForOpen) segments.push(...this.abandonOpenHold()) @@ -271,8 +318,7 @@ export class ThinkTagStream { return segments } - /** The close arrived — or the hold bound crossed without one: what was - * held was reasoning after all, and the rest of the block streams live. */ + /** The close arrived: what was held was reasoning after all. */ private confirmOpenHold(segments: ThinkStreamSegment[]): void { if (!this.holdingForOpen) return this.holdingForOpen = false @@ -301,21 +347,12 @@ export class ThinkTagStream { return } // Undecided: reasoning only if a close confirms it, so hold rather than - // send. Two give-up points settle it, in opposite directions. The - // implicit-head budget releases as TEXT — a held chain of thought from a - // leaking lane runs well past it, so past the budget the step is - // answering, not thinking. The explicit-open hold COMMITS as reasoning — - // a genuine block runs arbitrarily long, and buffering one until its - // close would freeze the thinking box for the whole trace, so past the - // bound the paired-block reading wins and the rest streams live. + // send. A held chain of thought — implicit head or armed explicit open — + // runs well past the budget, so past it the step is answering, not + // thinking, and the hold is released as text. this.held += cleaned - if (this.implicitOpen && this.held.length >= IMPLICIT_OPEN_BUDGET_CHARS) { + if (this.held.length >= IMPLICIT_OPEN_BUDGET_CHARS) { segments.push(...this.abandonOpenHold()) - } else if ( - !this.implicitOpen && - this.held.length >= EXPLICIT_OPEN_HOLD_CHARS - ) { - this.confirmOpenHold(segments) } }