From beeb9f765372a5ae6b295bdeb7064180bb22cecd Mon Sep 17 00:00:00 2001 From: ssakone Date: Fri, 10 Jul 2026 20:37:49 +0000 Subject: [PATCH 1/2] fix(executor): normalize reasoning and tool streams Emit structured reasoning, sanitize tool-call identifiers, avoid duplicate stream terminators, and report non-stream usage consistently. --- internal/runtime/executor/cursor_executor.go | 69 ++++++++++++------- .../runtime/executor/cursor_executor_test.go | 12 ++++ 2 files changed, 57 insertions(+), 24 deletions(-) create mode 100644 internal/runtime/executor/cursor_executor_test.go diff --git a/internal/runtime/executor/cursor_executor.go b/internal/runtime/executor/cursor_executor.go index d288e8fa..0b79d806 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" @@ -317,15 +318,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 +342,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 +574,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: @@ -590,27 +605,27 @@ func (e *CursorExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.A streamErr := processH2SessionFrames(sessionCtx, stream, params.BlobStore, params.McpTools, 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(`{"role":"assistant","content":""}`, "") } - sendChunkSwitchable(fmt.Sprintf(`{"content":%s}`, jsonString(text)), "") + sendChunkSwitchable(fmt.Sprintf(`{"reasoning_content":%s}`, jsonString(text)), "") } else { - if thinkingActive { - thinkingActive = false - sendChunkSwitchable(`{"content":""}`, "") - } + thinkingActive = false sendChunkSwitchable(fmt.Sprintf(`{"content":%s}`, jsonString(text)), "") } }, 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)) + 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"`) @@ -696,9 +711,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}`, @@ -958,7 +970,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 +1139,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 +1473,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..a33b0aee --- /dev/null +++ b/internal/runtime/executor/cursor_executor_test.go @@ -0,0 +1,12 @@ +package executor + +import "testing" + +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) + } +} From 4bd5387164a6b05e016c34d934b4d457ddf62cd3 Mon Sep 17 00:00:00 2001 From: ssakone Date: Sat, 11 Jul 2026 01:44:58 +0000 Subject: [PATCH 2/2] perf(cursor): smooth streamed deltas Coalesce bursty protobuf updates at display cadence while preserving immediate first output and tool-call ordering. --- internal/runtime/executor/cursor_executor.go | 197 ++++++++++++++++-- .../runtime/executor/cursor_executor_test.go | 74 ++++++- 2 files changed, 251 insertions(+), 20 deletions(-) diff --git a/internal/runtime/executor/cursor_executor.go b/internal/runtime/executor/cursor_executor.go index 0b79d806..643adf68 100644 --- a/internal/runtime/executor/cursor_executor.go +++ b/internal/runtime/executor/cursor_executor.go @@ -39,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. @@ -603,26 +605,36 @@ 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) { - // 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)), "") + 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) { + // 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)) @@ -669,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, @@ -686,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. @@ -829,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 diff --git a/internal/runtime/executor/cursor_executor_test.go b/internal/runtime/executor/cursor_executor_test.go index a33b0aee..7e4bfed0 100644 --- a/internal/runtime/executor/cursor_executor_test.go +++ b/internal/runtime/executor/cursor_executor_test.go @@ -1,6 +1,10 @@ package executor -import "testing" +import ( + "context" + "testing" + "time" +) func TestNormalizeToolCallID(t *testing.T) { input := "call-cce860e6-ab07-414d-812c-785db35b17ca-4\nfc_d2335004-a95f-93b4-977b-e9eee6316be7_0" @@ -10,3 +14,71 @@ func TestNormalizeToolCallID(t *testing.T) { 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") + } +}