diff --git a/internal/provider/provider.go b/internal/provider/provider.go index 6ac2be8..8590192 100644 --- a/internal/provider/provider.go +++ b/internal/provider/provider.go @@ -777,7 +777,7 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch rc <- respOrErr{resp, err} }() - ch := make(chan types.UnifiedChunk, 64) + inner := make(chan types.UnifiedChunk, 64) sel := <-rc if sel.err != nil { // 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() { defer p.Release() - defer close(ch) + defer close(inner) defer sel.resp.Body.Close() scanner := bufio.NewScanner(sel.resp.Body) scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) var chunks int + var realChunks int var doneSeen bool var doneSent bool for scanner.Scan() { @@ -821,7 +822,7 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch // finish_reason downstream. if !doneSent { select { - case ch <- types.UnifiedChunk{Done: true}: + case inner <- types.UnifiedChunk{Done: true}: case <-ctx.Done(): } } @@ -844,9 +845,12 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch if ck.Done { doneSent = true } + if !errorOnlyChunk(ck) { + realChunks++ + } chunks++ select { - case ch <- ck: + case inner <- ck: case <-ctx.Done(): 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 // 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. + // 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 doneSeen || chunks > 0 { + if realChunks > 0 || (doneSeen && chunks == 0) { p.RecordSuccess(model) } else { 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 diff --git a/internal/provider/provider_test.go b/internal/provider/provider_test.go index c36e0de..66b9d66 100644 --- a/internal/provider/provider_test.go +++ b/internal/provider/provider_test.go @@ -417,7 +417,8 @@ func TestChatClientCancelNotRecorded(t *testing.T) { } // 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) { up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") @@ -429,15 +430,12 @@ func TestChatStreamEmptyBodyNotSuccess(t *testing.T) { Model: "m", Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("x")}}, }) - if err != nil { - t.Fatalf("stream: %v", err) - } - got := 0 - for range ch { - got++ - } - if got != 0 { - t.Fatalf("want empty stream, got %d chunks", got) + if err == nil { + if ch != nil { + for range ch { + } + } + t.Fatal("empty stream must fail the candidate for scheduler fallback") } eventually(t, 2*time.Second, func() bool { _, fc, _ := p.ModelHealthInfo("m") @@ -445,6 +443,62 @@ func TestChatStreamEmptyBodyNotSuccess(t *testing.T) { }, "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 // send the image-kind model id, never the best chat model. func TestImageAutoUsesImageModel(t *testing.T) {