@@ -237,12 +237,12 @@ export async function processStream(
237237 const thinkTagStream = new ThinkTagStream ( {
238238 implicitOpen : historyLeaksThinkTags ( agentState . messageHistory ) ,
239239 } )
240- const emitThinkSegments = ( segments : ThinkStreamSegment [ ] ) : void => {
240+ const emitThinkSegments = async ( segments : ThinkStreamSegment [ ] ) => {
241241 for ( const segment of segments ) {
242242 if ( segment . type === 'text' ) {
243- onResponseChunk ( segment . text )
243+ await onResponseChunk ( segment . text )
244244 } else {
245- onResponseChunk ( {
245+ await onResponseChunk ( {
246246 type : 'reasoning_delta' ,
247247 text : segment . text ,
248248 ancestorRunIds,
@@ -595,8 +595,8 @@ export async function processStream(
595595 // A lane that populates the native reasoning field is by definition
596596 // not putting the thought in `content`, so this step is not leaking
597597 // and anything held on speculation is the answer.
598- emitThinkSegments ( thinkTagStream . disarmImplicitOpen ( ) )
599- onResponseChunk ( {
598+ await emitThinkSegments ( thinkTagStream . disarmImplicitOpen ( ) )
599+ await onResponseChunk ( {
600600 type : 'reasoning_delta' ,
601601 text : chunk . text ,
602602 ancestorRunIds,
@@ -609,10 +609,10 @@ export async function processStream(
609609 // isThinkOnlyResponse reads it to keep a thinking-only step from ending
610610 // the turn, and historyLeaksThinkTags reads it to arm the next step.
611611 // Stripping it here would erase both signals.
612- emitThinkSegments ( thinkTagStream . push ( chunk . text ) )
612+ await emitThinkSegments ( thinkTagStream . push ( chunk . text ) )
613613 fullResponseChunks . push ( chunk . text )
614614 } else if ( chunk . type === 'error' ) {
615- onResponseChunk ( chunk )
615+ await onResponseChunk ( chunk )
616616 if ( chunk . source ) {
617617 const recovery = RECOVERY_BY_SOURCE [ chunk . source ]
618618 sawStreamRecovery = true
@@ -708,7 +708,7 @@ export async function processStream(
708708 // recovery, so it is checked here. A step that called a tool is not
709709 // replying yet, so its open block stays a thought.
710710 const finishReason = streamFinishReason ?.( )
711- emitThinkSegments (
711+ await emitThinkSegments (
712712 thinkTagStream . flush ( {
713713 stepCompleted :
714714 ! signal . aborted &&
@@ -761,7 +761,7 @@ export async function processStream(
761761 // Release anything the think-tag split is still holding — a partial tag
762762 // that never completed, or a head held for an orphan `</think>` that never
763763 // came. Runs on the abort paths too, so speculation can never lose text.
764- emitThinkSegments ( thinkTagStream . flush ( ) )
764+ await emitThinkSegments ( thinkTagStream . flush ( ) )
765765
766766 // Trigger cleanup of the processStreamWithTools generator so it flushes any
767767 // remaining buffered text to assistantMessages before we build the history.
0 commit comments