mirror of
https://github.com/router-for-me/CLIProxyAPI.git
synced 2026-10-26 09:34:17 +08:00
fix(meta): measure streaming TTFT on effective token frames
Meta streaming armed protocol-aware TTFT tracking but never observed token events, so usage records kept the first-body-byte fallback even when substantive output arrived later. Observe Responses events before failure publication and translation using the existing shared helper. Add deterministic virtual-clock coverage for content, terminal events, error fallbacks, downstream translation, and non-streaming timing. Fixes #6664
This commit is contained in:
Dylan
@@ -198,7 +198,8 @@ func (e *MetaExecutor) ExecuteStream(ctx context.Context, auth *cliproxyauth.Aut
|
||||
filteringSubscriptionUsage = false
|
||||
}
|
||||
skipFilteredBlankLine = false
|
||||
reporter.ObserveResponseModel(eventData)
|
||||
// Observe upstream content before translation or failure publication.
|
||||
helps.ObserveResponsesTokenEvent(reporter, eventData)
|
||||
if errEvent := metaStreamEventError(eventData); errEvent != nil {
|
||||
helps.RecordAPIResponseError(ctx, e.cfg, errEvent)
|
||||
reporter.PublishFailure(ctx, errEvent)
|
||||
|
||||
204
internal/runtime/executor/meta_executor_ttft_test.go
Normal file
204
internal/runtime/executor/meta_executor_ttft_test.go
Normal file
@@ -0,0 +1,204 @@
|
||||
package executor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net/http"
|
||||
"strings"
|
||||
"testing"
|
||||
"testing/synctest"
|
||||
"time"
|
||||
|
||||
"github.com/router-for-me/CLIProxyAPI/v8/internal/config"
|
||||
cliproxyauth "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/auth"
|
||||
cliproxyexecutor "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/executor"
|
||||
coreusage "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/usage"
|
||||
sdktranslator "github.com/router-for-me/CLIProxyAPI/v8/sdk/translator"
|
||||
)
|
||||
|
||||
type metaTTFTRoundTripFunc func(*http.Request) (*http.Response, error)
|
||||
|
||||
func (f metaTTFTRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
|
||||
return f(req)
|
||||
}
|
||||
|
||||
func TestMetaExecutorTTFT(t *testing.T) {
|
||||
const metadata = ": keepalive\n\n" +
|
||||
"event: response.subscription_usage\n" +
|
||||
"data: {\"type\":\"response.subscription_usage\",\"model\":\"subscription-telemetry\",\"subscription\":{\"tier\":\"test\"}}\n\n" +
|
||||
"data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_ttft\",\"model\":\"muse-spark-1.3\"}}\n\n" +
|
||||
"data: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"reasoning\",\"encrypted_content\":\"opaque\",\"summary\":[]}}\n\n" +
|
||||
"data: {\"type\":\"response.output_text.delta\",\"delta\":\"\"}\n\n"
|
||||
const textDelta = "data: {\"type\":\"response.output_text.delta\",\"output_index\":0,\"content_index\":0,\"delta\":\"hello\"}\n\n"
|
||||
const completed = "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_ttft\",\"object\":\"response\",\"status\":\"completed\",\"model\":\"muse-spark-1.3\",\"output\":[{\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"hello\"}]}],\"usage\":{\"input_tokens\":10,\"output_tokens\":3,\"total_tokens\":13}}}\n\n"
|
||||
const emptyCompleted = "data: {\"type\":\"response.completed\",\"response\":{\"id\":\"resp_ttft\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":0,\"output_tokens\":0,\"total_tokens\":0}}}\n\n"
|
||||
const emptyIncomplete = "data: {\"type\":\"response.incomplete\",\"response\":{\"id\":\"resp_ttft\",\"status\":\"incomplete\",\"output\":[],\"usage\":{\"input_tokens\":0,\"output_tokens\":0,\"total_tokens\":0}}}\n\n"
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
event string
|
||||
content string
|
||||
format sdktranslator.Format
|
||||
terminal bool
|
||||
nonstream bool
|
||||
httpStatus int
|
||||
readError error
|
||||
wantFailed bool
|
||||
}{
|
||||
{name: "text", event: textDelta, content: "hello"},
|
||||
{name: "reasoning_summary", event: "data: {\"type\":\"response.reasoning_summary_text.delta\",\"output_index\":0,\"summary_index\":0,\"delta\":\"thinking\"}\n\n", content: "thinking"},
|
||||
{name: "function_arguments", event: "data: {\"type\":\"response.function_call_arguments.delta\",\"output_index\":0,\"delta\":\"{\\\"query\\\":\\\"hello\\\"}\"}\n\n", content: "query"},
|
||||
{name: "chat_downstream", event: textDelta, content: "hello", format: sdktranslator.FormatOpenAI},
|
||||
{name: "completed_without_delta", event: completed, content: "hello", terminal: true},
|
||||
{name: "empty_completed", event: emptyCompleted, terminal: true},
|
||||
{name: "empty_incomplete", event: emptyIncomplete, terminal: true},
|
||||
{name: "failed_event", event: "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"message\":\"upstream failed\"}}}\n\n", terminal: true, wantFailed: true},
|
||||
{name: "error_event", event: "data: {\"type\":\"error\",\"error\":{\"code\":502,\"message\":\"upstream failed\"}}\n\n", terminal: true, wantFailed: true},
|
||||
{name: "body_error_fallback", readError: io.ErrUnexpectedEOF, wantFailed: true},
|
||||
{name: "http_error_fallback", httpStatus: http.StatusTooManyRequests, wantFailed: true},
|
||||
{name: "nonstream_keeps_first_byte_timing", event: textDelta, content: "hello", nonstream: true},
|
||||
}
|
||||
// Keep the global usage dispatcher outside the fake-clock bubble.
|
||||
coreusage.StartDefault(context.Background())
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
alias := t.Name()
|
||||
capture := &multiProviderUsageCapture{alias: alias, records: make(chan coreusage.Record, 4)}
|
||||
coreusage.RegisterNamedPlugin(alias, capture)
|
||||
t.Cleanup(func() {
|
||||
coreusage.RegisterNamedPlugin(alias, multiProviderNoopUsagePlugin{})
|
||||
})
|
||||
|
||||
var forwarded strings.Builder
|
||||
var streamFailed bool
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
transport := metaTTFTRoundTripFunc(func(req *http.Request) (*http.Response, error) {
|
||||
// All delays use synctest's virtual clock. Headers precede the first body byte.
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
reader, writer := io.Pipe()
|
||||
go func() {
|
||||
defer func() {
|
||||
if errClose := writer.Close(); errClose != nil {
|
||||
t.Errorf("close upstream writer: %v", errClose)
|
||||
}
|
||||
}()
|
||||
write := func(s string) bool {
|
||||
_, errWrite := io.WriteString(writer, s)
|
||||
return errWrite == nil
|
||||
}
|
||||
time.Sleep(90 * time.Millisecond)
|
||||
if tt.httpStatus != 0 {
|
||||
write(`{"error":{"message":"rate limited"}}`)
|
||||
return
|
||||
}
|
||||
if !write(metadata) {
|
||||
return
|
||||
}
|
||||
time.Sleep(900 * time.Millisecond)
|
||||
if tt.readError != nil {
|
||||
if errClose := writer.CloseWithError(tt.readError); errClose != nil {
|
||||
t.Errorf("close upstream with error: %v", errClose)
|
||||
}
|
||||
return
|
||||
}
|
||||
if !write(tt.event) || tt.terminal {
|
||||
return
|
||||
}
|
||||
// A later completion must not replace the first effective-token time.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
write(completed)
|
||||
}()
|
||||
status := tt.httpStatus
|
||||
if status == 0 {
|
||||
status = http.StatusOK
|
||||
}
|
||||
return &http.Response{
|
||||
StatusCode: status,
|
||||
Header: http.Header{"Content-Type": {"text/event-stream"}},
|
||||
Body: reader,
|
||||
Request: req,
|
||||
}, nil
|
||||
})
|
||||
ctx := context.WithValue(context.Background(), "cliproxy.roundtripper", http.RoundTripper(transport))
|
||||
ctx = coreusage.WithRequestedModelAlias(ctx, alias)
|
||||
executor := NewMetaExecutor(&config.Config{})
|
||||
auth := &cliproxyauth.Auth{Provider: "meta", Attributes: map[string]string{
|
||||
"api_key": "test-token", "base_url": "http://example.invalid/v1",
|
||||
}}
|
||||
format := tt.format
|
||||
if format == "" {
|
||||
format = sdktranslator.FormatOpenAIResponse
|
||||
}
|
||||
payload := []byte(`{"model":"muse-spark-1.3","input":"hello"}`)
|
||||
if format == sdktranslator.FormatOpenAI {
|
||||
payload = []byte(`{"model":"muse-spark-1.3","messages":[{"role":"user","content":"hello"}]}`)
|
||||
}
|
||||
req := cliproxyexecutor.Request{Model: "muse-spark-1.3", Payload: payload}
|
||||
opts := cliproxyexecutor.Options{SourceFormat: format, Stream: !tt.nonstream}
|
||||
if tt.nonstream {
|
||||
resp, errExecute := executor.Execute(ctx, auth, req, opts)
|
||||
if errExecute != nil {
|
||||
t.Fatalf("Execute: %v", errExecute)
|
||||
}
|
||||
forwarded.Write(resp.Payload)
|
||||
return
|
||||
}
|
||||
result, errExecute := executor.ExecuteStream(ctx, auth, req, opts)
|
||||
if tt.httpStatus != 0 {
|
||||
if errExecute == nil {
|
||||
t.Fatal("expected HTTP error")
|
||||
}
|
||||
return
|
||||
}
|
||||
if errExecute != nil {
|
||||
t.Fatalf("ExecuteStream: %v", errExecute)
|
||||
}
|
||||
for chunk := range result.Chunks {
|
||||
forwarded.Write(chunk.Payload)
|
||||
if chunk.Err != nil {
|
||||
streamFailed = true
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
record := capture.await(t)
|
||||
wantTTFT := time.Second
|
||||
wantLatency := 1100 * time.Millisecond
|
||||
if tt.terminal || tt.readError != nil {
|
||||
wantLatency = time.Second
|
||||
}
|
||||
if tt.httpStatus != 0 {
|
||||
wantLatency = 100 * time.Millisecond
|
||||
}
|
||||
if tt.nonstream || tt.readError != nil || tt.httpStatus != 0 {
|
||||
wantTTFT = 100 * time.Millisecond
|
||||
}
|
||||
if record.TTFT != wantTTFT || record.Latency != wantLatency {
|
||||
t.Errorf("TTFT/latency = %v/%v, want %v/%v", record.TTFT, record.Latency, wantTTFT, wantLatency)
|
||||
}
|
||||
if record.Failed != tt.wantFailed {
|
||||
t.Errorf("Failed = %v, want %v", record.Failed, tt.wantFailed)
|
||||
}
|
||||
if tt.httpStatus == 0 && streamFailed != tt.wantFailed {
|
||||
t.Errorf("stream error = %v, want %v", streamFailed, tt.wantFailed)
|
||||
}
|
||||
if tt.httpStatus != 0 && record.Fail.StatusCode != tt.httpStatus {
|
||||
t.Errorf("failure status = %d, want %d", record.Fail.StatusCode, tt.httpStatus)
|
||||
}
|
||||
if tt.httpStatus == 0 && record.ResponseModel != "muse-spark-1.3" {
|
||||
t.Errorf("ResponseModel = %q, want muse-spark-1.3", record.ResponseModel)
|
||||
}
|
||||
if tt.content != "" {
|
||||
if !strings.Contains(forwarded.String(), tt.content) {
|
||||
t.Errorf("response lost content %q: %s", tt.content, forwarded.String())
|
||||
}
|
||||
if got := record.Detail; got.InputTokens != 10 || got.OutputTokens != 3 || got.TotalTokens != 13 {
|
||||
t.Errorf("usage = %+v, want input/output/total = 10/3/13", got)
|
||||
}
|
||||
}
|
||||
if strings.Contains(forwarded.String(), "subscription") {
|
||||
t.Errorf("subscription telemetry leaked into the response: %s", forwarded.String())
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user