package executor import (	"bufio"	"bytes"	"context"	"encoding/json"	"fmt"	"io"	"mime"	"mime/multipart"	"net/http"	"net/textproto"	"strconv"	"strings"	"time" 	"github.com/google/uuid"	"github.com/router-for-me/CLIProxyAPI/v7/internal/config"	"github.com/router-for-me/CLIProxyAPI/v7/internal/runtime/executor/helps"	"github.com/router-for-me/CLIProxyAPI/v7/internal/thinking"	"github.com/router-for-me/CLIProxyAPI/v7/internal/util"	cliproxyauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth"	cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor"	sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator"	log "github.com/sirupsen/logrus"	"github.com/tidwall/gjson"	"github.com/tidwall/sjson") const (	openAICompatImageHandlerType            = "openai-image"	openAICompatImagesGenerationsPath       = "/images/generations"	openAICompatImagesEditsPath             = "/images/edits"	openAICompatDefaultImageEndpoint        = openAICompatImagesGenerationsPath	openAICompatMultipartMemory       int64 = 32 << 20) // OpenAICompatExecutor implements a stateless executor for OpenAI-compatible providers.// It performs request/response translation and executes against the provider base URL// using per-auth credentials (API key) and per-auth HTTP transport (proxy) from context.type OpenAICompatExecutor struct {	provider string	cfg      *config.Config} // NewOpenAICompatExecutor creates an executor bound to a provider key (e.g., "openrouter").func NewOpenAICompatExecutor(provider string, cfg *config.Config) *OpenAICompatExecutor {	return &OpenAICompatExecutor{provider: provider, cfg: cfg}} // Identifier implements cliproxyauth.ProviderExecutor.func (e *OpenAICompatExecutor) Identifier() string { return e.provider } // PrepareRequest injects OpenAI-compatible credentials into the outgoing HTTP request.func (e *OpenAICompatExecutor) PrepareRequest(req *http.Request, auth *cliproxyauth.Auth) error {	if req == nil {		return nil	}	_, apiKey := e.resolveCredentials(auth)	if strings.TrimSpace(apiKey) != "" {		req.Header.Set("Authorization", "Bearer "+apiKey)	}	var attrs map[string]string	if auth != nil {		attrs = auth.Attributes	}	util.ApplyCustomHeadersFromAttrs(req, attrs)	return nil} // HttpRequest injects OpenAI-compatible credentials into the request and executes it.func (e *OpenAICompatExecutor) HttpRequest(ctx context.Context, auth *cliproxyauth.Auth, req *http.Request) (*http.Response, error) {	if req == nil {		return nil, fmt.Errorf("openai compat executor: request is nil")	}	if ctx == nil {		ctx = req.Context()	}	httpReq := req.WithContext(ctx)	if err := e.PrepareRequest(httpReq, auth); err != nil {		return nil, err	}	httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0)	return httpClient.Do(httpReq)} func (e *OpenAICompatExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) {	ctx = helps.EnsureSessionContext(ctx, opts, req.Payload)	if endpointPath := openAICompatImageEndpointPath(opts); endpointPath != "" {		return e.executeImages(ctx, auth, req, opts, endpointPath)	} 	baseModel := thinking.ParseSuffix(req.Model).ModelName 	reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)	defer reporter.TrackFailure(ctx, &err) 	baseURL, apiKey := e.resolveCredentials(auth)	if baseURL == "" {		err = statusErr{code: http.StatusUnauthorized, msg: "missing provider baseURL"}		return	} 	from := opts.SourceFormat	responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts)	to := sdktranslator.FromString("openai")	endpoint := "/chat/completions"	if opts.Alt == "responses/compact" {		to = sdktranslator.FromString("openai-response")		endpoint = "/responses/compact"	}	originalPayloadSource := req.Payload	if len(opts.OriginalRequest) > 0 {		originalPayloadSource = opts.OriginalRequest	}	originalPayload := originalPayloadSource	isCompat := helps.APIKeyModelIsCompat(req)	originalTranslated := helps.TranslateRequestWithAPIKeyModelCompatibility(ctx, opts.Headers, e.cfg, from, to, baseModel, originalPayload, opts.Stream, isCompat)	translated := helps.TranslateRequestWithAPIKeyModelCompatibility(ctx, opts.Headers, e.cfg, from, to, baseModel, req.Payload, opts.Stream, isCompat) 	translated, err = helps.ApplyRequestThinking(translated, req, opts, from.String(), to.String(), e.Identifier())	if err != nil {		return resp, err	} 	requestedModel := helps.PayloadRequestedModel(opts, req.Model)	requestPath := helps.PayloadRequestPath(opts)	translated = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", translated, originalTranslated, requestedModel, requestPath, opts.Headers)	if helps.ShouldNormalizeOpenAIToolResultsForModel(e.resolveCompatConfig(auth), baseModel, requestedModel) {		translated = helps.NormalizeOpenAIToolResultsTextOnly(translated)	}	if opts.Alt != "responses/compact" {		translated, err = e.applyPromptCacheKey(ctx, auth, from, baseModel, req, opts, translated)		if err != nil {			return resp, err		}	}	if opts.Alt == "responses/compact" {		if updated, errDelete := sjson.DeleteBytes(translated, "stream"); errDelete == nil {			translated = updated		}		translated = sanitizeOpenAIResponsesReasoningEncryptedContent(ctx, "openai compat executor", translated)	}	reporter.SetTranslatedReasoningEffort(translated, to.String()) 	url := strings.TrimSuffix(baseURL, "/") + endpoint	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(translated))	if err != nil {		return resp, err	}	httpReq.Header.Set("Content-Type", "application/json")	if apiKey != "" {		httpReq.Header.Set("Authorization", "Bearer "+apiKey)	}	httpReq.Header.Set("User-Agent", "cli-proxy-openai-compat")	var attrs map[string]string	if auth != nil {		attrs = auth.Attributes	}	util.ApplyCustomHeadersFromAttrs(httpReq, attrs, opts.Headers)	var authID, authLabel, authType, authValue string	if auth != nil {		authID = auth.ID		authLabel = auth.Label		authType, authValue = auth.AccountInfo()	}	helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{		URL:       url,		Method:    http.MethodPost,		Headers:   httpReq.Header.Clone(),		Body:      translated,		Provider:  e.Identifier(),		AuthID:    authID,		AuthLabel: authLabel,		AuthType:  authType,		AuthValue: authValue,	}) 	httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0)	httpClient = reporter.TrackHTTPClient(httpClient)	httpResp, err := httpClient.Do(httpReq)	if err != nil {		helps.RecordAPIResponseError(ctx, e.cfg, err)		return resp, err	}	defer func() {		if errClose := httpResp.Body.Close(); errClose != nil {			log.Errorf("openai compat executor: close response body error: %v", errClose)		}	}()	helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone())	if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 {		b, _ := io.ReadAll(httpResp.Body)		helps.AppendAPIResponseChunk(ctx, e.cfg, b)		helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b))		err = newOpenAICompatStatusError(httpResp.StatusCode, httpResp.Header, b)		return resp, err	}	body, err := io.ReadAll(httpResp.Body)	if err != nil {		helps.RecordAPIResponseError(ctx, e.cfg, err)		return resp, err	}	helps.AppendAPIResponseChunk(ctx, e.cfg, body)	reporter.Publish(ctx, helps.ParseOpenAIUsage(body))	// Ensure we at least record the request even if upstream doesn't return usage	reporter.EnsurePublished(ctx)	// Translate response back to source format when needed	var param any	out := sdktranslator.TranslateNonStream(ctx, to, responseFormat, req.Model, opts.OriginalRequest, translated, body, &param)	if responseFormat == sdktranslator.FormatOpenAIResponse {		out = helps.EnsureResponsesUsageDetails(out)	}	resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()}	return resp, nil} func (e *OpenAICompatExecutor) executeImages(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, endpointPath string) (resp cliproxyexecutor.Response, err error) {	baseModel := thinking.ParseSuffix(req.Model).ModelName 	reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)	defer reporter.TrackFailure(ctx, &err) 	baseURL, apiKey := e.resolveCredentials(auth)	if baseURL == "" {		err = statusErr{code: http.StatusUnauthorized, msg: "missing provider baseURL"}		return resp, err	} 	payload, contentType, errPrepare := prepareOpenAICompatImagesPayload(req.Payload, baseModel, opts.Headers.Get("Content-Type"), false)	if errPrepare != nil {		err = errPrepare		return resp, err	}	if contentType == "" {		contentType = "application/json"	}	reporter.SetTranslatedReasoningEffort(payload, "openai") 	url := strings.TrimSuffix(baseURL, "/") + endpointPath	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(payload))	if err != nil {		return resp, err	}	httpReq.Header.Set("Content-Type", contentType)	if apiKey != "" {		httpReq.Header.Set("Authorization", "Bearer "+apiKey)	}	httpReq.Header.Set("User-Agent", "cli-proxy-openai-compat")	var attrs map[string]string	if auth != nil {		attrs = auth.Attributes	}	util.ApplyCustomHeadersFromAttrs(httpReq, attrs, opts.Headers)	var authID, authLabel, authType, authValue string	if auth != nil {		authID = auth.ID		authLabel = auth.Label		authType, authValue = auth.AccountInfo()	}	helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{		URL:       url,		Method:    http.MethodPost,		Headers:   httpReq.Header.Clone(),		Body:      payload,		Provider:  e.Identifier(),		AuthID:    authID,		AuthLabel: authLabel,		AuthType:  authType,		AuthValue: authValue,	}) 	httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0)	httpClient = reporter.TrackHTTPClient(httpClient)	httpResp, err := httpClient.Do(httpReq)	if err != nil {		helps.RecordAPIResponseError(ctx, e.cfg, err)		return resp, err	}	defer func() {		if errClose := httpResp.Body.Close(); errClose != nil {			log.Errorf("openai compat executor: close response body error: %v", errClose)		}	}()	helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone()) 	body, errRead := io.ReadAll(httpResp.Body)	if errRead != nil {		helps.RecordAPIResponseError(ctx, e.cfg, errRead)		err = errRead		return resp, err	}	helps.AppendAPIResponseChunk(ctx, e.cfg, body) 	if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 {		helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), body))		err = newOpenAICompatStatusError(httpResp.StatusCode, httpResp.Header, body)		return resp, err	} 	reporter.Publish(ctx, helps.ParseOpenAIUsage(body))	reporter.EnsurePublished(ctx)	resp = cliproxyexecutor.Response{Payload: body, Headers: httpResp.Header.Clone()}	return resp, nil} func (e *OpenAICompatExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (_ *cliproxyexecutor.StreamResult, err error) {	ctx = helps.EnsureSessionContext(ctx, opts, req.Payload)	if endpointPath := openAICompatImageEndpointPath(opts); endpointPath != "" {		return e.executeImagesStream(ctx, auth, req, opts, endpointPath)	} 	baseModel := thinking.ParseSuffix(req.Model).ModelName 	reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)	defer reporter.TrackFailure(ctx, &err) 	baseURL, apiKey := e.resolveCredentials(auth)	if baseURL == "" {		err = statusErr{code: http.StatusUnauthorized, msg: "missing provider baseURL"}		return nil, err	} 	from := opts.SourceFormat	responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts)	to := sdktranslator.FromString("openai")	originalPayloadSource := req.Payload	if len(opts.OriginalRequest) > 0 {		originalPayloadSource = opts.OriginalRequest	}	originalPayload := originalPayloadSource	isCompat := helps.APIKeyModelIsCompat(req)	originalTranslated := helps.TranslateRequestWithAPIKeyModelCompatibility(ctx, opts.Headers, e.cfg, from, to, baseModel, originalPayload, true, isCompat)	translated := helps.TranslateRequestWithAPIKeyModelCompatibility(ctx, opts.Headers, e.cfg, from, to, baseModel, req.Payload, true, isCompat) 	translated, err = helps.ApplyRequestThinking(translated, req, opts, from.String(), to.String(), e.Identifier())	if err != nil {		return nil, err	} 	requestedModel := helps.PayloadRequestedModel(opts, req.Model)	requestPath := helps.PayloadRequestPath(opts)	translated = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", translated, originalTranslated, requestedModel, requestPath, opts.Headers)	if helps.ShouldNormalizeOpenAIToolResultsForModel(e.resolveCompatConfig(auth), baseModel, requestedModel) {		translated = helps.NormalizeOpenAIToolResultsTextOnly(translated)	}	if opts.Alt != "responses/compact" {		translated, err = e.applyPromptCacheKey(ctx, auth, from, baseModel, req, opts, translated)		if err != nil {			return nil, err		}	} 	// Request usage data in the final streaming chunk so that token statistics	// are captured even when the upstream is an OpenAI-compatible provider.	translated = helps.SetBoolIfDifferent(translated, "stream_options.include_usage", true)	reporter.SetTranslatedReasoningEffort(translated, to.String()) 	url := strings.TrimSuffix(baseURL, "/") + "/chat/completions"	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(translated))	if err != nil {		return nil, err	}	httpReq.Header.Set("Content-Type", "application/json")	if apiKey != "" {		httpReq.Header.Set("Authorization", "Bearer "+apiKey)	}	httpReq.Header.Set("User-Agent", "cli-proxy-openai-compat")	var attrs map[string]string	if auth != nil {		attrs = auth.Attributes	}	util.ApplyCustomHeadersFromAttrs(httpReq, attrs, opts.Headers)	httpReq.Header.Set("Accept", "text/event-stream")	httpReq.Header.Set("Cache-Control", "no-cache")	var authID, authLabel, authType, authValue string	if auth != nil {		authID = auth.ID		authLabel = auth.Label		authType, authValue = auth.AccountInfo()	}	helps.RecordAPIRequest(ctx, e.cfg, helps.UpstreamRequestLog{		URL:       url,		Method:    http.MethodPost,		Headers:   httpReq.Header.Clone(),		Body:      translated,		Provider:  e.Identifier(),		AuthID:    authID,		AuthLabel: authLabel,		AuthType:  authType,		AuthValue: authValue,	}) 	httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0)	httpClient = reporter.TrackHTTPClient(httpClient)	httpResp, err := httpClient.Do(httpReq)	if err != nil {		helps.RecordAPIResponseError(ctx, e.cfg, err)		return nil, err	}	helps.RecordAPIResponseMetadata(ctx, e.cfg, httpResp.StatusCode, httpResp.Header.Clone())	if httpResp.StatusCode < 200 || httpResp.StatusCode >= 300 {		b, _ := io.ReadAll(httpResp.Body)		helps.AppendAPIResponseChunk(ctx, e.cfg, b)		helps.LogWithRequestID(ctx).Debugf("request error, error status: %d, error message: %s", httpResp.StatusCode, helps.SummarizeErrorBody(httpResp.Header.Get("Content-Type"), b))		if errClose := httpResp.Body.Close(); errClose != nil {			log.Errorf("openai compat executor: close response body error: %v", errClose)		}		err = newOpenAICompatStatusError(httpResp.StatusCode, httpResp.Header, b)		return nil, err	}	out := make(chan cliproxyexecutor.StreamChunk)	go func() {		defer close(out)		defer func() {			if errClose := httpResp.Body.Close(); errClose != nil {				log.Errorf("openai compat executor: close response body error: %v", errClose)			}		}()		scanner := bufio.NewScanner(httpResp.Body)		scanner.Buffer(nil, 52_428_800) // 50MB		claudeInputTokens := helps.NewClaudeInputTokenState(from, to, responseFormat, originalPayload)		var param any		var streamUsage helps.StreamUsageBuffer		var seenDone bool		var streamFailed bool		var streamAborted bool		var upstreamEvent string		var frameData [][]byte		defer streamUsage.Publish(ctx, reporter) 		publishStreamError := func(streamErr statusErr, containsPayload bool) {			loggedErr := streamErr			if containsPayload {				loggedErr = statusErr{code: streamErr.code, msg: "upstream stream returned an error payload"}			}			helps.RecordAPIResponseError(ctx, e.cfg, loggedErr)			reporter.PublishFailure(ctx, loggedErr)			select {			case out <- cliproxyexecutor.StreamChunk{Err: streamErr}:			case <-ctx.Done():			}			streamFailed = true		} 		processFrame := func() bool {			eventName := upstreamEvent			upstreamEvent = ""			dataLines := frameData			frameData = nil			if len(dataLines) == 0 {				if openAICompatErrorEvent(eventName) {					publishStreamError(statusErr{code: http.StatusBadGateway, msg: "upstream error event ended without data"}, false)					return true				}				return false			} 			if len(dataLines) > 1 {				for _, dataLine := range dataLines {					if bytes.Equal(bytes.TrimSpace(dataLine), []byte("[DONE]")) {						publishStreamError(statusErr{code: http.StatusBadGateway, msg: "upstream stream ended with incomplete data before [DONE]"}, false)						return true					}				}			}			dataPayload := bytes.TrimSpace(bytes.Join(dataLines, []byte("\n")))			isDone := bytes.Equal(dataPayload, []byte("[DONE]"))			if isDone && openAICompatErrorEvent(eventName) {				publishStreamError(statusErr{code: http.StatusBadGateway, msg: "upstream error event ended before [DONE]"}, false)				return true			}			if !isDone && !json.Valid(dataPayload) {				publishStreamError(statusErr{code: http.StatusBadGateway, msg: "upstream stream ended with incomplete SSE data frame"}, false)				return true			}			if !isDone {				if streamErr, isError := openAICompatStreamDataError(dataPayload, eventName); isError {					publishStreamError(streamErr, true)					return true				}			} 			streamLine := append([]byte("data: "), dataPayload...)			chunks := helps.TranslateStreamWithClaudeInputTokens(ctx, to, responseFormat, req.Model, opts.OriginalRequest, translated, streamLine, &param, claudeInputTokens)			for i := range chunks {				select {				case out <- cliproxyexecutor.StreamChunk{Payload: chunks[i]}:				case <-ctx.Done():					streamAborted = true					return true				}			}			if isDone {				seenDone = true				return true			}			return false		}

Lines 1–500 of 1072. Load more