package executor import ( "bufio" "bytes" "context" "fmt" "io" "net/http" "os" "path/filepath" "runtime" "strings" "time" kimiauth "github.com/router-for-me/CLIProxyAPI/v7/internal/auth/kimi" "github.com/router-for-me/CLIProxyAPI/v7/internal/buildinfo" "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 kimiReasoningUnavailable = "[reasoning unavailable]" // KimiExecutor is a stateless executor for Kimi API using OpenAI-compatible chat completions.type KimiExecutor struct { ClaudeExecutor cfg *config.Config} // NewKimiExecutor creates a new Kimi executor.func NewKimiExecutor(cfg *config.Config) *KimiExecutor { return &KimiExecutor{ ClaudeExecutor: ClaudeExecutor{ cfg: cfg, requestLogProvider: "kimi", upstreamModelNormalizer: normalizeKimiUpstreamModel, }, cfg: cfg, }} // Identifier returns the executor identifier.func (e *KimiExecutor) Identifier() string { return "kimi" } // RequestToFormat reports the upstream request format used after auth selection.func (e *KimiExecutor) RequestToFormat(_ cliproxyexecutor.Request, opts cliproxyexecutor.Options) sdktranslator.Format { if opts.SourceFormat == sdktranslator.FormatClaude { return sdktranslator.FormatClaude } if opts.SourceFormat == sdktranslator.FormatOpenAIResponse { return sdktranslator.FormatOpenAIResponse } return sdktranslator.FormatOpenAI} // PrepareRequest injects Kimi credentials into the outgoing HTTP request.func (e *KimiExecutor) PrepareRequest(req *http.Request, auth *cliproxyauth.Auth) error { if req == nil { return nil } token := kimiCreds(auth) if strings.TrimSpace(token) != "" { req.Header.Set("Authorization", "Bearer "+token) } var attrs map[string]string if auth != nil { attrs = auth.Attributes } util.ApplyCustomHeadersFromAttrs(req, attrs) return nil} // HttpRequest injects Kimi credentials into the request and executes it.func (e *KimiExecutor) HttpRequest(ctx context.Context, auth *cliproxyauth.Auth, req *http.Request) (*http.Response, error) { if req == nil { return nil, fmt.Errorf("kimi 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)} // Execute performs a non-streaming chat completion request to Kimi.func (e *KimiExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { from := opts.SourceFormat if from.String() == "claude" { auth.Attributes["base_url"] = kimiauth.KimiAPIBaseURL preparedReq, replayScope := prepareKimiThinkingReplayRequest(ctx, req, opts) claudeResp, errExecute := e.ClaudeExecutor.Execute(ctx, auth, preparedReq, opts) if errExecute != nil { if replayScope.replayApplied && shouldClearKimiThinkingReplayAfterError(errExecute) { clearKimiThinkingReplayContent(ctx, replayScope) } return claudeResp, errExecute } cacheKimiThinkingReplayResponse(ctx, replayScope, claudeResp.Payload) return claudeResp, nil } if from == sdktranslator.FormatOpenAIResponse { return e.executeResponses(ctx, auth, req, opts) } responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts) baseModel := thinking.ParseSuffix(req.Model).ModelName token := kimiCreds(auth) reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) defer reporter.TrackFailure(ctx, &err) to := sdktranslator.FromString("openai") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := bytes.Clone(originalPayloadSource) originalTranslated := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, originalPayload, false) body := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, bytes.Clone(req.Payload), false) // Strip kimi- prefix and any [1m] suffix for upstream API upstreamModel := normalizeKimiUpstreamModel(baseModel) body, err = sjson.SetBytes(body, "model", upstreamModel) if err != nil { return resp, fmt.Errorf("kimi executor: failed to set model in payload: %w", err) } body, err = helps.ApplyRequestThinking(body, req, opts, from.String(), "kimi", e.Identifier()) if err != nil { return resp, err } requestedModel := helps.PayloadRequestedModel(opts, req.Model) requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", body, originalTranslated, requestedModel, requestPath, opts.Headers) body, err = normalizeKimiToolMessageLinks(body) if err != nil { return resp, err } body = normalizeKimiTools(body) body = normalizeKimiTemperature(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) url := kimiauth.KimiAPIBaseURL + "/v1/chat/completions" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) if err != nil { return resp, err } applyKimiHeadersWithAuth(httpReq, token, false, auth) 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: body, 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("kimi 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 } data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data) reporter.Publish(ctx, helps.ParseOpenAIUsage(data)) var param any // Note: TranslateNonStream uses req.Model (original with suffix) to preserve // the original model name in the response for client compatibility. out := sdktranslator.TranslateNonStream(ctx, to, responseFormat, req.Model, opts.OriginalRequest, body, data, ¶m) if responseFormat == sdktranslator.FormatOpenAIResponse { out = helps.EnsureResponsesUsageDetails(out) } resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil} // ExecuteStream performs a streaming chat completion request to Kimi.func (e *KimiExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (_ *cliproxyexecutor.StreamResult, err error) { from := opts.SourceFormat if from.String() == "claude" { auth.Attributes["base_url"] = kimiauth.KimiAPIBaseURL preparedReq, replayScope := prepareKimiThinkingReplayRequest(ctx, req, opts) claudeResult, errExecute := e.ClaudeExecutor.ExecuteStream(ctx, auth, preparedReq, opts) if errExecute != nil { if replayScope.replayApplied && shouldClearKimiThinkingReplayAfterError(errExecute) { clearKimiThinkingReplayContent(ctx, replayScope) } return nil, errExecute } return wrapKimiThinkingReplayStream(ctx, claudeResult, replayScope), nil } if from == sdktranslator.FormatOpenAIResponse { return e.executeResponsesStream(ctx, auth, req, opts) } responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts) baseModel := thinking.ParseSuffix(req.Model).ModelName token := kimiCreds(auth) reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) defer reporter.TrackFailure(ctx, &err) to := sdktranslator.FromString("openai") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := bytes.Clone(originalPayloadSource) originalTranslated := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, originalPayload, true) body := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, bytes.Clone(req.Payload), true) // Strip kimi- prefix and any [1m] suffix for upstream API upstreamModel := normalizeKimiUpstreamModel(baseModel) body, err = sjson.SetBytes(body, "model", upstreamModel) if err != nil { return nil, fmt.Errorf("kimi executor: failed to set model in payload: %w", err) } body, err = helps.ApplyRequestThinking(body, req, opts, from.String(), "kimi", e.Identifier()) if err != nil { return nil, err } body, err = sjson.SetBytes(body, "stream_options.include_usage", true) if err != nil { return nil, fmt.Errorf("kimi executor: failed to set stream_options in payload: %w", err) } requestedModel := helps.PayloadRequestedModel(opts, req.Model) requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", body, originalTranslated, requestedModel, requestPath, opts.Headers) body, err = normalizeKimiToolMessageLinks(body) if err != nil { return nil, err } body = normalizeKimiTools(body) body = normalizeKimiTemperature(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) url := kimiauth.KimiAPIBaseURL + "/v1/chat/completions" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) if err != nil { return nil, err } applyKimiHeadersWithAuth(httpReq, token, true, auth) 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: body, 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("kimi 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("kimi executor: close response body error: %v", errClose) } }() scanner := bufio.NewScanner(httpResp.Body) scanner.Buffer(nil, 1_048_576) // 1MB claudeInputTokens := helps.NewClaudeInputTokenState(from, to, responseFormat, originalPayload) var param any var streamUsage helps.StreamUsageBuffer defer streamUsage.Publish(ctx, reporter) for scanner.Scan() { line := scanner.Bytes() helps.AppendAPIResponseChunk(ctx, e.cfg, line) streamUsage.ObserveOpenAIStream(line) chunks := helps.TranslateStreamWithClaudeInputTokens(ctx, to, responseFormat, req.Model, opts.OriginalRequest, body, bytes.Clone(line), ¶m, claudeInputTokens) for i := range chunks { select { case out <- cliproxyexecutor.StreamChunk{Payload: chunks[i]}: case <-ctx.Done(): return } } } doneChunks := helps.TranslateStreamWithClaudeInputTokens(ctx, to, responseFormat, req.Model, opts.OriginalRequest, body, []byte("[DONE]"), ¶m, claudeInputTokens) for i := range doneChunks { select { case out <- cliproxyexecutor.StreamChunk{Payload: doneChunks[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(): } } }() return &cliproxyexecutor.StreamResult{Headers: httpResp.Header.Clone(), Chunks: out}, nil} func (e *KimiExecutor) executeResponses(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { if opts.Alt == "responses/compact" { return resp, statusErr{code: http.StatusNotImplemented, msg: "/responses/compact not supported"} } responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts) baseModel := thinking.ParseSuffix(req.Model).ModelName token := kimiCreds(auth) reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) defer reporter.TrackFailure(ctx, &err) body := bytes.Clone(req.Payload) upstreamModel := normalizeKimiUpstreamModel(baseModel) var errSet error body, errSet = sjson.SetBytes(body, "model", upstreamModel) if errSet != nil { return resp, fmt.Errorf("kimi executor: failed to set model in payload: %w", errSet) } body = helps.SetBoolIfDifferent(body, "stream", false) var errThinking error body, errThinking = helps.ApplyRequestThinking(body, req, opts, opts.SourceFormat.String(), sdktranslator.FormatCodex.String(), e.Identifier()) if errThinking != nil { return resp, errThinking } requestedModel := helps.PayloadRequestedModel(opts, req.Model) requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, "openai-response", opts.SourceFormat.String(), "", body, req.Payload, requestedModel, requestPath, opts.Headers) body = normalizeKimiTools(body) body = normalizeKimiTemperature(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) url := helps.ResolveKimiResponsesURL(auth) httpReq, errNewRequest := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) if errNewRequest != nil { return resp, errNewRequest } applyKimiHeadersWithAuth(httpReq, token, false, auth) 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: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpClient = reporter.TrackHTTPClient(httpClient) httpResp, errDo := httpClient.Do(httpReq) if errDo != nil { helps.RecordAPIResponseError(ctx, e.cfg, errDo) return resp, errDo } defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("kimi 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 } data, errRead := io.ReadAll(httpResp.Body) if errRead != nil { helps.RecordAPIResponseError(ctx, e.cfg, errRead) return resp, errRead } helps.AppendAPIResponseChunk(ctx, e.cfg, data) if usage, ok := helps.ParseCodexUsage(data); ok && (usage.TotalTokens > 0 || usage.InputTokens > 0) { reporter.Publish(ctx, usage) } else if usage := helps.ParseOpenAIUsage(data); usage.TotalTokens > 0 || usage.InputTokens > 0 { reporter.Publish(ctx, usage) } out := data if responseFormat != sdktranslator.FormatOpenAIResponse { var param any out = sdktranslator.TranslateNonStream(ctx, sdktranslator.FormatOpenAIResponse, responseFormat, req.Model, opts.OriginalRequest, body, data, ¶m) } resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil} func (e *KimiExecutor) executeResponsesStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (_ *cliproxyexecutor.StreamResult, err error) { if opts.Alt == "responses/compact" { return nil, statusErr{code: http.StatusBadRequest, msg: "streaming not supported for /responses/compact"} } responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts) baseModel := thinking.ParseSuffix(req.Model).ModelName token := kimiCreds(auth) reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) defer reporter.TrackFailure(ctx, &err) Lines 1–500 of 1231. Load more
package executor import ( "bufio" "bytes" "context" "fmt" "io" "net/http" "os" "path/filepath" "runtime" "strings" "time" kimiauth "github.com/router-for-me/CLIProxyAPI/v7/internal/auth/kimi" "github.com/router-for-me/CLIProxyAPI/v7/internal/buildinfo" "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 kimiReasoningUnavailable = "[reasoning unavailable]" // KimiExecutor is a stateless executor for Kimi API using OpenAI-compatible chat completions. type KimiExecutor struct { ClaudeExecutor cfg *config.Config } // NewKimiExecutor creates a new Kimi executor. func NewKimiExecutor(cfg *config.Config) *KimiExecutor { return &KimiExecutor{ ClaudeExecutor: ClaudeExecutor{ cfg: cfg, requestLogProvider: "kimi", upstreamModelNormalizer: normalizeKimiUpstreamModel, }, cfg: cfg, } } // Identifier returns the executor identifier. func (e *KimiExecutor) Identifier() string { return "kimi" } // RequestToFormat reports the upstream request format used after auth selection. func (e *KimiExecutor) RequestToFormat(_ cliproxyexecutor.Request, opts cliproxyexecutor.Options) sdktranslator.Format { if opts.SourceFormat == sdktranslator.FormatClaude { return sdktranslator.FormatClaude } if opts.SourceFormat == sdktranslator.FormatOpenAIResponse { return sdktranslator.FormatOpenAIResponse } return sdktranslator.FormatOpenAI } // PrepareRequest injects Kimi credentials into the outgoing HTTP request. func (e *KimiExecutor) PrepareRequest(req *http.Request, auth *cliproxyauth.Auth) error { if req == nil { return nil } token := kimiCreds(auth) if strings.TrimSpace(token) != "" { req.Header.Set("Authorization", "Bearer "+token) } var attrs map[string]string if auth != nil { attrs = auth.Attributes } util.ApplyCustomHeadersFromAttrs(req, attrs) return nil } // HttpRequest injects Kimi credentials into the request and executes it. func (e *KimiExecutor) HttpRequest(ctx context.Context, auth *cliproxyauth.Auth, req *http.Request) (*http.Response, error) { if req == nil { return nil, fmt.Errorf("kimi 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) } // Execute performs a non-streaming chat completion request to Kimi. func (e *KimiExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { from := opts.SourceFormat if from.String() == "claude" { auth.Attributes["base_url"] = kimiauth.KimiAPIBaseURL preparedReq, replayScope := prepareKimiThinkingReplayRequest(ctx, req, opts) claudeResp, errExecute := e.ClaudeExecutor.Execute(ctx, auth, preparedReq, opts) if errExecute != nil { if replayScope.replayApplied && shouldClearKimiThinkingReplayAfterError(errExecute) { clearKimiThinkingReplayContent(ctx, replayScope) } return claudeResp, errExecute } cacheKimiThinkingReplayResponse(ctx, replayScope, claudeResp.Payload) return claudeResp, nil } if from == sdktranslator.FormatOpenAIResponse { return e.executeResponses(ctx, auth, req, opts) } responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts) baseModel := thinking.ParseSuffix(req.Model).ModelName token := kimiCreds(auth) reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) defer reporter.TrackFailure(ctx, &err) to := sdktranslator.FromString("openai") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := bytes.Clone(originalPayloadSource) originalTranslated := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, originalPayload, false) body := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, bytes.Clone(req.Payload), false) // Strip kimi- prefix and any [1m] suffix for upstream API upstreamModel := normalizeKimiUpstreamModel(baseModel) body, err = sjson.SetBytes(body, "model", upstreamModel) if err != nil { return resp, fmt.Errorf("kimi executor: failed to set model in payload: %w", err) } body, err = helps.ApplyRequestThinking(body, req, opts, from.String(), "kimi", e.Identifier()) if err != nil { return resp, err } requestedModel := helps.PayloadRequestedModel(opts, req.Model) requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", body, originalTranslated, requestedModel, requestPath, opts.Headers) body, err = normalizeKimiToolMessageLinks(body) if err != nil { return resp, err } body = normalizeKimiTools(body) body = normalizeKimiTemperature(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) url := kimiauth.KimiAPIBaseURL + "/v1/chat/completions" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) if err != nil { return resp, err } applyKimiHeadersWithAuth(httpReq, token, false, auth) 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: body, 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("kimi 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 } data, err := io.ReadAll(httpResp.Body) if err != nil { helps.RecordAPIResponseError(ctx, e.cfg, err) return resp, err } helps.AppendAPIResponseChunk(ctx, e.cfg, data) reporter.Publish(ctx, helps.ParseOpenAIUsage(data)) var param any // Note: TranslateNonStream uses req.Model (original with suffix) to preserve // the original model name in the response for client compatibility. out := sdktranslator.TranslateNonStream(ctx, to, responseFormat, req.Model, opts.OriginalRequest, body, data, ¶m) if responseFormat == sdktranslator.FormatOpenAIResponse { out = helps.EnsureResponsesUsageDetails(out) } resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil } // ExecuteStream performs a streaming chat completion request to Kimi. func (e *KimiExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (_ *cliproxyexecutor.StreamResult, err error) { from := opts.SourceFormat if from.String() == "claude" { auth.Attributes["base_url"] = kimiauth.KimiAPIBaseURL preparedReq, replayScope := prepareKimiThinkingReplayRequest(ctx, req, opts) claudeResult, errExecute := e.ClaudeExecutor.ExecuteStream(ctx, auth, preparedReq, opts) if errExecute != nil { if replayScope.replayApplied && shouldClearKimiThinkingReplayAfterError(errExecute) { clearKimiThinkingReplayContent(ctx, replayScope) } return nil, errExecute } return wrapKimiThinkingReplayStream(ctx, claudeResult, replayScope), nil } if from == sdktranslator.FormatOpenAIResponse { return e.executeResponsesStream(ctx, auth, req, opts) } responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts) baseModel := thinking.ParseSuffix(req.Model).ModelName token := kimiCreds(auth) reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) defer reporter.TrackFailure(ctx, &err) to := sdktranslator.FromString("openai") originalPayloadSource := req.Payload if len(opts.OriginalRequest) > 0 { originalPayloadSource = opts.OriginalRequest } originalPayload := bytes.Clone(originalPayloadSource) originalTranslated := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, originalPayload, true) body := helps.TranslateRequestWithCodexMultiAgentV2(ctx, opts.Headers, e.cfg, from, to, baseModel, bytes.Clone(req.Payload), true) // Strip kimi- prefix and any [1m] suffix for upstream API upstreamModel := normalizeKimiUpstreamModel(baseModel) body, err = sjson.SetBytes(body, "model", upstreamModel) if err != nil { return nil, fmt.Errorf("kimi executor: failed to set model in payload: %w", err) } body, err = helps.ApplyRequestThinking(body, req, opts, from.String(), "kimi", e.Identifier()) if err != nil { return nil, err } body, err = sjson.SetBytes(body, "stream_options.include_usage", true) if err != nil { return nil, fmt.Errorf("kimi executor: failed to set stream_options in payload: %w", err) } requestedModel := helps.PayloadRequestedModel(opts, req.Model) requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, to.String(), from.String(), "", body, originalTranslated, requestedModel, requestPath, opts.Headers) body, err = normalizeKimiToolMessageLinks(body) if err != nil { return nil, err } body = normalizeKimiTools(body) body = normalizeKimiTemperature(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) url := kimiauth.KimiAPIBaseURL + "/v1/chat/completions" httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) if err != nil { return nil, err } applyKimiHeadersWithAuth(httpReq, token, true, auth) 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: body, 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("kimi 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("kimi executor: close response body error: %v", errClose) } }() scanner := bufio.NewScanner(httpResp.Body) scanner.Buffer(nil, 1_048_576) // 1MB claudeInputTokens := helps.NewClaudeInputTokenState(from, to, responseFormat, originalPayload) var param any var streamUsage helps.StreamUsageBuffer defer streamUsage.Publish(ctx, reporter) for scanner.Scan() { line := scanner.Bytes() helps.AppendAPIResponseChunk(ctx, e.cfg, line) streamUsage.ObserveOpenAIStream(line) chunks := helps.TranslateStreamWithClaudeInputTokens(ctx, to, responseFormat, req.Model, opts.OriginalRequest, body, bytes.Clone(line), ¶m, claudeInputTokens) for i := range chunks { select { case out <- cliproxyexecutor.StreamChunk{Payload: chunks[i]}: case <-ctx.Done(): return } } } doneChunks := helps.TranslateStreamWithClaudeInputTokens(ctx, to, responseFormat, req.Model, opts.OriginalRequest, body, []byte("[DONE]"), ¶m, claudeInputTokens) for i := range doneChunks { select { case out <- cliproxyexecutor.StreamChunk{Payload: doneChunks[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(): } } }() return &cliproxyexecutor.StreamResult{Headers: httpResp.Header.Clone(), Chunks: out}, nil } func (e *KimiExecutor) executeResponses(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (resp cliproxyexecutor.Response, err error) { if opts.Alt == "responses/compact" { return resp, statusErr{code: http.StatusNotImplemented, msg: "/responses/compact not supported"} } responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts) baseModel := thinking.ParseSuffix(req.Model).ModelName token := kimiCreds(auth) reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) defer reporter.TrackFailure(ctx, &err) body := bytes.Clone(req.Payload) upstreamModel := normalizeKimiUpstreamModel(baseModel) var errSet error body, errSet = sjson.SetBytes(body, "model", upstreamModel) if errSet != nil { return resp, fmt.Errorf("kimi executor: failed to set model in payload: %w", errSet) } body = helps.SetBoolIfDifferent(body, "stream", false) var errThinking error body, errThinking = helps.ApplyRequestThinking(body, req, opts, opts.SourceFormat.String(), sdktranslator.FormatCodex.String(), e.Identifier()) if errThinking != nil { return resp, errThinking } requestedModel := helps.PayloadRequestedModel(opts, req.Model) requestPath := helps.PayloadRequestPath(opts) body = helps.ApplyPayloadConfigWithRequest(e.cfg, baseModel, "openai-response", opts.SourceFormat.String(), "", body, req.Payload, requestedModel, requestPath, opts.Headers) body = normalizeKimiTools(body) body = normalizeKimiTemperature(body) reporter.SetTranslatedReasoningEffort(body, e.Identifier()) url := helps.ResolveKimiResponsesURL(auth) httpReq, errNewRequest := http.NewRequestWithContext(ctx, http.MethodPost, url, bytes.NewReader(body)) if errNewRequest != nil { return resp, errNewRequest } applyKimiHeadersWithAuth(httpReq, token, false, auth) 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: body, Provider: e.Identifier(), AuthID: authID, AuthLabel: authLabel, AuthType: authType, AuthValue: authValue, }) httpClient := helps.NewProxyAwareHTTPClient(ctx, e.cfg, auth, 0) httpClient = reporter.TrackHTTPClient(httpClient) httpResp, errDo := httpClient.Do(httpReq) if errDo != nil { helps.RecordAPIResponseError(ctx, e.cfg, errDo) return resp, errDo } defer func() { if errClose := httpResp.Body.Close(); errClose != nil { log.Errorf("kimi 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 } data, errRead := io.ReadAll(httpResp.Body) if errRead != nil { helps.RecordAPIResponseError(ctx, e.cfg, errRead) return resp, errRead } helps.AppendAPIResponseChunk(ctx, e.cfg, data) if usage, ok := helps.ParseCodexUsage(data); ok && (usage.TotalTokens > 0 || usage.InputTokens > 0) { reporter.Publish(ctx, usage) } else if usage := helps.ParseOpenAIUsage(data); usage.TotalTokens > 0 || usage.InputTokens > 0 { reporter.Publish(ctx, usage) } out := data if responseFormat != sdktranslator.FormatOpenAIResponse { var param any out = sdktranslator.TranslateNonStream(ctx, sdktranslator.FormatOpenAIResponse, responseFormat, req.Model, opts.OriginalRequest, body, data, ¶m) } resp = cliproxyexecutor.Response{Payload: out, Headers: httpResp.Header.Clone()} return resp, nil } func (e *KimiExecutor) executeResponsesStream(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options) (_ *cliproxyexecutor.StreamResult, err error) { if opts.Alt == "responses/compact" { return nil, statusErr{code: http.StatusBadRequest, msg: "streaming not supported for /responses/compact"} } responseFormat := cliproxyexecutor.ResponseFormatOrSource(opts) baseModel := thinking.ParseSuffix(req.Model).ModelName token := kimiCreds(auth) reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth) defer reporter.TrackFailure(ctx, &err)