fix(ai): parse MiniMax streaming <think> tags as thinking events

MiniMax models embed thinking content as <think> tags in the content
field rather than using a separate reasoning field. Add a streaming
think-tag parser that detects these tags in real-time during streaming
and routes them to the thinking UI instead of displaying raw tags.

Co-authored-by: Cursor <cursoragent@cursor.com>

packages/core/src/ai/__tests__/streaming-think-parser.test.ts

@@ -0,0 +1,131 @@
+import assert from 'node:assert/strict'
+import { describe, it } from 'node:test'
+import { StreamingThinkTagParser, type StreamParserEvent } from '../streaming-think-parser'
+
+function collect(chunks: string[]): StreamParserEvent[] {
+  const events: StreamParserEvent[] = []
+  const parser = new StreamingThinkTagParser((ev) => events.push(ev))
+  for (const chunk of chunks) {
+    parser.feed(chunk)
+  }
+  parser.flush()
+  return events
+}
+
+describe('StreamingThinkTagParser', () => {
+  it('passes through plain text as content', () => {
+    const events = collect(['Hello', ' world'])
+    assert.deepEqual(events, [
+      { type: 'content', content: 'Hello' },
+      { type: 'content', content: ' world' },
+    ])
+  })
+
+  it('detects <think> tag in a single chunk', () => {
+    const events = collect(['<think>reasoning</think>answer'])
+    assert.deepEqual(events, [
+      { type: 'thinking_start', tag: 'think' },
+      { type: 'thinking_delta', content: 'reasoning' },
+      { type: 'thinking_end' },
+      { type: 'content', content: 'answer' },
+    ])
+  })
+
+  it('handles tag split across chunks', () => {
+    const events = collect(['<thi', 'nk>reasoning</think>answer'])
+    assert.deepEqual(events, [
+      { type: 'thinking_start', tag: 'think' },
+      { type: 'thinking_delta', content: 'reasoning' },
+      { type: 'thinking_end' },
+      { type: 'content', content: 'answer' },
+    ])
+  })
+
+  it('handles closing tag split across chunks', () => {
+    const events = collect(['<think>reasoning</thi', 'nk>answer'])
+    assert.deepEqual(events, [
+      { type: 'thinking_start', tag: 'think' },
+      { type: 'thinking_delta', content: 'reasoning' },
+      { type: 'thinking_end' },
+      { type: 'content', content: 'answer' },
+    ])
+  })
+
+  it('handles content before and after think block', () => {
+    const events = collect(['before<think>middle</think>after'])
+    assert.deepEqual(events, [
+      { type: 'content', content: 'before' },
+      { type: 'thinking_start', tag: 'think' },
+      { type: 'thinking_delta', content: 'middle' },
+      { type: 'thinking_end' },
+      { type: 'content', content: 'after' },
+    ])
+  })
+
+  it('handles < that is not a think tag', () => {
+    const events = collect(['a < b > c'])
+    assert.deepEqual(events, [
+      { type: 'content', content: 'a ' },
+      { type: 'content', content: '<' },
+      { type: 'content', content: ' b > c' },
+    ])
+  })
+
+  it('handles other think tags like <reasoning>', () => {
+    const events = collect(['<reasoning>step 1</reasoning>done'])
+    assert.deepEqual(events, [
+      { type: 'thinking_start', tag: 'reasoning' },
+      { type: 'thinking_delta', content: 'step 1' },
+      { type: 'thinking_end' },
+      { type: 'content', content: 'done' },
+    ])
+  })
+
+  it('flushes buffered thinking content on flush()', () => {
+    const events: StreamParserEvent[] = []
+    const parser = new StreamingThinkTagParser((ev) => events.push(ev))
+    parser.feed('<think>partial thinking')
+    parser.flush()
+    assert.deepEqual(events, [
+      { type: 'thinking_start', tag: 'think' },
+      { type: 'thinking_delta', content: 'partial thinking' },
+    ])
+  })
+
+  it('flushes incomplete close tag as thinking content', () => {
+    const events: StreamParserEvent[] = []
+    const parser = new StreamingThinkTagParser((ev) => events.push(ev))
+    parser.feed('<think>content</thi')
+    parser.flush()
+    assert.deepEqual(events, [
+      { type: 'thinking_start', tag: 'think' },
+      { type: 'thinking_delta', content: 'content' },
+      { type: 'thinking_delta', content: '</thi' },
+    ])
+  })
+
+  it('handles character-by-character streaming', () => {
+    const text = '<think>hi</think>ok'
+    const events = collect(text.split(''))
+    assert.deepEqual(events, [
+      { type: 'thinking_start', tag: 'think' },
+      { type: 'thinking_delta', content: 'h' },
+      { type: 'thinking_delta', content: 'i' },
+      { type: 'thinking_end' },
+      { type: 'content', content: 'o' },
+      { type: 'content', content: 'k' },
+    ])
+  })
+
+  it('handles < inside think block that is not closing tag', () => {
+    const events = collect(['<think>a < b</think>done'])
+    assert.deepEqual(events, [
+      { type: 'thinking_start', tag: 'think' },
+      { type: 'thinking_delta', content: 'a ' },
+      { type: 'thinking_delta', content: '<' },
+      { type: 'thinking_delta', content: ' b' },
+      { type: 'thinking_end' },
+      { type: 'content', content: 'done' },
+    ])
+  })
+})

