feat(usage): introduce executor type tracking in usage reporting

- Replaced `NewUsageReporter` with `NewExecutorUsageReporter` to include executor type in usage records.
- Updated all executors to use the new reporter implementation.
- Extended `UsageReporter` to track and publish executor type.
- Added tests to validate proper executor type recording and handling.
- Enhanced RedisQueue plugin and payload schema with executor type support.
This commit is contained in:
Luis Pater
2026-06-02 00:43:16 +08:00
parent 05b972479a
commit 959067edfb
17 changed files with 110 additions and 54 deletions
+6
View File
@@ -42,6 +42,10 @@ func (p *usageQueuePlugin) HandleUsage(ctx context.Context, record coreusage.Rec
if provider == "" {
provider = "unknown"
}
executorType := strings.TrimSpace(record.ExecutorType)
if executorType == "" {
executorType = "unknown"
}
authType := strings.TrimSpace(record.AuthType)
if authType == "" {
authType = "unknown"
@@ -94,6 +98,7 @@ func (p *usageQueuePlugin) HandleUsage(ctx context.Context, record coreusage.Rec
payload, err := json.Marshal(queuedUsageDetail{
requestDetail: detail,
Provider: provider,
ExecutorType: executorType,
Model: modelName,
Alias: aliasName,
Endpoint: resolveEndpoint(ctx),
@@ -112,6 +117,7 @@ func (p *usageQueuePlugin) HandleUsage(ctx context.Context, record coreusage.Rec
type queuedUsageDetail struct {
requestDetail
Provider string `json:"provider"`
ExecutorType string `json:"executor_type"`
Model string `json:"model"`
Alias string `json:"alias"`
Endpoint string `json:"endpoint"`
+2
View File
@@ -26,6 +26,7 @@ func TestUsageQueuePluginPayloadIncludesStableFieldsAndSuccess(t *testing.T) {
plugin := &usageQueuePlugin{}
plugin.HandleUsage(ctx, coreusage.Record{
Provider: "openai",
ExecutorType: "KimiExecutor",
Model: "gpt-5.4",
Alias: "client-gpt",
APIKey: "test-key",
@@ -47,6 +48,7 @@ func TestUsageQueuePluginPayloadIncludesStableFieldsAndSuccess(t *testing.T) {
payload := popSinglePayload(t)
requireStringField(t, payload, "provider", "openai")
requireStringField(t, payload, "executor_type", "KimiExecutor")
requireStringField(t, payload, "model", "gpt-5.4")
requireStringField(t, payload, "alias", "client-gpt")
requireStringField(t, payload, "endpoint", "POST /v1/chat/completions")
@@ -128,7 +128,7 @@ func (e *AIStudioExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth,
return resp, statusErr{code: http.StatusNotImplemented, msg: "/responses/compact not supported"}
}
baseModel := thinking.ParseSuffix(req.Model).ModelName
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
translatedReq, body, err := e.translateRequest(req, opts, false)
@@ -196,7 +196,7 @@ func (e *AIStudioExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth
return nil, statusErr{code: http.StatusNotImplemented, msg: "/responses/compact not supported"}
}
baseModel := thinking.ParseSuffix(req.Model).ModelName
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
translatedReq, body, err := e.translateRequest(req, opts, true)
@@ -530,7 +530,7 @@ func (e *AntigravityExecutor) Execute(ctx context.Context, auth *cliproxyauth.Au
return e.executeClaudeNonStream(ctx, auth, req, opts)
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -730,7 +730,7 @@ func (e *AntigravityExecutor) executeClaudeNonStream(ctx context.Context, auth *
return resp, statusErr{code: http.StatusTooManyRequests, msg: fmt.Sprintf("auth in short cooldown, %s remaining", remaining), retryAfter: &d}
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -1192,7 +1192,7 @@ func (e *AntigravityExecutor) ExecuteStream(ctx context.Context, auth *cliproxya
return nil, statusErr{code: http.StatusTooManyRequests, msg: fmt.Sprintf("auth in short cooldown, %s remaining", remaining), retryAfter: &d}
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
+2 -2
View File
@@ -171,7 +171,7 @@ func (e *ClaudeExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, r
baseURL = "https://api.anthropic.com"
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
to := sdktranslator.FromString("claude")
@@ -354,7 +354,7 @@ func (e *ClaudeExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.A
baseURL = "https://api.anthropic.com"
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
to := sdktranslator.FromString("claude")
+3 -3
View File
@@ -264,7 +264,7 @@ func (e *CodexExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, re
baseURL = "https://chatgpt.com/backend-api/codex"
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -431,7 +431,7 @@ func (e *CodexExecutor) executeCompact(ctx context.Context, auth *cliproxyauth.A
baseURL = "https://chatgpt.com/backend-api/codex"
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -536,7 +536,7 @@ func (e *CodexExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Au
baseURL = "https://chatgpt.com/backend-api/codex"
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -89,7 +89,7 @@ func (e *CodexExecutor) executeOpenAIImage(ctx context.Context, auth *cliproxyau
}
mainModel := e.resolveGPTImage2BaseModel()
reporter := helps.NewUsageReporter(ctx, e.Identifier(), mainModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, mainModel, auth)
defer reporter.TrackFailure(ctx, &err)
body, errBuild := e.prepareCodexOpenAIImageBody(prepared.Body, req, opts, mainModel)
@@ -182,7 +182,7 @@ func (e *CodexExecutor) executeOpenAIImageStream(ctx context.Context, auth *clip
}
mainModel := e.resolveGPTImage2BaseModel()
reporter := helps.NewUsageReporter(ctx, e.Identifier(), mainModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, mainModel, auth)
defer reporter.TrackFailure(ctx, &err)
body, errBuild := e.prepareCodexOpenAIImageBody(prepared.Body, req, opts, mainModel)
@@ -184,7 +184,7 @@ func (e *CodexWebsocketsExecutor) Execute(ctx context.Context, auth *cliproxyaut
baseURL = "https://chatgpt.com/backend-api/codex"
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -404,7 +404,7 @@ func (e *CodexWebsocketsExecutor) ExecuteStream(ctx context.Context, auth *clipr
baseURL = "https://chatgpt.com/backend-api/codex"
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -118,7 +118,7 @@ func (e *GeminiCLIExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth
return resp, err
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -277,7 +277,7 @@ func (e *GeminiCLIExecutor) ExecuteStream(ctx context.Context, auth *cliproxyaut
return nil, err
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
+2 -2
View File
@@ -112,7 +112,7 @@ func (e *GeminiExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, r
apiKey, bearer := geminiCreds(auth)
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
// Official Gemini API via API key or OAuth bearer
@@ -224,7 +224,7 @@ func (e *GeminiExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.A
apiKey, bearer := geminiCreds(auth)
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -306,7 +306,7 @@ func (e *GeminiVertexExecutor) Refresh(ctx context.Context, auth *cliproxyauth.A
func (e *GeminiVertexExecutor) executeWithServiceAccount(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, projectID, location string, saJSON []byte) (resp cliproxyexecutor.Response, err error) {
baseModel := thinking.ParseSuffix(req.Model).ModelName
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
var body []byte
@@ -441,7 +441,7 @@ func (e *GeminiVertexExecutor) executeWithServiceAccount(ctx context.Context, au
func (e *GeminiVertexExecutor) executeWithAPIKey(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, apiKey, baseURL string) (resp cliproxyexecutor.Response, err error) {
baseModel := thinking.ParseSuffix(req.Model).ModelName
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -555,7 +555,7 @@ func (e *GeminiVertexExecutor) executeWithAPIKey(ctx context.Context, auth *clip
func (e *GeminiVertexExecutor) executeStreamWithServiceAccount(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, projectID, location string, saJSON []byte) (_ *cliproxyexecutor.StreamResult, err error) {
baseModel := thinking.ParseSuffix(req.Model).ModelName
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -699,7 +699,7 @@ func (e *GeminiVertexExecutor) executeStreamWithServiceAccount(ctx context.Conte
func (e *GeminiVertexExecutor) executeStreamWithAPIKey(ctx context.Context, auth *cliproxyauth.Auth, req cliproxyexecutor.Request, opts cliproxyexecutor.Options, apiKey, baseURL string) (_ *cliproxyexecutor.StreamResult, err error) {
baseModel := thinking.ParseSuffix(req.Model).ModelName
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
from := opts.SourceFormat
@@ -7,6 +7,7 @@ import (
"fmt"
"io"
"net/http"
"reflect"
"strings"
"sync"
"time"
@@ -21,22 +22,37 @@ import (
)
type UsageReporter struct {
provider string
model string
alias string
authID string
authIndex string
authType string
apiKey string
source string
reasoning string
serviceTier string
requestedAt time.Time
ttftMu sync.RWMutex
ttft time.Duration
ttftStart time.Time
ttftSet bool
once sync.Once
provider string
executorType string
model string
alias string
authID string
authIndex string
authType string
apiKey string
source string
reasoning string
serviceTier string
requestedAt time.Time
ttftMu sync.RWMutex
ttft time.Duration
ttftStart time.Time
ttftSet bool
once sync.Once
}
type usageExecutor interface {
Identifier() string
}
func NewExecutorUsageReporter(ctx context.Context, executor usageExecutor, model string, auth *cliproxyauth.Auth) *UsageReporter {
provider := ""
if executor != nil {
provider = executor.Identifier()
}
reporter := NewUsageReporter(ctx, provider, model, auth)
reporter.executorType = ExecutorTypeName(executor)
return reporter
}
func NewUsageReporter(ctx context.Context, provider, model string, auth *cliproxyauth.Auth) *UsageReporter {
@@ -63,6 +79,17 @@ func NewUsageReporter(ctx context.Context, provider, model string, auth *cliprox
return reporter
}
func ExecutorTypeName(executor any) string {
if executor == nil {
return ""
}
executorType := reflect.TypeOf(executor)
for executorType.Kind() == reflect.Pointer {
executorType = executorType.Elem()
}
return strings.TrimSpace(executorType.Name())
}
func (r *UsageReporter) Publish(ctx context.Context, detail usage.Detail) {
r.publishWithOutcome(ctx, detail, false, usage.Failure{})
}
@@ -234,6 +261,7 @@ func (r *UsageReporter) buildRecordForModel(model string, detail usage.Detail, f
}
return usage.Record{
Provider: r.provider,
ExecutorType: r.executorType,
Model: model,
Alias: r.alias,
Source: r.source,
@@ -231,6 +231,18 @@ func TestUsageReporterBuildRecordIncludesRequestedModelAlias(t *testing.T) {
}
}
func TestNewExecutorUsageReporterIncludesExecutorType(t *testing.T) {
reporter := NewExecutorUsageReporter(context.Background(), &TestUsageExecutor{}, "gpt-5.4", nil)
record := reporter.buildRecord(usage.Detail{TotalTokens: 3}, false)
if record.Provider != "test-provider" {
t.Fatalf("provider = %q, want %q", record.Provider, "test-provider")
}
if record.ExecutorType != "TestUsageExecutor" {
t.Fatalf("executor type = %q, want %q", record.ExecutorType, "TestUsageExecutor")
}
}
func TestUsageReporterBuildRecordIncludesReasoningEffort(t *testing.T) {
ctx := usage.WithReasoningEffort(context.Background(), "medium")
reporter := NewUsageReporter(ctx, "openai", "gpt-5.4", nil)
@@ -297,3 +309,9 @@ type roundTripFunc func(*http.Request) (*http.Response, error)
func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
return f(req)
}
type TestUsageExecutor struct{}
func (TestUsageExecutor) Identifier() string {
return "test-provider"
}
+2 -2
View File
@@ -83,7 +83,7 @@ func (e *KimiExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req
token := kimiCreds(auth)
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
to := sdktranslator.FromString("openai")
@@ -191,7 +191,7 @@ func (e *KimiExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Aut
baseModel := thinking.ParseSuffix(req.Model).ModelName
token := kimiCreds(auth)
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
to := sdktranslator.FromString("openai")
@@ -89,7 +89,7 @@ func (e *OpenAICompatExecutor) Execute(ctx context.Context, auth *cliproxyauth.A
baseModel := thinking.ParseSuffix(req.Model).ModelName
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
baseURL, apiKey := e.resolveCredentials(auth)
@@ -201,7 +201,7 @@ func (e *OpenAICompatExecutor) Execute(ctx context.Context, auth *cliproxyauth.A
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.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
baseURL, apiKey := e.resolveCredentials(auth)
@@ -294,7 +294,7 @@ func (e *OpenAICompatExecutor) ExecuteStream(ctx context.Context, auth *cliproxy
baseModel := thinking.ParseSuffix(req.Model).ModelName
reporter := helps.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
baseURL, apiKey := e.resolveCredentials(auth)
@@ -459,7 +459,7 @@ func (e *OpenAICompatExecutor) ExecuteStream(ctx context.Context, auth *cliproxy
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.NewUsageReporter(ctx, e.Identifier(), baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
baseURL, apiKey := e.resolveCredentials(auth)
+2 -2
View File
@@ -114,7 +114,7 @@ func (e *XAIExecutor) Execute(ctx context.Context, auth *cliproxyauth.Auth, req
return resp, err
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), prepared.baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, prepared.baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
reporter.SetTranslatedReasoningEffort(prepared.body, e.Identifier())
@@ -302,7 +302,7 @@ func (e *XAIExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Auth
return nil, err
}
reporter := helps.NewUsageReporter(ctx, e.Identifier(), prepared.baseModel, auth)
reporter := helps.NewExecutorUsageReporter(ctx, e, prepared.baseModel, auth)
defer reporter.TrackFailure(ctx, &err)
reporter.SetTranslatedReasoningEffort(prepared.body, e.Identifier())
+10 -8
View File
@@ -15,14 +15,16 @@ const DefaultServiceTier = "default"
// Record contains the usage statistics captured for a single provider request.
type Record struct {
Provider string
Model string
Alias string
APIKey string
AuthID string
AuthIndex string
AuthType string
Source string
Provider string
// ExecutorType stores the concrete executor type that handled the request.
ExecutorType string
Model string
Alias string
APIKey string
AuthID string
AuthIndex string
AuthType string
Source string
// ReasoningEffort stores the translated upstream thinking level for request event logs.
ReasoningEffort string
// ServiceTier stores the client-requested service tier for request event logs.