fix(bench): stream reasoning_content and accept clean zero-content finishes

internal/vllmbench/http_load.go

@@ -114,8 +114,9 @@ type openAIChoice struct {
 }
 
 type openAIMessage struct {
-	Role    string `json:"role,omitempty"`
-	Content string `json:"content,omitempty"`
+	Role             string `json:"role,omitempty"`
+	Content          string `json:"content,omitempty"`
+	ReasoningContent string `json:"reasoning_content,omitempty"`
 }
 
 type openAIUsage struct {
@@ -610,18 +611,25 @@ func (stream *httpStreamResult) consume(body io.Reader) *httpLoadFailure {
 		}
 	}
 	stream.completedAt = time.Now().UTC()
-	if err := scanner.Err(); err != nil {
-		return newHTTPLoadFailure("response_read", "", err.Error(), stream.completedAt, nil)
-	}
-	// EOF before [DONE] is a truncated stream: recording it as completed
-	// would silently keep partial output and fallback token counts.
-	if !terminated {
+	return stream.finalize(terminated, scanner.Err())
+}
+
+// finalize validates a fully-read stream. EOF before [DONE] is a truncated
+// stream (recording it would keep partial output). A stream that finished
+// cleanly is valid even with zero streamed tokens — a 1-token prefill point
+// or a reasoning model that emitted only a final message; only a stream with
+// neither any token nor a finish reason is malformed.
+func (stream *httpStreamResult) finalize(terminated bool, scanErr error) *httpLoadFailure {
+	switch {
+	case scanErr != nil:
+		return newHTTPLoadFailure("response_read", "", scanErr.Error(), stream.completedAt, nil)
+	case !terminated:
 		return newHTTPLoadFailure("response_read", "", "stream ended before [DONE] terminator", stream.completedAt, nil)
+	case stream.firstTokenAt == nil && stream.finishReason == "":
+		return newHTTPLoadFailure("response_shape", "", "stream produced neither tokens nor a finish reason", stream.completedAt, nil)
+	default:
+		return nil
 	}
-	if stream.firstTokenAt == nil {
-		return newHTTPLoadFailure("response_shape", "", "stream produced no completion content", stream.completedAt, nil)
-	}
-	return nil
 }
 
 func ssePayload(line string) (string, bool) {
@@ -669,11 +677,18 @@ func (stream *httpStreamResult) applyChoice(choice openAIStreamChoice) {
 	stream.content.WriteString(text)
 }
 
-// streamChoiceText extracts the token text of one stream choice: chat chunks
-// carry a delta message, completions chunks carry plain text.
+// streamChoiceText extracts the token text of one stream choice. Chat chunks
+// carry a delta message; reasoning models stream their first tokens as
+// reasoning_content before any content, so both count for TTFT/ITL timing.
+// Completions chunks carry plain text.
 func streamChoiceText(choice openAIStreamChoice) string {
-	if choice.Delta != nil && choice.Delta.Content != "" {
-		return choice.Delta.Content
+	if choice.Delta != nil {
+		if choice.Delta.Content != "" {
+			return choice.Delta.Content
+		}
+		if choice.Delta.ReasoningContent != "" {
+			return choice.Delta.ReasoningContent
+		}
 	}
 	return choice.Text
 }
@@ -687,15 +702,24 @@ func (stream *httpStreamResult) applyToSample(sample RequestSample, request Cano
 	if stream.firstByteAt != nil {
 		sample.FirstByteMillis = stream.firstByteAt.Sub(sample.StartedAt).Seconds() * 1000
 	}
-	sample.TTFTMillis = stream.firstTokenAt.Sub(sample.StartedAt).Seconds() * 1000
 	sample.PromptTokens = usageInt(stream.usage.PromptTokens, request.InputTokensExpected)
-	sample.CompletionTokens = usageInt(stream.usage.CompletionTokens, request.OutputTokensExpected)
+	// Completion tokens fall back to the observed streamed-chunk count, never
+	// to the requested output length: a backend that omits usage on an empty
+	// stream must not be credited phantom output tokens.
+	sample.CompletionTokens = usageInt(stream.usage.CompletionTokens, stream.tokenChunks)
 	sample.TotalTokens = usageInt(stream.usage.TotalTokens, sample.PromptTokens+sample.CompletionTokens)
-	if sample.CompletionTokens > 1 {
-		sample.TPOTMillis = stream.completedAt.Sub(*stream.firstTokenAt).Seconds() * 1000 / float64(sample.CompletionTokens-1)
-	}
-	if stream.tokenChunks > 1 {
-		sample.ITLMeanMillis = stream.lastTokenAt.Sub(*stream.firstTokenAt).Seconds() * 1000 / float64(stream.tokenChunks-1)
+	// firstTokenAt is nil only for a clean finish that streamed no token
+	// (accepted above); such a point contributes no TTFT/TPOT/ITL, which is
+	// honest rather than a fabricated zero. Token counts are assigned first
+	// so TPOT sees the real completion-token count.
+	if stream.firstTokenAt != nil {
+		sample.TTFTMillis = stream.firstTokenAt.Sub(sample.StartedAt).Seconds() * 1000
+		if sample.CompletionTokens > 1 {
+			sample.TPOTMillis = stream.completedAt.Sub(*stream.firstTokenAt).Seconds() * 1000 / float64(sample.CompletionTokens-1)
+		}
+		if stream.tokenChunks > 1 {
+			sample.ITLMeanMillis = stream.lastTokenAt.Sub(*stream.firstTokenAt).Seconds() * 1000 / float64(stream.tokenChunks-1)
+		}
 	}
 	sample.ResponseSHA256 = sha256Hex([]byte(stream.content.String()))
 	sample.ResponseMetadata = streamResponseMetadata(stream)

internal/vllmbench/http_stream_test.go

@@ -97,6 +97,9 @@ func TestStreamingBenchmarkMeasuresTTFT(t *testing.T) {
 		if sample.ITLMeanMillis <= 0 {
 			t.Fatalf("sample ITL = %.3f, want positive from chunk gaps", sample.ITLMeanMillis)
 		}
+		if sample.TPOTMillis <= 0 {
+			t.Fatalf("sample TPOT = %.3f, want positive for a multi-token completion", sample.TPOTMillis)
+		}
 		if sample.PromptTokens != 32 || sample.CompletionTokens != 4 {
 			t.Fatalf("sample tokens = %d/%d, want 32/4 from stream usage", sample.PromptTokens, sample.CompletionTokens)
 		}
@@ -136,21 +139,66 @@ func TestNonStreamingBenchmarkRecordsNoTTFT(t *testing.T) {
 	}
 }
 
-func TestStreamWithNoContentFailsShape(t *testing.T) {
-	server := sseTestServer(t, 0, 0, 0)
+func TestStreamWithNeitherTokenNorFinishFailsShape(t *testing.T) {
+	// A stream that ends ([DONE]) having emitted neither a token nor a
+	// finish_reason is genuinely malformed and must fail. (A clean finish
+	// with zero content is valid — see TestStreamingFinishWithNoContentCompletes.)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		fl := w.(http.Flusher)
+		fmt.Fprint(w, "data: {\"id\":\"x\",\"choices\":[{\"delta\":{}}]}\n\n")
+		fmt.Fprint(w, "data: [DONE]\n\n")
+		fl.Flush()
+	}))
 	defer server.Close()
 	result, err := runHTTPBenchmark(context.Background(), streamTestPlannedRun(server.URL, nil))
 	if err != nil {
 		t.Fatal(err)
 	}
 	if result.Failed != 2 {
-		t.Fatalf("failed = %d, want 2 for empty streams", result.Failed)
+		t.Fatalf("failed = %d, want 2 for streams with no token and no finish reason", result.Failed)
 	}
 	if result.RequestSamples[0].ErrorType != "response_shape" {
 		t.Fatalf("error type = %q, want response_shape", result.RequestSamples[0].ErrorType)
 	}
 }
 
+func TestStreamWithoutUsageDoesNotFabricateTokens(t *testing.T) {
+	// A backend that ignores stream_options.include_usage: content streams
+	// but no usage chunk arrives. Completion tokens must reflect observed
+	// chunks, never the requested output length (which would inflate
+	// throughput). Empty-but-finished streams must land at zero.
+	makeServer := func(contentChunks int) *httptest.Server {
+		return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Set("Content-Type", "text/event-stream")
+			fl := w.(http.Flusher)
+			for i := 0; i < contentChunks; i++ {
+				fmt.Fprintf(w, "data: {\"id\":\"u\",\"choices\":[{\"delta\":{\"content\":\"t\"}}]}\n\n")
+			}
+			fmt.Fprint(w, "data: {\"id\":\"u\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n")
+			fmt.Fprint(w, "data: [DONE]\n\n")
+			fl.Flush()
+		}))
+	}
+	// request.OutputTokensExpected is 8 (RandomOutputLen); assert we never see it.
+	for _, tc := range []struct{ chunks, wantTokens int }{{0, 0}, {3, 3}} {
+		srv := makeServer(tc.chunks)
+		result, err := runHTTPBenchmark(context.Background(), streamTestPlannedRun(srv.URL, nil))
+		srv.Close()
+		if err != nil {
+			t.Fatalf("chunks=%d: %v", tc.chunks, err)
+		}
+		if result.Completed != 2 {
+			t.Fatalf("chunks=%d: completed=%d, want 2", tc.chunks, result.Completed)
+		}
+		for _, s := range result.RequestSamples {
+			if s.CompletionTokens != tc.wantTokens {
+				t.Fatalf("chunks=%d: completion_tokens=%d, want %d (observed chunks, not the requested 8)", tc.chunks, s.CompletionTokens, tc.wantTokens)
+			}
+		}
+	}
+}
+
 func TestExtraBodyCannotFlipStreaming(t *testing.T) {
 	client := openAIHTTPClient{
 		profile: Profile{Model: "m"},
@@ -294,3 +342,73 @@ func TestStreamErrorChunkFailsRequest(t *testing.T) {
 		t.Fatalf("error = %q/%q, want server_error/boom", sample.ErrorType, sample.ErrorMessage)
 	}
 }
+
+func TestStreamingReasoningContentCountsForTTFT(t *testing.T) {
+	// A reasoning model streams its first tokens as reasoning_content, then
+	// a normal content token, then finishes. TTFT must be observed from the
+	// first reasoning token, not lost.
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		fl := w.(http.Flusher)
+		time.Sleep(20 * time.Millisecond)
+		fmt.Fprint(w, "data: {\"id\":\"r1\",\"choices\":[{\"delta\":{\"reasoning_content\":\"think\"}}]}\n\n")
+		fl.Flush()
+		time.Sleep(5 * time.Millisecond)
+		fmt.Fprint(w, "data: {\"id\":\"r1\",\"choices\":[{\"delta\":{\"content\":\"answer\"}}]}\n\n")
+		fl.Flush()
+		fmt.Fprint(w, "data: {\"id\":\"r1\",\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n")
+		fmt.Fprint(w, "data: {\"id\":\"r1\",\"choices\":[],\"usage\":{\"prompt_tokens\":32,\"completion_tokens\":2,\"total_tokens\":34}}\n\n")
+		fmt.Fprint(w, "data: [DONE]\n\n")
+		fl.Flush()
+	}))
+	defer server.Close()
+	result, err := runHTTPBenchmark(context.Background(), streamTestPlannedRun(server.URL, nil))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.Completed != 2 || result.TTFTSource != TTFTSourceStream {
+		t.Fatalf("completed=%d ttft_source=%q, want 2 / stream", result.Completed, result.TTFTSource)
+	}
+	for _, s := range result.RequestSamples {
+		if s.TTFTMillis <= 0 {
+			t.Fatalf("TTFT=%.1f, want reasoning_content to seed TTFT", s.TTFTMillis)
+		}
+		if s.ITLMeanMillis <= 0 {
+			t.Fatalf("ITL=%.3f, want gap between reasoning and content tokens", s.ITLMeanMillis)
+		}
+	}
+}
+
+func TestStreamingFinishWithNoContentCompletes(t *testing.T) {
+	// A 1-token prefill / empty-generation stream that finishes cleanly with
+	// zero streamed tokens must complete, not fail response_shape. It carries
+	// no TTFT (honest), and the request still counts as completed.
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		fl := w.(http.Flusher)
+		fmt.Fprint(w, "data: {\"id\":\"p1\",\"choices\":[{\"delta\":{},\"finish_reason\":\"length\"}]}\n\n")
+		fl.Flush()
+		fmt.Fprint(w, "data: {\"id\":\"p1\",\"choices\":[],\"usage\":{\"prompt_tokens\":16000,\"completion_tokens\":1,\"total_tokens\":16001}}\n\n")
+		fmt.Fprint(w, "data: [DONE]\n\n")
+		fl.Flush()
+	}))
+	defer server.Close()
+	result, err := runHTTPBenchmark(context.Background(), streamTestPlannedRun(server.URL, nil))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.Completed != 2 || result.Failed != 0 {
+		t.Fatalf("completed/failed=%d/%d, want 2/0: a clean finish with no content is valid", result.Completed, result.Failed)
+	}
+	for _, s := range result.RequestSamples {
+		if s.Status != "completed" || !s.Streamed {
+			t.Fatalf("sample status/streamed=%s/%v, want completed/true", s.Status, s.Streamed)
+		}
+		if s.TTFTMillis != 0 {
+			t.Fatalf("TTFT=%.1f, want 0 (no token was streamed)", s.TTFTMillis)
+		}
+	}
+	if result.TTFTSource != "" {
+		t.Fatalf("ttft_source=%q, want empty when no point produced a streamed token", result.TTFTSource)
+	}
+}