packages/core/src/ai/content-parser.ts

@@ -3,7 +3,7 @@
  * Extracts thinking tags and strips tool-call markup from LLM output.
  */
 
-const THINK_TAGS = ['think', 'analysis', 'reasoning', 'reflection', 'thought', 'thinking']
+export const THINK_TAGS = ['think', 'analysis', 'reasoning', 'reflection', 'thought', 'thinking']
 
 export function extractThinkingContent(content: string): { thinking: string; cleanContent: string } {
   if (!content) {

packages/core/src/ai/index.ts

@@ -24,4 +24,8 @@ export { BUILTIN_PROVIDERS, getBuiltinProviderById } from './provider-registry'
 export { BUILTIN_MODELS, getBuiltinModelsByProvider, getBuiltinModelById } from './model-catalog'
 
 // Content parsing (thinking-tag extraction, tool-call stripping, avatar sanitization)
-export { extractThinkingContent, stripToolCallTags, stripAvatarFields } from './content-parser'
+export { THINK_TAGS, extractThinkingContent, stripToolCallTags, stripAvatarFields } from './content-parser'
+
+// Streaming think-tag parser (for models that embed <think> in content)
+export { StreamingThinkTagParser, needsStreamingThinkParsing } from './streaming-think-parser'
+export type { StreamParserEvent } from './streaming-think-parser'

packages/core/src/ai/streaming-think-parser.ts

@@ -0,0 +1,142 @@
+/**
+ * Streaming think-tag parser.
+ *
+ * Detects <think>, <reasoning>, etc. tags embedded in streaming text deltas
+ * and splits them into separate thinking/content events. Handles tags split
+ * across multiple chunks via internal buffering.
+ */
+
+import { THINK_TAGS } from './content-parser'
+
+export type StreamParserEvent =
+  | { type: 'content'; content: string }
+  | { type: 'thinking_start'; tag: string }
+  | { type: 'thinking_delta'; content: string }
+  | { type: 'thinking_end' }
+
+type ParserState = 'content' | 'thinking'
+
+export class StreamingThinkTagParser {
+  private state: ParserState = 'content'
+  private buffer = ''
+  private activeTag = ''
+  private emit: (event: StreamParserEvent) => void
+
+  constructor(emit: (event: StreamParserEvent) => void) {
+    this.emit = emit
+  }
+
+  feed(text: string): void {
+    this.buffer += text
+    this.processBuffer()
+  }
+
+  /** Flush any remaining buffer (call at stream end). */
+  flush(): void {
+    if (!this.buffer) return
+    if (this.state === 'thinking') {
+      this.emit({ type: 'thinking_delta', content: this.buffer })
+    } else {
+      this.emit({ type: 'content', content: this.buffer })
+    }
+    this.buffer = ''
+  }
+
+  private processBuffer(): void {
+    while (this.buffer.length > 0) {
+      const ltIdx = this.buffer.indexOf('<')
+
+      if (ltIdx === -1) {
+        this.emitBufferContent(this.buffer)
+        this.buffer = ''
+        return
+      }
+
+      if (ltIdx > 0) {
+        this.emitBufferContent(this.buffer.slice(0, ltIdx))
+        this.buffer = this.buffer.slice(ltIdx)
+      }
+
+      // Buffer starts with '<', try to match a tag
+      const tagResult = this.tryMatchTag()
+      if (tagResult === 'incomplete') {
+        // Need more data; keep buffer for next feed()
+        return
+      }
+      if (tagResult === 'matched') {
+        // Tag was consumed and events emitted, continue processing
+        continue
+      }
+      // 'no_match': the '<' is not a think tag, emit it and advance
+      this.emitBufferContent('<')
+      this.buffer = this.buffer.slice(1)
+    }
+  }
+
+  /**
+   * Try to match a think tag at the start of buffer.
+   * Returns: 'matched' if a tag was consumed, 'incomplete' if more data
+   * needed, 'no_match' if it's not a valid think tag.
+   */
+  private tryMatchTag(): 'matched' | 'incomplete' | 'no_match' {
+    if (this.state === 'content') {
+      return this.tryMatchOpenTag()
+    }
+    return this.tryMatchCloseTag()
+  }
+
+  private tryMatchOpenTag(): 'matched' | 'incomplete' | 'no_match' {
+    // Look for <tagName> pattern
+    for (const tag of THINK_TAGS) {
+      const openTag = `<${tag}>`
+      if (this.buffer.startsWith(openTag)) {
+        this.buffer = this.buffer.slice(openTag.length)
+        this.state = 'thinking'
+        this.activeTag = tag
+        this.emit({ type: 'thinking_start', tag })
+        return 'matched'
+      }
+      // Could be a partial match — check if buffer is a prefix of the tag
+      if (openTag.startsWith(this.buffer) && this.buffer.length < openTag.length) {
+        return 'incomplete'
+      }
+    }
+    return 'no_match'
+  }
+
+  private tryMatchCloseTag(): 'matched' | 'incomplete' | 'no_match' {
+    const closeTag = `</${this.activeTag}>`
+    if (this.buffer.startsWith(closeTag)) {
+      this.buffer = this.buffer.slice(closeTag.length)
+      this.state = 'content'
+      this.emit({ type: 'thinking_end' })
+      this.activeTag = ''
+      return 'matched'
+    }
+    // Could be a partial match
+    if (closeTag.startsWith(this.buffer) && this.buffer.length < closeTag.length) {
+      return 'incomplete'
+    }
+    // Not a close tag — emit the '<' as thinking content and continue
+    this.emit({ type: 'thinking_delta', content: '<' })
+    this.buffer = this.buffer.slice(1)
+    return 'matched'
+  }
+
+  private emitBufferContent(text: string): void {
+    if (!text) return
+    if (this.state === 'thinking') {
+      this.emit({ type: 'thinking_delta', content: text })
+    } else {
+      this.emit({ type: 'content', content: text })
+    }
+  }
+}
+
+/**
+ * Check if a model likely embeds think tags in content
+ * (i.e. uses native <think> output rather than separate reasoning fields).
+ */
+export function needsStreamingThinkParsing(provider: string, _modelId: string): boolean {
+  return provider === 'minimax'
+}

packages/core/src/index.ts

@@ -225,10 +225,14 @@ export {
   BUILTIN_MODELS,
   getBuiltinModelsByProvider,
   getBuiltinModelById,
+  THINK_TAGS,
   extractThinkingContent,
   stripToolCallTags,
   stripAvatarFields,
+  StreamingThinkTagParser,
+  needsStreamingThinkParsing,
 } from './ai'
+export type { StreamParserEvent } from './ai'
 
 // Import utilities
 export { generateMessageKey, buildMemberIdMap } from './import'

packages/node-runtime/src/ai/agent/core.ts

@@ -12,6 +12,7 @@ import {
   type Usage as PiUsage,
   streamSimple as defaultStreamSimple,
 } from '@mariozechner/pi-ai'
+import { StreamingThinkTagParser, needsStreamingThinkParsing } from '@openchatlab/core'
 
 import type { AgentCoreOptions, AgentCoreResult, AgentTokenUsage, SimpleHistoryMessage } from './types'
 
@@ -104,11 +105,42 @@ export async function runAgentCore(options: AgentCoreOptions): Promise<AgentCore
   let hasReachedToolRoundLimit = false
   const thinkingStartTime = new Map<number, number>()
 
+  // For providers that embed <think> tags in content (e.g. MiniMax),
+  // use a streaming parser to split thinking from content.
+  const useThinkParser = needsStreamingThinkParsing(piModel.provider, piModel.id)
+  let thinkParserStartTime: number | undefined
+  const thinkParser = useThinkParser
+    ? new StreamingThinkTagParser((ev) => {
+        switch (ev.type) {
+          case 'content':
+            onEvent({ type: 'content', content: ev.content })
+            break
+          case 'thinking_start':
+            thinkParserStartTime = Date.now()
+            onEvent({ type: 'thinking_start' })
+            break
+          case 'thinking_delta':
+            onEvent({ type: 'thinking_delta', content: ev.content })
+            break
+          case 'thinking_end': {
+            const durationMs = thinkParserStartTime ? Date.now() - thinkParserStartTime : undefined
+            thinkParserStartTime = undefined
+            onEvent({ type: 'thinking_end', durationMs })
+            break
+          }
+        }
+      })
+    : null
+
   const unsubscribe = coreAgent.subscribe((event: PiAgentEvent) => {
     if (event.type === 'message_update') {
       const update = event.assistantMessageEvent
       if (update.type === 'text_delta') {
-        onEvent({ type: 'content', content: update.delta })
+        if (thinkParser) {
+          thinkParser.feed(update.delta)
+        } else {
+          onEvent({ type: 'content', content: update.delta })
+        }
       } else if (update.type === 'thinking_start') {
         thinkingStartTime.set(update.contentIndex, Date.now())
         onEvent({ type: 'thinking_start' })
@@ -146,6 +178,7 @@ export async function runAgentCore(options: AgentCoreOptions): Promise<AgentCore
       onEvent({ type: 'turn_end', round: toolRounds, hadToolCalls })
     } else if (event.type === 'message_end') {
       if (event.message.role === 'assistant') {
+        thinkParser?.flush()
         addPiUsage(event.message.usage)
         onEvent({ type: 'usage_update', usage: { ...totalUsage } })
       }