diff --git a/internal/runtime/executor/cursor_executor.go b/internal/runtime/executor/cursor_executor.go
index d288e8fa..643adf68 100644
--- a/internal/runtime/executor/cursor_executor.go
+++ b/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,15 +320,23 @@ 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))
@@ -333,8 +344,13 @@ func (e *CursorExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, r
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":""}`, "")
- }
- sendChunkSwitchable(fmt.Sprintf(`{"content":%s}`, jsonString(text)), "")
- } else {
- if thinkingActive {
- thinkingActive = false
- sendChunkSwitchable(`{"content":""}`, "")
- }
- 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 ... 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":""}`, "")
- }
- 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":""}`, "")
- }
// 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 "{}"
diff --git a/internal/runtime/executor/cursor_executor_test.go b/internal/runtime/executor/cursor_executor_test.go
new file mode 100644
index 00000000..7e4bfed0
--- /dev/null
+++ b/internal/runtime/executor/cursor_executor_test.go
@@ -0,0 +1,84 @@
+package executor
+
+import (
+ "context"
+ "testing"
+ "time"
+)
+
+func TestNormalizeToolCallID(t *testing.T) {
+ input := "call-cce860e6-ab07-414d-812c-785db35b17ca-4\nfc_d2335004-a95f-93b4-977b-e9eee6316be7_0"
+ want := "call-cce860e6-ab07-414d-812c-785db35b17ca-4fc_d2335004-a95f-93b4-977b-e9eee6316be7_0"
+
+ if got := normalizeToolCallID(input); got != want {
+ t.Fatalf("normalizeToolCallID() = %q, want %q", got, want)
+ }
+}
+
+func TestCursorStreamCoalescerBatchesAdjacentDeltas(t *testing.T) {
+ type emittedDelta struct {
+ text string
+ isThinking bool
+ }
+
+ emitted := make(chan emittedDelta, 4)
+ coalescer := newCursorStreamCoalescer(
+ context.Background(),
+ time.Hour,
+ func(text string, isThinking bool) {
+ emitted <- emittedDelta{text: text, isThinking: isThinking}
+ },
+ )
+
+ coalescer.push("first", true)
+ coalescer.push(" second", true)
+ coalescer.push(" third", true)
+ coalescer.push("answer", false)
+ coalescer.close()
+ close(emitted)
+
+ got := make([]emittedDelta, 0, 3)
+ for delta := range emitted {
+ got = append(got, delta)
+ }
+ want := []emittedDelta{
+ {text: "first", isThinking: true},
+ {text: " second third", isThinking: true},
+ {text: "answer", isThinking: false},
+ }
+ if len(got) != len(want) {
+ t.Fatalf("got %d emitted deltas, want %d: %#v", len(got), len(want), got)
+ }
+ for index := range want {
+ if got[index] != want[index] {
+ t.Fatalf("delta %d = %#v, want %#v", index, got[index], want[index])
+ }
+ }
+}
+
+func TestCursorStreamCoalescerFlushesOnTimer(t *testing.T) {
+ emitted := make(chan string, 2)
+ coalescer := newCursorStreamCoalescer(
+ context.Background(),
+ 5*time.Millisecond,
+ func(text string, _ bool) {
+ emitted <- text
+ },
+ )
+ defer coalescer.close()
+
+ coalescer.push("first", false)
+ if got := <-emitted; got != "first" {
+ t.Fatalf("first emitted delta = %q, want %q", got, "first")
+ }
+
+ coalescer.push(" second", false)
+ select {
+ case got := <-emitted:
+ if got != " second" {
+ t.Fatalf("timed emitted delta = %q, want %q", got, " second")
+ }
+ case <-time.After(250 * time.Millisecond):
+ t.Fatal("timed delta was not emitted")
+ }
+}