package executor import (	"bufio"	"bytes"	"context"	"encoding/json"	"fmt"	"io"	"mime"	"mime/multipart"	"net/http"	"net/textproto"	"strings"	"time" 	"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/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) {	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	originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, opts.Stream)	translated := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, opts.Stream) 	translated, err = thinking.ApplyThinking(translated, req.Model, 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 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)	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 = statusErr{code: httpResp.StatusCode, msg: string(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)	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)	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 = statusErr{code: httpResp.StatusCode, msg: string(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) {	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	originalTranslated := sdktranslator.TranslateRequest(from, to, baseModel, originalPayload, true)	translated := sdktranslator.TranslateRequest(from, to, baseModel, req.Payload, true) 	translated, err = thinking.ApplyThinking(translated, req.Model, 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) 	// Request usage data in the final streaming chunk so that token statistics	// are captured even when the upstream is an OpenAI-compatible provider.	translated, _ = sjson.SetBytes(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)	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 = statusErr{code: httpResp.StatusCode, msg: string(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		var param any		for scanner.Scan() {			line := scanner.Bytes()			helps.AppendAPIResponseChunk(ctx, e.cfg, line)			if detail, ok := helps.ParseOpenAIStreamUsage(line); ok {				reporter.Publish(ctx, detail)			}			trimmedLine := bytes.TrimSpace(line)			if len(trimmedLine) == 0 {				continue			} 			if !bytes.HasPrefix(trimmedLine, []byte("data:")) {				if bytes.HasPrefix(trimmedLine, []byte(":")) || bytes.HasPrefix(trimmedLine, []byte("event:")) ||					bytes.HasPrefix(trimmedLine, []byte("id:")) || bytes.HasPrefix(trimmedLine, []byte("retry:")) {					continue				}				if bytes.HasPrefix(trimmedLine, []byte("{")) || bytes.HasPrefix(trimmedLine, []byte("[")) {					streamErr := statusErr{code: http.StatusBadGateway, msg: string(trimmedLine)}					helps.RecordAPIResponseError(ctx, e.cfg, streamErr)					reporter.PublishFailure(ctx, streamErr)					select {					case out <- cliproxyexecutor.StreamChunk{Err: streamErr}:					case <-ctx.Done():					}					return				}				continue			} 			// OpenAI-compatible streams must use SSE data lines.			chunks := sdktranslator.TranslateStream(ctx, to, responseFormat, req.Model, opts.OriginalRequest, translated, bytes.Clone(trimmedLine), &param)			for i := range chunks {				select {				case out <- cliproxyexecutor.StreamChunk{Payload: chunks[i]}:				case <-ctx.Done():					return				}			}		}		if errScan := scanner.Err(); errScan != nil {			helps.RecordAPIResponseError(ctx, e.cfg, errScan)			reporter.PublishFailure(ctx, errScan)			select {			case out <- cliproxyexecutor.StreamChunk{Err: errScan}:			case <-ctx.Done():			}		} else {			// In case the upstream close the stream without a terminal [DONE] marker.			// Feed a synthetic done marker through the translator so pending			// response.completed events are still emitted exactly once.			chunks := sdktranslator.TranslateStream(ctx, to, responseFormat, req.Model, opts.OriginalRequest, translated, []byte("data: [DONE]"), &param)			for i := range chunks {				select {				case out <- cliproxyexecutor.StreamChunk{Payload: chunks[i]}:				case <-ctx.Done():					return				}			}		}		// Ensure we record the request if no usage chunk was ever seen		reporter.EnsurePublished(ctx)	}()	return &cliproxyexecutor.StreamResult{Headers: httpResp.Header.Clone(), Chunks: out}, nil} func (e *OpenAICompatExecutor) executeImagesStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, endpointPath string) (_ *cliproxyexecutor.StreamResult, 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 nil, err	} 	payload, contentType, errPrepare := prepareOpenAICompatImagesPayload(req.Payload, baseModel, opts.Headers.Get("Content-Type"), true)	if errPrepare != nil {		err = errPrepare		return nil, 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 nil, err	}	httpReq.Header.Set("Content-Type", contentType)	httpReq.Header.Set("Accept", "text/event-stream")	httpReq.Header.Set("Cache-Control", "no-cache")	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)	var authID, authLabel, authType, authValue string

Lines 1–500 of 797. Load more