fix(logs): count a finished stream as success after disconnect

changelog/unreleased/completed-stream-not-canceled.md

@@ -0,0 +1,7 @@
+### English
+
+- A stream that already finished is logged as success when the client disconnects afterward. A disconnect before the stream finishes is still canceled.
+
+### 中文
+
+- 流已经结束后客户端再断开，请求日志记为成功。流还没结束就断开，仍然记为取消。

internal/gateway/anthropic.go

@@ -68,15 +68,15 @@ func (h *Handler) handleAnthropicMessagesStream(w http.ResponseWriter, r *http.R
 	}
 	writer := compatibilityStreamWriter(w)
 	stats, relayErr := RelayAnthropicStream(writer, upstream.Response.Body, execution.RequestID, firstNonEmpty(execution.PublicModel, execution.Request.Model))
-	status := streamRequestStatus(relayErr)
-	if r.Context().Err() != nil || errors.Is(relayErr, context.Canceled) || errors.Is(relayErr, context.DeadlineExceeded) {
-		status = accounts.RequestStatusCanceled
-	}
+	status := streamLogStatus(relayErr, stats.SawDone, r.Context().Err() != nil)
 	h.recordStreamDiagnostic(execution.RequestID, upstream.Response, execution.Started, stats, relayErr, r.Context().Err())
 	ttfb := streamTTFB(execution.Started, upstream.TTFBMs, stats)
 	logErr := relayErr
