diff --git a/changelog/unreleased/completed-stream-not-canceled.md b/changelog/unreleased/completed-stream-not-canceled.md new file mode 100644 index 0000000..272f09b --- /dev/null +++ b/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. + +### 中文 + +- 流已经结束后客户端再断开,请求日志记为成功。流还没结束就断开,仍然记为取消。 diff --git a/internal/gateway/anthropic.go b/internal/gateway/anthropic.go index 33bcf5b..7ed713f 100644 --- a/internal/gateway/anthropic.go +++ b/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 { diff --git a/internal/gateway/compat_stream.go b/internal/gateway/compat_stream.go index a859f6d..04caab5 100644 --- a/internal/gateway/compat_stream.go +++ b/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 } diff --git a/internal/gateway/native_responses.go b/internal/gateway/native_responses.go index f930d89..d2e65f7 100644 --- a/internal/gateway/native_responses.go +++ b/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 } diff --git a/internal/gateway/openai.go b/internal/gateway/openai.go index d573728..38e9b7d 100644 --- a/internal/gateway/openai.go +++ b/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 { diff --git a/internal/gateway/openai_stream.go b/internal/gateway/openai_stream.go index ed704c2..6b3977b 100644 --- a/internal/gateway/openai_stream.go +++ b/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 } diff --git a/internal/gateway/responses.go b/internal/gateway/responses.go index aa30a43..b0d3fed 100644 --- a/internal/gateway/responses.go +++ b/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 { diff --git a/internal/gateway/responses_test.go b/internal/gateway/responses_test.go index f77ad1b..1a7a962 100644 --- a/internal/gateway/responses_test.go +++ b/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" {