mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-09-20 00:48:00 +00:00
fix(provider): fail the candidate on error-only streams instead of serving empty replies
Some upstreams (zen free pool) answer HTTP 200 with a single-chunk stream whose only payload is finish_reason:"network_error" and an empty delta. ChatStream used to hand that channel up as a success, so clients received a laundered empty reply (agents ended their turn mid-loop) and the broken slot kept its scheduling preference. ChatStream now holds back the first chunk: an error-only done chunk fails the candidate before any byte reaches the gateway, letting the scheduler fall through to the next source. Standard finish reasons (including instant-empty "stop") are never classified as errors.
This commit is contained in:
@ -777,7 +777,7 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
|||||||
rc <- respOrErr{resp, err}
|
rc <- respOrErr{resp, err}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
ch := make(chan types.UnifiedChunk, 64)
|
inner := make(chan types.UnifiedChunk, 64)
|
||||||
sel := <-rc
|
sel := <-rc
|
||||||
if sel.err != nil {
|
if sel.err != nil {
|
||||||
// client disconnect/cancel before the first byte: not a scheduling
|
// client disconnect/cancel before the first byte: not a scheduling
|
||||||
@ -797,11 +797,12 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
|||||||
}
|
}
|
||||||
go func() {
|
go func() {
|
||||||
defer p.Release()
|
defer p.Release()
|
||||||
defer close(ch)
|
defer close(inner)
|
||||||
defer sel.resp.Body.Close()
|
defer sel.resp.Body.Close()
|
||||||
scanner := bufio.NewScanner(sel.resp.Body)
|
scanner := bufio.NewScanner(sel.resp.Body)
|
||||||
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||||
var chunks int
|
var chunks int
|
||||||
|
var realChunks int
|
||||||
var doneSeen bool
|
var doneSeen bool
|
||||||
var doneSent bool
|
var doneSent bool
|
||||||
for scanner.Scan() {
|
for scanner.Scan() {
|
||||||
@ -821,7 +822,7 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
|||||||
// finish_reason downstream.
|
// finish_reason downstream.
|
||||||
if !doneSent {
|
if !doneSent {
|
||||||
select {
|
select {
|
||||||
case ch <- types.UnifiedChunk{Done: true}:
|
case inner <- types.UnifiedChunk{Done: true}:
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -844,9 +845,12 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
|||||||
if ck.Done {
|
if ck.Done {
|
||||||
doneSent = true
|
doneSent = true
|
||||||
}
|
}
|
||||||
|
if !errorOnlyChunk(ck) {
|
||||||
|
realChunks++
|
||||||
|
}
|
||||||
chunks++
|
chunks++
|
||||||
select {
|
select {
|
||||||
case ch <- ck:
|
case inner <- ck:
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@ -857,15 +861,71 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
|||||||
// success nor failure for scheduling purposes. A 200 that produced
|
// success nor failure for scheduling purposes. A 200 that produced
|
||||||
// zero chunks and no [DONE] is an empty stream, i.e. a failure
|
// zero chunks and no [DONE] is an empty stream, i.e. a failure
|
||||||
// before the first chunk — record it so the slot can fall back.
|
// before the first chunk — record it so the slot can fall back.
|
||||||
|
// A stream whose only payload was error-only done chunks (e.g.
|
||||||
|
// finish_reason:"network_error") is likewise a failure, not a success.
|
||||||
if ctx.Err() == nil && scanner.Err() == nil {
|
if ctx.Err() == nil && scanner.Err() == nil {
|
||||||
if doneSeen || chunks > 0 {
|
if realChunks > 0 || (doneSeen && chunks == 0) {
|
||||||
p.RecordSuccess(model)
|
p.RecordSuccess(model)
|
||||||
} else {
|
} else {
|
||||||
p.RecordFailure(model, 0)
|
p.RecordFailure(model, 0)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
return ch, nil
|
|
||||||
|
// Hold back the first chunk to validate the stream actually carries
|
||||||
|
// content: some upstreams answer HTTP 200 with a degenerate stream whose
|
||||||
|
// only payload is an error finish reason (zen free pool sends
|
||||||
|
// finish_reason:"network_error" with empty delta). Failing the candidate
|
||||||
|
// here — before any byte reaches the gateway — lets the scheduler fall
|
||||||
|
// through to the next source instead of serving the client an empty reply.
|
||||||
|
type heldChunk struct {
|
||||||
|
ck types.UnifiedChunk
|
||||||
|
ok bool
|
||||||
|
}
|
||||||
|
var first heldChunk
|
||||||
|
select {
|
||||||
|
case ck, ok := <-inner:
|
||||||
|
first = heldChunk{ck, ok}
|
||||||
|
case <-ctx.Done():
|
||||||
|
return nil, ctx.Err()
|
||||||
|
}
|
||||||
|
if !first.ok {
|
||||||
|
return nil, fmt.Errorf("provider %s: empty stream", p.Name())
|
||||||
|
}
|
||||||
|
if errorOnlyChunk(first.ck) {
|
||||||
|
go func() {
|
||||||
|
for range inner {
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
return nil, fmt.Errorf("provider %s: upstream returned %q stream",
|
||||||
|
p.Name(), first.ck.FinishReason)
|
||||||
|
}
|
||||||
|
out := make(chan types.UnifiedChunk, 64)
|
||||||
|
go func() {
|
||||||
|
defer close(out)
|
||||||
|
out <- first.ck
|
||||||
|
for ck := range inner {
|
||||||
|
out <- ck
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
return out, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// errorOnlyChunk reports whether ck carries nothing but an upstream error
|
||||||
|
// signal: a done chunk with a non-standard finish reason and zero content,
|
||||||
|
// tool calls, reasoning text or usage. Standard OpenAI finish reasons are
|
||||||
|
// never classified as errors, so legitimate instant-empty completions
|
||||||
|
// (finish_reason:"stop", no output) still reach the client.
|
||||||
|
func errorOnlyChunk(ck types.UnifiedChunk) bool {
|
||||||
|
if !ck.Done || ck.FinishReason == "" {
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
switch ck.FinishReason {
|
||||||
|
case "stop", "length", "tool_calls", "function_call", "content_filter":
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
return ck.Content == "" && len(ck.ToolCalls) == 0 &&
|
||||||
|
ck.ReasoningContent == "" && ck.Usage == nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// Image generates images via /v1/images/generations. Same scheduling-state
|
// Image generates images via /v1/images/generations. Same scheduling-state
|
||||||
|
|||||||
@ -417,7 +417,8 @@ func TestChatClientCancelNotRecorded(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// TestChatStreamEmptyBodyNotSuccess: a 200 that yields zero chunks and no
|
// TestChatStreamEmptyBodyNotSuccess: a 200 that yields zero chunks and no
|
||||||
// [DONE] is a failure before the first chunk — the slot must back off.
|
// [DONE] must fail the candidate (error return, no delivered chunks) so the
|
||||||
|
// scheduler falls through to the next source, and record the failure.
|
||||||
func TestChatStreamEmptyBodyNotSuccess(t *testing.T) {
|
func TestChatStreamEmptyBodyNotSuccess(t *testing.T) {
|
||||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
w.Header().Set("Content-Type", "text/event-stream")
|
w.Header().Set("Content-Type", "text/event-stream")
|
||||||
@ -429,15 +430,12 @@ func TestChatStreamEmptyBodyNotSuccess(t *testing.T) {
|
|||||||
Model: "m",
|
Model: "m",
|
||||||
Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("x")}},
|
Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("x")}},
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err == nil {
|
||||||
t.Fatalf("stream: %v", err)
|
if ch != nil {
|
||||||
}
|
for range ch {
|
||||||
got := 0
|
}
|
||||||
for range ch {
|
}
|
||||||
got++
|
t.Fatal("empty stream must fail the candidate for scheduler fallback")
|
||||||
}
|
|
||||||
if got != 0 {
|
|
||||||
t.Fatalf("want empty stream, got %d chunks", got)
|
|
||||||
}
|
}
|
||||||
eventually(t, 2*time.Second, func() bool {
|
eventually(t, 2*time.Second, func() bool {
|
||||||
_, fc, _ := p.ModelHealthInfo("m")
|
_, fc, _ := p.ModelHealthInfo("m")
|
||||||
@ -445,6 +443,62 @@ func TestChatStreamEmptyBodyNotSuccess(t *testing.T) {
|
|||||||
}, "empty stream must record a failure")
|
}, "empty stream must record a failure")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestChatStreamErrorFinishFailsCandidate: zen free pool answers HTTP 200
|
||||||
|
// with a single-chunk stream whose only payload is finish_reason:"network_error"
|
||||||
|
// and an empty delta. That must fail the candidate (not deliver a laundered
|
||||||
|
// empty reply), so the scheduler can fall through to the next source.
|
||||||
|
func TestChatStreamErrorFinishFailsCandidate(t *testing.T) {
|
||||||
|
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "text/event-stream")
|
||||||
|
fmt.Fprint(w, `data: {"choices":[{"index":0,"finish_reason":"network_error","delta":{"role":"assistant","content":""}}]}`+"\n\n")
|
||||||
|
fmt.Fprint(w, "data: [DONE]\n\n")
|
||||||
|
}))
|
||||||
|
defer up.Close()
|
||||||
|
p := newTestProvider(t, src("mock", up.URL, "opencode", "m"))
|
||||||
|
_, err := p.ChatStream(context.Background(), &types.ChatRequest{
|
||||||
|
Model: "m",
|
||||||
|
Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("x")}},
|
||||||
|
})
|
||||||
|
if err == nil {
|
||||||
|
t.Fatal("network_error-only stream must fail the candidate")
|
||||||
|
}
|
||||||
|
eventually(t, 2*time.Second, func() bool {
|
||||||
|
_, fc, _ := p.ModelHealthInfo("m")
|
||||||
|
return fc >= 1
|
||||||
|
}, "error-only stream must record a failure")
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestChatStreamInstantEmptyStopStillDelivered: a legitimate completion that
|
||||||
|
// ends immediately with the standard finish_reason:"stop" and zero content is
|
||||||
|
// NOT an upstream error and must still reach the client.
|
||||||
|
func TestChatStreamInstantEmptyStopStillDelivered(t *testing.T) {
|
||||||
|
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
|
w.Header().Set("Content-Type", "text/event-stream")
|
||||||
|
fmt.Fprint(w, `data: {"choices":[{"index":0,"finish_reason":"stop","delta":{}}]}`+"\n\n")
|
||||||
|
fmt.Fprint(w, "data: [DONE]\n\n")
|
||||||
|
}))
|
||||||
|
defer up.Close()
|
||||||
|
p := newTestProvider(t, src("mock", up.URL, "openai", "m"))
|
||||||
|
ch, err := p.ChatStream(context.Background(), &types.ChatRequest{
|
||||||
|
Model: "m",
|
||||||
|
Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("x")}},
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("standard stop stream must be delivered: %v", err)
|
||||||
|
}
|
||||||
|
n := 0
|
||||||
|
for range ch {
|
||||||
|
n++
|
||||||
|
}
|
||||||
|
if n == 0 {
|
||||||
|
t.Fatal("expected at least the terminating chunk")
|
||||||
|
}
|
||||||
|
eventually(t, 2*time.Second, func() bool {
|
||||||
|
_, fc, _ := p.ModelHealthInfo("m")
|
||||||
|
return fc == 0
|
||||||
|
}, "standard stop must count as success")
|
||||||
|
}
|
||||||
|
|
||||||
// TestImageAutoUsesImageModel: AUTO image generation on a mixed source must
|
// TestImageAutoUsesImageModel: AUTO image generation on a mixed source must
|
||||||
// send the image-kind model id, never the best chat model.
|
// send the image-kind model id, never the best chat model.
|
||||||
func TestImageAutoUsesImageModel(t *testing.T) {
|
func TestImageAutoUsesImageModel(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user