-	if status == accounts.RequestStatusCanceled {
+	switch status {
+	case accounts.RequestStatusCanceled:
 		logErr = context.Canceled
+	case accounts.RequestStatusOK:
+		logErr = nil
 	}
 	h.finishCompatibility(execution, upstream.AccountID, upstream.Provider, upstream.Routing, status, ttfb, &stats, logErr, upstream.AttemptCount, upstream.ReasoningLevel)
 	if relayErr == nil {

internal/gateway/compat_stream.go

@@ -54,6 +54,24 @@ func streamRequestStatus(err error) string {
 	return accounts.RequestStatusError
 }
 
+// streamLogStatus records a completed upstream stream as success even when the
+// caller closes the connection after the terminal frame. A disconnect before
+// that frame stays canceled; anything else stays an error.
+func streamLogStatus(err error, completed bool, clientGone bool) string {
+	status := streamRequestStatus(err)
+	if err == nil {
+		return status
+	}
+	disconnected := clientGone || IsStreamClientDisconnect(err) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)
+	if !disconnected {
+		return accounts.RequestStatusError
+	}
+	if completed {
+		return accounts.RequestStatusOK
+	}
+	return accounts.RequestStatusCanceled
+}
+
 func streamTTFB(started time.Time, fallback int, stats StreamRelayStats) int {
 	if stats.FirstTokenAt == nil {
 		return fallback
@@ -196,6 +214,7 @@ func consumeOpenAIStream(body io.Reader, handle func(json.RawMessage, *streamedC
 		}
 		if usage, ok := ParseStreamUsageLine("data: " + string(payload)); ok {
 			usage.FirstTokenAt = stats.FirstTokenAt
+			usage.SawDone = stats.SawDone
 			stats = usage
 		}
 		beforeContent := output.content.Len()
@@ -225,6 +244,7 @@ func consumeOpenAIStream(body io.Reader, handle func(json.RawMessage, *streamedC
 	if !sawDone {
 		return stats, output, executor.StreamIncompleteError()
 	}
+	stats.SawDone = true
 	return stats, output, nil
 }
 
@@ -369,6 +389,7 @@ func RelayAnthropicStream(writer io.Writer, body io.Reader, requestID, model str
 	if err := writeSSEEvent(writer, "message_stop", map[string]string{"type": "message_stop"}); err != nil {
 		return stats, err
 	}
+	stats.SawDone = true
 	return stats, nil
 }
 

internal/gateway/native_responses.go

@@ -85,6 +85,7 @@ func RelayNativeResponsesStream(writer io.Writer, body io.Reader, names map[stri
 	if !sawTerminal {
 		return stats, executor.StreamIncompleteError()
 	}
+	stats.SawDone = true
 	return stats, nil
 }
 

internal/gateway/openai.go

@@ -69,14 +69,7 @@ func (h *Handler) HandleChatCompletions(w http.ResponseWriter, r *http.Request)
 			flusher.Flush()
 		}
 		stats, relayErr := RelayOpenAIStream(w, upstream.Response.Body)
-		status := accounts.RequestStatusOK
-		if relayErr != nil {
-			if IsStreamClientDisconnect(relayErr) || r.Context().Err() != nil || errors.Is(relayErr, context.Canceled) || errors.Is(relayErr, context.DeadlineExceeded) {
-				status = accounts.RequestStatusCanceled
-			} else {
-				status = accounts.RequestStatusError
-			}
-		}
+		status := streamLogStatus(relayErr, stats.SawDone, r.Context().Err() != nil)
 		h.recordStreamDiagnostic(requestID, upstream.Response, started, stats, relayErr, r.Context().Err())
 		ttfb := upstream.TTFBMs
 		if stats.FirstTokenAt != nil {
@@ -86,8 +79,11 @@ func (h *Handler) HandleChatCompletions(w http.ResponseWriter, r *http.Request)
 			}
 		}
 		logErr := relayErr
-		if status == accounts.RequestStatusCanceled {
+		switch status {
+		case accounts.RequestStatusCanceled:
 			logErr = context.Canceled
+		case accounts.RequestStatusOK:
+			logErr = nil
 		}
 		h.finishRequestLog(requestID, started, req, publicModel, upstream.AccountID, firstNonEmpty(upstream.Provider, providerFilter), upstream.Routing, status, ttfb, &stats, logErr, upstream.AttemptCount, upstream.ReasoningLevel)
 		if relayErr == nil {

internal/gateway/openai_stream.go

@@ -118,17 +118,20 @@ func RelayOpenAIStream(w http.ResponseWriter, body io.Reader) (stats StreamRelay
 				usage.BytesRead = stats.BytesRead
 				usage.LastEvent = stats.LastEvent
 				usage.SawDone = stats.SawDone
+				usage.FinishReason = stats.FinishReason
 				stats = usage
 			}
 			if strings.TrimSpace(strings.TrimPrefix(line, "data:")) == "[DONE]" {
 				sawDone = true
-				stats.SawDone = true
 			}
 		}
 		output := strings.Join(frame, "\n") + "\n\n"
 		if _, err := io.WriteString(writer, output); err != nil {
 			return &StreamRelayWriteError{err: err}
 		}
+		if sawDone {
+			stats.SawDone = true
+		}
 		frame = nil
 		return nil
 	}

internal/gateway/responses.go

@@ -134,18 +134,19 @@ func (h *Handler) handleResponsesStream(w http.ResponseWriter, r *http.Request,
 	} else {
 		stats, relayErr = RelayResponsesStream(writer, upstream.Response.Body, execution.RequestID, firstNonEmpty(execution.PublicModel, execution.Request.Model), execution.ResponseToolNames)
 	}
-	status := streamRequestStatus(relayErr)
+	completed := stats.SawDone || stats.FinishReason != ""
+	status := streamLogStatus(relayErr, completed, r.Context().Err() != nil)
 	if relayErr == nil {
 		status = responsesRequestStatus(stats.FinishReason)
 	}
-	if r.Context().Err() != nil || errors.Is(relayErr, context.Canceled) || errors.Is(relayErr, context.DeadlineExceeded) {
-		status = accounts.RequestStatusCanceled
-	}
 	h.recordStreamDiagnostic(execution.RequestID, upstream.Response, execution.Started, stats, relayErr, r.Context().Err())
 	ttfb := streamTTFB(execution.Started, upstream.TTFBMs, stats)
 	logErr := relayErr
-	if status == accounts.RequestStatusCanceled {
+	switch status {
+	case accounts.RequestStatusCanceled:
 		logErr = context.Canceled
+	case accounts.RequestStatusOK:
+		logErr = nil
 	}
 	h.finishCompatibility(execution, upstream.AccountID, upstream.Provider, upstream.Routing, status, ttfb, &stats, logErr, upstream.AttemptCount, upstream.ReasoningLevel)
 	if relayErr == nil {

internal/gateway/responses_test.go

@@ -1,11 +1,53 @@
 package gateway
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
+	"net/http/httptest"
 	"strings"
 	"testing"
+
+	"github.com/caigee-cmd/cli2api/internal/accounts"
 )
 
+func TestRelayOpenAIStreamDoneSurvivesUsageReplacement(t *testing.T) {
+	body := strings.Join([]string{
+		`data: {"choices":[{"delta":{"content":"hi"}}]}`,
+		"",
+		`data: {"usage":{"prompt_tokens":1,"completion_tokens":1}}`,
+		`data: [DONE]`,
+		"",
+	}, "\n")
+	recorder := httptest.NewRecorder()
+	stats, err := RelayOpenAIStream(recorder, strings.NewReader(body))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !stats.SawDone || !strings.Contains(recorder.Body.String(), "[DONE]") {
+		t.Fatalf("stats=%+v body=%s", stats, recorder.Body.String())
+	}
+}
+
+func TestStreamLogStatusKeepsCompletedDisconnectSuccessful(t *testing.T) {
+	disconnect := &StreamRelayWriteError{err: errors.New("client closed")}
+	if got := streamLogStatus(disconnect, true, true); got != accounts.RequestStatusOK {
+		t.Fatalf("completed disconnect status=%s", got)
+	}
+	if got := streamLogStatus(context.Canceled, true, false); got != accounts.RequestStatusOK {
+		t.Fatalf("completed context cancel status=%s", got)
+	}
+	if got := streamLogStatus(context.Canceled, false, true); got != accounts.RequestStatusCanceled {
+		t.Fatalf("mid-stream cancel status=%s", got)
+	}
+	if got := streamLogStatus(errors.New("upstream reset"), true, false); got != accounts.RequestStatusError {
+		t.Fatalf("upstream failure status=%s", got)
+	}
+	if got := streamLogStatus(nil, false, false); got != accounts.RequestStatusOK {
+		t.Fatalf("clean stream status=%s", got)
+	}
+}
+
 func TestResponsesResponseMapsLengthToIncomplete(t *testing.T) {
 	response := responsesResponse("req", "model", "", "reasoning", nil, 10, 32, "length", nil, nil)
 	if response["status"] != "incomplete" {