fix(cursor): normalize and smooth agent streams

internal/runtime/executor/cursor_executor.go

@@ -15,6 +15,7 @@ import (
 	"strings"
 	"sync"
 	"time"
+	"unicode"
 
 	"github.com/google/uuid"
 	cursorauth "github.com/router-for-me/CLIProxyAPI/v7/internal/auth/cursor"
@@ -38,6 +39,8 @@ const (
 	cursorHeartbeatInterval = 5 * time.Second
 	cursorSessionTTL        = 5 * time.Minute
 	cursorCheckpointTTL     = 30 * time.Minute
+	cursorStreamFlushDelay  = 16 * time.Millisecond
+	cursorStreamMaxBatch    = 512
 )
 
 // CursorExecutor handles requests to the Cursor API via Connect+Protobuf protocol.
@@ -317,24 +320,37 @@ func (e *CursorExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, r
 	defer sessionCancel()
 	go cursorH2Heartbeat(sessionCtx, stream)
 
-	// Collect full text from streaming response
+	// Collect full text from streaming response, keeping thinking separate so
+	// it can be reported as `reasoning_content` instead of polluting `content`.
 	var fullText strings.Builder
+	var thinkingText strings.Builder
+	usage := &cursorTokenUsage{}
+	usage.setInputEstimate(len(payload))
 	if streamErr := processH2SessionFrames(sessionCtx, stream, params.BlobStore, nil,
 		func(text string, isThinking bool) {
-			fullText.WriteString(text)
+			if isThinking {
+				thinkingText.WriteString(text)
+			} else {
+				fullText.WriteString(text)
+			}
 		},
 		nil,
 		nil,
-		nil, // tokenUsage - non-streaming
+		usage,
 		nil, // onCheckpoint - non-streaming doesn't persist
 	); streamErr != nil && fullText.Len() == 0 {
 		return resp, classifyCursorError(fmt.Errorf("cursor: stream error: %w", streamErr))
 	}
 
 	id := "chatcmpl-" + uuid.New().String()[:28]
 	created := time.Now().Unix()
-	openaiResp := fmt.Sprintf(`{"id":"%s","object":"chat.completion","created":%d,"model":"%s","choices":[{"index":0,"message":{"role":"assistant","content":%s},"finish_reason":"stop"}],"usage":{"prompt_tokens":0,"completion_tokens":0,"total_tokens":0}}`,
-		id, created, parsed.Model, jsonString(fullText.String()))
+	inputTok, outputTok := usage.get()
+	reasoningField := ""
+	if thinkingText.Len() > 0 {
+		reasoningField = fmt.Sprintf(`,"reasoning_content":%s`, jsonString(thinkingText.String()))
+	}
+	openaiResp := fmt.Sprintf(`{"id":"%s","object":"chat.completion","created":%d,"model":"%s","choices":[{"index":0,"message":{"role":"assistant","content":%s%s},"finish_reason":"stop"}],"usage":{"prompt_tokens":%d,"completion_tokens":%d,"total_tokens":%d}}`,
+		id, created, parsed.Model, jsonString(fullText.String()), reasoningField, inputTok, outputTok, inputTok+outputTok)
 
 	// Translate response back to source format if needed
 	result := []byte(openaiResp)
@@ -560,9 +576,10 @@ func (e *CursorExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.A
 			for _, d := range done {
 				emitToOut(cliproxyexecutor.StreamChunk{Payload: bytes.Clone(d)})
 			}
-		} else {
-			emitToOut(cliproxyexecutor.StreamChunk{Payload: []byte("[DONE]")})
 		}
+		// No explicit [DONE] in the non-translated (OpenAI) case: the HTTP
+		// handler already writes `data: [DONE]` when the chunk channel closes,
+		// so emitting one here produced a duplicated [DONE] marker downstream.
 	}
 
 	// Pre-response error detection for transparent failover:
@@ -588,29 +605,39 @@ func (e *CursorExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.A
 		usage := &cursorTokenUsage{}
 		usage.setInputEstimate(len(payload))
 
-		streamErr := processH2SessionFrames(sessionCtx, stream, params.BlobStore, params.McpTools,
-			func(text string, isThinking bool) {
-				if isThinking {
-					if !thinkingActive {
-						thinkingActive = true
-						sendChunkSwitchable(`{"role":"assistant","content":"<think>"}`, "")
-					}
-					sendChunkSwitchable(fmt.Sprintf(`{"content":%s}`, jsonString(text)), "")
-				} else {
-					if thinkingActive {
-						thinkingActive = false
-						sendChunkSwitchable(`{"content":"</think>"}`, "")
-					}
-					sendChunkSwitchable(fmt.Sprintf(`{"content":%s}`, jsonString(text)), "")
+		emitTextDelta := func(text string, isThinking bool) {
+			// Emit thinking as the standard OpenAI `reasoning_content` delta
+			// field instead of inline <think>...</think> tags. Inline tags
+			// pollute `content` for OpenAI clients and end up rendered as
+			// literal text in Anthropic/Claude clients; `reasoning_content`
+			// is understood by the translators (mapped to thinking blocks
+			// for Claude) and by downstream proxies.
+			if isThinking {
+				if !thinkingActive {
+					thinkingActive = true
+					sendChunkSwitchable(`{"role":"assistant","content":""}`, "")
 				}
-			},
+				sendChunkSwitchable(fmt.Sprintf(`{"reasoning_content":%s}`, jsonString(text)), "")
+			} else {
+				thinkingActive = false
+				sendChunkSwitchable(fmt.Sprintf(`{"content":%s}`, jsonString(text)), "")
+			}
+		}
+		streamCoalescer := newCursorStreamCoalescer(
+			sessionCtx,
+			cursorStreamFlushDelay,
+			emitTextDelta,
+		)
+
+		streamErr := processH2SessionFrames(sessionCtx, stream, params.BlobStore, params.McpTools,
+			streamCoalescer.push,
 			func(exec pendingMcpExec) {
-				if thinkingActive {
-					thinkingActive = false
-					sendChunkSwitchable(`{"content":"</think>"}`, "")
-				}
-				toolCallJSON := fmt.Sprintf(`{"tool_calls":[{"index":%d,"id":"%s","type":"function","function":{"name":"%s","arguments":%s}}]}`,
-					toolCallIndex, exec.ToolCallId, exec.ToolName, jsonString(exec.Args))
+				// Preserve ordering: all assistant text must reach the client
+				// before the tool-call boundary is emitted.
+				streamCoalescer.flush()
+				thinkingActive = false
+				toolCallJSON := fmt.Sprintf(`{"tool_calls":[{"index":%d,"id":%s,"type":"function","function":{"name":%s,"arguments":%s}}]}`,
+					toolCallIndex, jsonString(exec.ToolCallId), jsonString(exec.ToolName), jsonString(exec.Args))
 				toolCallIndex++
 				sendChunkSwitchable(toolCallJSON, "")
 				sendChunkSwitchable(`{}`, `"tool_calls"`)
@@ -654,7 +681,7 @@ func (e *CursorExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.A
 				resumeOutCh = resumeOut
 
 				// processH2SessionFrames will now block on toolResultCh (inline wait loop)
-				// while continuing to handle KV messages
+				// while continuing to handle KV messages.
 			},
 			toolResultCh,
 			usage,
@@ -671,6 +698,7 @@ func (e *CursorExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.A
 				log.Debugf("cursor: saved checkpoint (%d bytes) for conv=%s auth=%s", len(cpData), checkpointKey, authID)
 			},
 		)
+		streamCoalescer.close()
 
 		// processH2SessionFrames returned — stream is done.
 		// Check if error happened before any chunks were emitted.
@@ -696,9 +724,6 @@ func (e *CursorExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.A
 			}
 		}
 
-		if thinkingActive {
-			sendChunkSwitchable(`{"content":"</think>"}`, "")
-		}
 		// Include token usage in the final stop chunk
 		inputTok, outputTok := usage.get()
 		stopDelta := fmt.Sprintf(`{},"usage":{"prompt_tokens":%d,"completion_tokens":%d,"total_tokens":%d}`,
@@ -817,6 +842,152 @@ func cursorH2Heartbeat(ctx context.Context, stream *cursorproto.H2Stream) {
 
 // --- Response processing ---
 
+type cursorStreamDeltaCommand struct {
+	text       string
+	isThinking bool
+	flush      bool
+	close      bool
+	ack        chan struct{}
+}
+
+// cursorStreamCoalescer turns Cursor's bursty protobuf deltas into SSE-sized
+// updates at roughly one display frame. The first delta remains immediate,
+// while later adjacent deltas are grouped for at most cursorStreamFlushDelay.
+type cursorStreamCoalescer struct {
+	commands chan cursorStreamDeltaCommand
+	done     chan struct{}
+}
+
+func newCursorStreamCoalescer(
+	ctx context.Context,
+	flushDelay time.Duration,
+	emit func(text string, isThinking bool),
+) *cursorStreamCoalescer {
+	coalescer := &cursorStreamCoalescer{
+		commands: make(chan cursorStreamDeltaCommand),
+		done:     make(chan struct{}),
+	}
+	go coalescer.run(ctx, flushDelay, emit)
+	return coalescer
+}
+
+func (c *cursorStreamCoalescer) push(text string, isThinking bool) {
+	if text == "" {
+		return
+	}
+	select {
+	case c.commands <- cursorStreamDeltaCommand{text: text, isThinking: isThinking}:
+	case <-c.done:
+	}
+}
+
+func (c *cursorStreamCoalescer) flush() {
+	c.sync(cursorStreamDeltaCommand{flush: true, ack: make(chan struct{})})
+}
+
+func (c *cursorStreamCoalescer) close() {
+	c.sync(cursorStreamDeltaCommand{close: true, ack: make(chan struct{})})
+}
+
+func (c *cursorStreamCoalescer) sync(command cursorStreamDeltaCommand) {
+	select {
+	case c.commands <- command:
+	case <-c.done:
+		return
+	}
+	select {
+	case <-command.ack:
+	case <-c.done:
+	}
+}
+
+func (c *cursorStreamCoalescer) run(
+	ctx context.Context,
+	flushDelay time.Duration,
+	emit func(text string, isThinking bool),
+) {
+	defer close(c.done)
+
+	timer := time.NewTimer(flushDelay)
+	if !timer.Stop() {
+		<-timer.C
+	}
+	defer timer.Stop()
+
+	var pending strings.Builder
+	var pendingThinking bool
+	var timerC <-chan time.Time
+	emittedFirst := false
+
+	stopTimer := func() {
+		if timerC == nil {
+			return
+		}
+		if !timer.Stop() {
+			select {
+			case <-timer.C:
+			default:
+			}
+		}
+		timerC = nil
+	}
+	flushPending := func() {
+		stopTimer()
+		if pending.Len() == 0 {
+			return
+		}
+		text := pending.String()
+		pending.Reset()
+		emit(text, pendingThinking)
+	}
+	queue := func(text string, isThinking bool) {
+		if !emittedFirst {
+			emittedFirst = true
+			emit(text, isThinking)
+			return
+		}
+		if pending.Len() > 0 && pendingThinking != isThinking {
+			flushPending()
+		}
+		if pending.Len() == 0 {
+			pendingThinking = isThinking
+			timer.Reset(flushDelay)
+			timerC = timer.C
+		}
+		pending.WriteString(text)
+		if pending.Len() >= cursorStreamMaxBatch {
+			flushPending()
+		}
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			flushPending()
+			return
+		case <-timerC:
+			timerC = nil
+			if pending.Len() > 0 {
+				text := pending.String()
+				pending.Reset()
+				emit(text, pendingThinking)
+			}
+		case command := <-c.commands:
+			switch {
+			case command.close:
+				flushPending()
+				close(command.ack)
+				return
+			case command.flush:
+				flushPending()
+				close(command.ack)
+			default:
+				queue(command.text, command.isThinking)
+			}
+		}
+	}
+}
+
 // cursorTokenUsage tracks token counts from Cursor's TokenDeltaUpdate messages.
 type cursorTokenUsage struct {
 	mu             sync.Mutex
@@ -958,7 +1129,7 @@ func processH2SessionFrames(
 				case cursorproto.ServerMsgExecMcpArgs:
 					if onMcpExec != nil {
 						decodedArgs := decodeMcpArgsToJSON(msg.McpArgs)
-						toolCallId := msg.McpToolCallId
+						toolCallId := normalizeToolCallID(msg.McpToolCallId)
 						if toolCallId == "" {
 							toolCallId = uuid.New().String()
 						}
@@ -1127,7 +1298,7 @@ func parseOpenAIRequest(payload []byte) *parsedOpenAIRequest {
 			continue
 		case "tool":
 			p.ToolResults = append(p.ToolResults, toolResultInfo{
-				ToolCallId: msg.Get("tool_call_id").String(),
+				ToolCallId: normalizeToolCallID(msg.Get("tool_call_id").String()),
 				Content:    extractTextContent(msg.Get("content")),
 			})
 		case "user":
@@ -1461,6 +1632,15 @@ func jsonString(s string) string {
 	return string(b)
 }
 
+func normalizeToolCallID(id string) string {
+	return strings.Map(func(r rune) rune {
+		if unicode.IsControl(r) || unicode.IsSpace(r) {
+			return -1
+		}
+		return r
+	}, id)
+}
+
 func decodeMcpArgsToJSON(args map[string][]byte) string {
 	if len(args) == 0 {
 		return "{}"