package executor import (	"bytes"	"context"	"crypto/sha256"	"errors"	"crypto/tls"	"encoding/base64"	"encoding/hex"	"encoding/json"	"fmt"	"io"	"net/http"	"strings"	"sync"	"time" 	"github.com/google/uuid"	cursorauth "github.com/router-for-me/CLIProxyAPI/v6/internal/auth/cursor"	cursorproto "github.com/router-for-me/CLIProxyAPI/v6/internal/auth/cursor/proto"	"github.com/router-for-me/CLIProxyAPI/v6/internal/config"	"github.com/router-for-me/CLIProxyAPI/v6/internal/registry"	cliproxyauth "github.com/router-for-me/CLIProxyAPI/v6/sdk/cliproxy/auth"	cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v6/sdk/cliproxy/executor"	sdktranslator "github.com/router-for-me/CLIProxyAPI/v6/sdk/translator"	log "github.com/sirupsen/logrus"	"github.com/tidwall/gjson"	"golang.org/x/net/http2") const (	cursorAPIURL        = "https://api2.cursor.sh"	cursorRunPath       = "/agent.v1.AgentService/Run"	cursorModelsPath    = "/agent.v1.AgentService/GetUsableModels"	cursorClientVersion = "cli-2026.02.13-41ac335"	cursorAuthType      = "cursor"	cursorHeartbeatInterval = 5 * time.Second	cursorSessionTTL      = 5 * time.Minute	cursorCheckpointTTL   = 30 * time.Minute) // CursorExecutor handles requests to the Cursor API via Connect+Protobuf protocol.type CursorExecutor struct {	cfg         *config.Config	mu          sync.Mutex	sessions    map[string]*cursorSession	checkpoints map[string]*savedCheckpoint // keyed by conversationId} // savedCheckpoint stores the server's conversation_checkpoint_update for reuse.type savedCheckpoint struct {	data      []byte            // raw ConversationStateStructure protobuf bytes	blobStore map[string][]byte // blobs referenced by the checkpoint	authID    string            // auth that produced this checkpoint (checkpoint is auth-specific)	updatedAt time.Time} type cursorSession struct {	stream       *cursorproto.H2Stream	blobStore    map[string][]byte	mcpTools     []cursorproto.McpToolDef	pending      []pendingMcpExec	cancel       context.CancelFunc // cancels the session-scoped heartbeat (NOT tied to HTTP request)	createdAt    time.Time	authID       string // auth file ID that created this session (for multi-account isolation)	toolResultCh chan []toolResultInfo                // receives tool results from the next HTTP request	resumeOutCh  chan cliproxyexecutor.StreamChunk    // output channel for resumed response	switchOutput func(ch chan cliproxyexecutor.StreamChunk) // callback to switch output channel} type pendingMcpExec struct {	ExecMsgId  uint32	ExecId     string	ToolCallId string	ToolName   string	Args       string // JSON-encoded args} // NewCursorExecutor constructs a new executor instance.func NewCursorExecutor(cfg *config.Config) *CursorExecutor {	e := &CursorExecutor{		cfg:         cfg,		sessions:    make(map[string]*cursorSession),		checkpoints: make(map[string]*savedCheckpoint),	}	go e.cleanupLoop()	return e} // Identifier implements ProviderExecutor.func (e *CursorExecutor) Identifier() string { return cursorAuthType } // CloseExecutionSession implements ExecutionSessionCloser.func (e *CursorExecutor) CloseExecutionSession(sessionID string) {	e.mu.Lock()	defer e.mu.Unlock()	if sessionID == cliproxyauth.CloseAllExecutionSessionsID {		for k, s := range e.sessions {			s.cancel()			delete(e.sessions, k)		}		return	}	if s, ok := e.sessions[sessionID]; ok {		s.cancel()		delete(e.sessions, sessionID)	}} func (e *CursorExecutor) cleanupLoop() {	ticker := time.NewTicker(1 * time.Minute)	defer ticker.Stop()	for range ticker.C {		e.mu.Lock()		for k, s := range e.sessions {			if time.Since(s.createdAt) > cursorSessionTTL {				s.cancel()				delete(e.sessions, k)			}		}		for k, cp := range e.checkpoints {			if time.Since(cp.updatedAt) > cursorCheckpointTTL {				delete(e.checkpoints, k)			}		}		e.mu.Unlock()	}} // findSessionByConversationLocked searches for a session matching the given// conversationId regardless of authID. Used to find and clean up stale sessions// from a previous auth after quota failover. Caller must hold e.mu.func (e *CursorExecutor) findSessionByConversationLocked(convId string) string {	suffix := ":" + convId	for k := range e.sessions {		if strings.HasSuffix(k, suffix) {			return k		}	}	return ""} // cursorStatusErr implements the StatusError and RetryAfter interfaces so the// conductor can classify Cursor errors (e.g. 429 → quota cooldown).type cursorStatusErr struct {	code int	msg  string} func (e cursorStatusErr) Error() string             { return e.msg }func (e cursorStatusErr) StatusCode() int            { return e.code }func (e cursorStatusErr) RetryAfter() *time.Duration { return nil } // no retry-after info from Cursor; conductor uses exponential backoff // classifyCursorError maps Cursor Connect/H2 errors to HTTP status codes.// Layer 1: precise match on ConnectError.Code (gRPC standard codes).// Layer 2: fuzzy string match for H2 frame errors and unknown formats.// Unclassified errors pass through unchanged.func classifyCursorError(err error) error {	if err == nil {		return nil	} 	// Layer 1: structured ConnectError from ParseConnectEndStream	var ce *cursorproto.ConnectError	if errors.As(err, &ce) {		log.Infof("cursor: Connect error code=%q message=%q", ce.Code, ce.Message)		switch ce.Code {		case "resource_exhausted":			return cursorStatusErr{code: 429, msg: err.Error()}		case "unauthenticated":			return cursorStatusErr{code: 401, msg: err.Error()}		case "permission_denied":			return cursorStatusErr{code: 403, msg: err.Error()}		case "unavailable":			return cursorStatusErr{code: 503, msg: err.Error()}		case "internal":			return cursorStatusErr{code: 500, msg: err.Error()}		default:			// Unknown Connect code — log for observation, treat as 502			return cursorStatusErr{code: 502, msg: err.Error()}		}	} 	// Layer 2: fuzzy match for H2 errors and unstructured messages	msg := strings.ToLower(err.Error())	switch {	case strings.Contains(msg, "rate limit") || strings.Contains(msg, "quota") ||		strings.Contains(msg, "too many"):		return cursorStatusErr{code: 429, msg: err.Error()}	case strings.Contains(msg, "rst_stream") || strings.Contains(msg, "goaway"):		return cursorStatusErr{code: 502, msg: err.Error()}	} 	return err} // PrepareRequest implements ProviderExecutor (for HttpRequest support).func (e *CursorExecutor) PrepareRequest(req *http.Request, auth *cliproxyauth.Auth) error {	token := cursorAccessToken(auth)	if token == "" {		return fmt.Errorf("cursor: access token not found")	}	req.Header.Set("Authorization", "Bearer "+token)	return nil} // HttpRequest injects credentials and executes the request.func (e *CursorExecutor) HttpRequest(ctx context.Context, auth *cliproxyauth.Auth, req *http.Request) (*http.Response, error) {	if req == nil {		return nil, fmt.Errorf("cursor: request is nil")	}	if err := e.PrepareRequest(req, auth); err != nil {		return nil, err	}	return http.DefaultClient.Do(req)} // CountTokens estimates token count locally using tiktoken.func (e *CursorExecutor) CountTokens(_ context.Context, _ *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) {	defer func() {		if err != nil {			log.Warnf("cursor CountTokens error: %v", err)		} else {			log.Debugf("cursor CountTokens: model=%s result=%s", req.Model, string(resp.Payload))		}	}()	model := gjson.GetBytes(req.Payload, "model").String()	if model == "" {		model = req.Model	} 	enc, err := getTokenizer(model)	if err != nil {		// Fallback: return zero tokens rather than error (avoids 502)		return cliproxyexecutor.Response{Payload: buildOpenAIUsageJSON(0)}, nil	} 	// Detect format: Claude (/v1/messages) vs OpenAI (/v1/chat/completions)	var count int64	if gjson.GetBytes(req.Payload, "system").Exists() || opts.SourceFormat.String() == "claude" {		count, _ = countClaudeChatTokens(enc, req.Payload)	} else {		count, _ = countOpenAIChatTokens(enc, req.Payload)	} 	return cliproxyexecutor.Response{Payload: buildOpenAIUsageJSON(count)}, nil} // Refresh attempts to refresh the Cursor access token.func (e *CursorExecutor) Refresh(ctx context.Context, auth *cliproxyauth.Auth) (*cliproxyauth.Auth, error) {	refreshToken := cursorRefreshToken(auth)	if refreshToken == "" {		return nil, fmt.Errorf("cursor: no refresh token available")	} 	tokens, err := cursorauth.RefreshToken(ctx, refreshToken)	if err != nil {		return nil, err	} 	expiresAt := cursorauth.GetTokenExpiry(tokens.AccessToken) 	newAuth := auth.Clone()	newAuth.Metadata["access_token"] = tokens.AccessToken	newAuth.Metadata["refresh_token"] = tokens.RefreshToken	newAuth.Metadata["expires_at"] = expiresAt.Format(time.RFC3339)	return newAuth, nil} // Execute handles non-streaming requests.func (e *CursorExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) {	log.Debugf("cursor Execute: model=%s sourceFormat=%s payloadLen=%d", req.Model, opts.SourceFormat, len(req.Payload))	defer func() {		if r := recover(); r != nil {			log.Errorf("cursor Execute PANIC: %v", r)			err = fmt.Errorf("cursor: internal panic: %v", r)		}		if err != nil {			log.Warnf("cursor Execute error: %v", err)		}	}()	accessToken := cursorAccessToken(auth)	if accessToken == "" {		return resp, fmt.Errorf("cursor: access token not found")	} 	// Translate input to OpenAI format if needed (e.g. Claude /v1/messages format)	from := opts.SourceFormat	to := sdktranslator.FromString("openai")	payload := req.Payload	if from.String() != "" && from.String() != "openai" {		payload = sdktranslator.TranslateRequest(from, to, req.Model, bytes.Clone(payload), false)	} 	parsed := parseOpenAIRequest(payload)	ccSessId := extractClaudeCodeSessionId(req.Payload)	conversationId := deriveConversationId(apiKeyFromContext(ctx), ccSessId, parsed.SystemPrompt)	params := buildRunRequestParams(parsed, conversationId) 	requestBytes := cursorproto.EncodeRunRequest(params)	framedRequest := cursorproto.FrameConnectMessage(requestBytes, 0) 	stream, err := openCursorH2Stream(accessToken)	if err != nil {		return resp, err	}	defer stream.Close() 	// Send the request frame	if err := stream.Write(framedRequest); err != nil {		return resp, fmt.Errorf("cursor: failed to send request: %w", err)	} 	// Start heartbeat	sessionCtx, sessionCancel := context.WithCancel(ctx)	defer sessionCancel()	go cursorH2Heartbeat(sessionCtx, stream) 	// Collect full text from streaming response	var fullText strings.Builder	if streamErr := processH2SessionFrames(sessionCtx, stream, params.BlobStore, nil,		func(text string, isThinking bool) {			fullText.WriteString(text)		},		nil,		nil,		nil, // tokenUsage - non-streaming		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())) 	// Translate response back to source format if needed	result := []byte(openaiResp)	if from.String() != "" && from.String() != "openai" {		var param any		result = sdktranslator.TranslateNonStream(ctx, to, from, req.Model, bytes.Clone(opts.OriginalRequest), payload, result, &param)	}	resp.Payload = result	return resp, nil} // ExecuteStream handles streaming requests.// It supports MCP tool call sessions: when Cursor returns an MCP tool call,// the H2 stream is kept alive. When Claude Code returns the tool result in// the next request, the result is sent back on the same stream (session resume).// This mirrors the activeSessions/resumeWithToolResults pattern in cursor-fetch.ts.func (e *CursorExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (_ *cliproxyexecutor.StreamResult, err error) {	log.Debugf("cursor ExecuteStream: model=%s sourceFormat=%s payloadLen=%d", req.Model, opts.SourceFormat, len(req.Payload))	defer func() {		if r := recover(); r != nil {			log.Errorf("cursor ExecuteStream PANIC: %v", r)			err = fmt.Errorf("cursor: internal panic: %v", r)		}		if err != nil {			log.Warnf("cursor ExecuteStream error: %v", err)		}	}()	accessToken := cursorAccessToken(auth)	if accessToken == "" {		return nil, fmt.Errorf("cursor: access token not found")	} 	// Extract session_id from metadata BEFORE translation (translation strips metadata)	ccSessionId := extractClaudeCodeSessionId(req.Payload)	if ccSessionId == "" && len(opts.OriginalRequest) > 0 {		ccSessionId = extractClaudeCodeSessionId(opts.OriginalRequest)	} 	// Translate input to OpenAI format if needed	from := opts.SourceFormat	to := sdktranslator.FromString("openai")	payload := req.Payload	originalPayload := bytes.Clone(req.Payload)	if len(opts.OriginalRequest) > 0 {		originalPayload = bytes.Clone(opts.OriginalRequest)	}	if from.String() != "" && from.String() != "openai" {		log.Debugf("cursor: translating request from %s to openai", from)		payload = sdktranslator.TranslateRequest(from, to, req.Model, bytes.Clone(payload), true)		log.Debugf("cursor: translated payload len=%d", len(payload))	} 	parsed := parseOpenAIRequest(payload)	log.Debugf("cursor: parsed request: model=%s userText=%d chars, turns=%d, tools=%d, toolResults=%d",		parsed.Model, len(parsed.UserText), len(parsed.Turns), len(parsed.Tools), len(parsed.ToolResults)) 	conversationId := deriveConversationId(apiKeyFromContext(ctx), ccSessionId, parsed.SystemPrompt)	authID := auth.ID // e.g. "cursor.json" or "cursor-account2.json"	log.Debugf("cursor: conversationId=%s authID=%s", conversationId, authID) 	// Session key includes authID (H2 stream is auth-specific, not transferable).	// Checkpoint key uses conversationId only — allows detecting auth migration.	sessionKey := authID + ":" + conversationId	checkpointKey := conversationId	needsTranslate := from.String() != "" && from.String() != "openai" 	// Check if we can resume an existing session with tool results	if len(parsed.ToolResults) > 0 {		e.mu.Lock()		session, hasSession := e.sessions[sessionKey]		if hasSession {			delete(e.sessions, sessionKey)		}		// If no session found for current auth, check for stale sessions from		// a different auth on the same conversation (quota failover scenario).		// Clean them up since the H2 stream belongs to the old account.		if !hasSession {			if oldKey := e.findSessionByConversationLocked(conversationId); oldKey != "" {				oldSession := e.sessions[oldKey]				log.Infof("cursor: cleaning up stale session from auth %s for conv=%s (auth migrated to %s)", oldSession.authID, conversationId, authID)				oldSession.cancel()				if oldSession.stream != nil {					oldSession.stream.Close()				}				delete(e.sessions, oldKey)			}		}		e.mu.Unlock() 		if hasSession && session.stream != nil && session.authID == authID {			log.Debugf("cursor: resuming session %s with %d tool results", sessionKey, len(parsed.ToolResults))			return e.resumeWithToolResults(ctx, session, parsed, from, to, req, originalPayload, payload, needsTranslate)		}		if hasSession && session.authID != authID {			log.Warnf("cursor: session %s belongs to auth %s, but request is from %s — skipping resume", sessionKey, session.authID, authID)		}	} 	// Clean up any stale session for this key (or from a previous auth on same conversation)	e.mu.Lock()	if old, ok := e.sessions[sessionKey]; ok {		old.cancel()		delete(e.sessions, sessionKey)	} else if oldKey := e.findSessionByConversationLocked(conversationId); oldKey != "" {		old := e.sessions[oldKey]		old.cancel()		if old.stream != nil {			old.stream.Close()		}		delete(e.sessions, oldKey)	}	e.mu.Unlock() 	// Look up saved checkpoint for this conversation (keyed by conversationId only).	// Checkpoint is auth-specific: if auth changed (e.g. quota exhaustion failover),	// the old checkpoint is useless on the new account — discard and flatten.	e.mu.Lock()	saved, hasCheckpoint := e.checkpoints[checkpointKey]	e.mu.Unlock() 	params := buildRunRequestParams(parsed, conversationId) 	if hasCheckpoint && saved.data != nil && saved.authID == authID {		// Same auth — use checkpoint normally		log.Debugf("cursor: using saved checkpoint (%d bytes) for conv=%s auth=%s", len(saved.data), checkpointKey, authID)		params.RawCheckpoint = saved.data		// Merge saved blobStore into params		if params.BlobStore == nil {			params.BlobStore = make(map[string][]byte)		}		for k, v := range saved.blobStore {			if _, exists := params.BlobStore[k]; !exists {				params.BlobStore[k] = v			}		}	} else if hasCheckpoint && saved.data != nil && saved.authID != authID {		// Auth changed (quota failover) — checkpoint is not portable across accounts.		// Discard and flatten conversation history into userText.		log.Infof("cursor: auth migrated (%s → %s) for conv=%s, discarding checkpoint and flattening context", saved.authID, authID, checkpointKey)		e.mu.Lock()		delete(e.checkpoints, checkpointKey)		e.mu.Unlock()		if len(parsed.ToolResults) > 0 || len(parsed.Turns) > 0 {			flattenConversationIntoUserText(parsed)			params = buildRunRequestParams(parsed, conversationId)		}	} else if len(parsed.ToolResults) > 0 || len(parsed.Turns) > 0 {		// Fallback: no checkpoint available (cold resume / proxy restart).		// Flatten the full conversation history (including tool interactions) into userText.		// Cursor's turns encoding is not reliably read by the model, but userText always works.		log.Debugf("cursor: no checkpoint, flattening %d turns + %d tool results into userText", len(parsed.Turns), len(parsed.ToolResults))		flattenConversationIntoUserText(parsed)		params = buildRunRequestParams(parsed, conversationId)	}	requestBytes := cursorproto.EncodeRunRequest(params)	framedRequest := cursorproto.FrameConnectMessage(requestBytes, 0) 	stream, err := openCursorH2Stream(accessToken)	if err != nil {		return nil, err	} 	if err := stream.Write(framedRequest); err != nil {

Lines 1–500 of 1719. Load more