From 65ef6927b03383944edab47eecfa4a24a28cd7e1 Mon Sep 17 00:00:00 2001 From: liyang Date: Sat, 6 Jun 2026 21:24:05 +0800 Subject: [PATCH 1/2] fix(claude-stream): always emit closing events when upstream sends finish_reason MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Problem When converting OpenAI streaming responses to Claude format, if the OpenAI-compatible upstream sends a chunk with `finish_reason` but no `usage` field, and never follows up with a separate usage-only chunk, the Claude stream is silently truncated — `message_delta` and `message_stop` events are never emitted. This violates the Claude Messages SSE protocol, which requires every stream to end with `message_stop`. As a result, Claude clients (e.g. Claude Code) hang indefinitely waiting for the stream to terminate. ## Affected Upstreams This is the protocol behavior of any OpenAI-compatible upstream that omits `usage` when the client doesn't pass `stream_options.include_usage=true`, which is fully compliant with the OpenAI spec. Confirmed reproducers: - LiteLLM proxy - Custom OpenAI-compatible gateways - Some Azure OpenAI deployments ## Root Cause In `service/convert.go`, the `doneChunk` branch deferred emitting closing events when `usage` was missing, expecting a follow-up usage-only chunk that never arrives: if oaiUsage == nil { oaiUsage = info.ClaudeConvertInfo.Usage // Defer closing until usage is available... return claudeResponses // ← stream silently truncated } ## Fix 1. **service/convert.go**: emit `message_delta` + `message_stop` immediately in the `doneChunk` branch, regardless of whether `usage` is available. The `message_delta` event includes usage when available, omits it otherwise — both forms are valid per Claude protocol spec. 2. **service/convert.go**: export `BuildClaudeUsageFromOpenAIUsage` and `StopReasonOpenAI2Claude` (as wrappers) so the fallback path in `relay/channel/openai/helper.go` can reuse the conversion logic without duplication. 3. **relay/channel/openai/helper.go**: add fallback closing events in `HandleFinalResponse` for cases where the stream is truncated without `finish_reason` at all (e.g. connection dropped, network timeout). Ensures `message_stop` is always emitted. ## Protocol Responsibility The Anthropic Messages SSE spec requires every stream to terminate with `message_stop`: > The end of the stream is indicated by a `message_stop` event. > https://docs.anthropic.com/en/api/messages-streaming new-api is a protocol converter (OpenAI ↔ Claude), and its contract is to accept any *valid* OpenAI input and produce *valid* Claude output. The OpenAI input here is fully spec-compliant — the fix belongs in new-api. ## Testing Verified locally with: - LiteLLM proxy (gpt-4) → Claude Code: previously hung, now closes correctly - Direct OpenAI API → Claude Code: still works (no regression) - Tool calls / streaming with thinking blocks: still works (no regression) --- relay/channel/openai/helper.go | 30 +++++++++++++++++++++++ service/convert.go | 45 +++++++++++++++++++++++++++++++--- 2 files changed, 72 insertions(+), 3 deletions(-) diff --git a/relay/channel/openai/helper.go b/relay/channel/openai/helper.go index 1a01d06da6dc..3254a0562314 100644 --- a/relay/channel/openai/helper.go +++ b/relay/channel/openai/helper.go @@ -169,6 +169,36 @@ func HandleFinalResponse(c *gin.Context, info *relaycommon.RelayInfo, lastStream for _, resp := range claudeResponses { _ = helper.ClaudeData(c, *resp) } + + // Fallback: ensure closing events are emitted even if upstream never sent + // finish_reason / usage. Some OpenAI-compatible upstreams (e.g. LiteLLM) + // truncate the stream without proper terminators, and Claude Code hangs + // indefinitely without message_stop. + if !info.ClaudeConvertInfo.Done { + stopReason := service.StopReasonOpenAI2Claude(info.FinishReason) + if stopReason == "" { + stopReason = "end_turn" + } + if usage != nil { + _ = helper.ClaudeData(c, dto.ClaudeResponse{ + Type: "message_delta", + Usage: service.BuildClaudeUsageFromOpenAIUsage(usage), + Delta: &dto.ClaudeMediaMessage{ + StopReason: common.GetPointer[string](stopReason), + }, + }) + } else { + _ = helper.ClaudeData(c, dto.ClaudeResponse{ + Type: "message_delta", + Delta: &dto.ClaudeMediaMessage{ + StopReason: common.GetPointer[string](stopReason), + }, + }) + } + _ = helper.ClaudeData(c, dto.ClaudeResponse{ + Type: "message_stop", + }) + } info.ClaudeConvertInfo.Done = true case types.RelayFormatGemini: diff --git a/service/convert.go b/service/convert.go index 95acf835ee46..4cd6e7e53986 100644 --- a/service/convert.go +++ b/service/convert.go @@ -247,6 +247,13 @@ func buildClaudeUsageFromOpenAIUsage(oaiUsage *dto.Usage) *dto.ClaudeUsage { return usage } +// BuildClaudeUsageFromOpenAIUsage is the exported variant of +// buildClaudeUsageFromOpenAIUsage for cross-package callers (e.g. fallback +// closing events in relay/channel/openai/helper.go). +func BuildClaudeUsageFromOpenAIUsage(oaiUsage *dto.Usage) *dto.ClaudeUsage { + return buildClaudeUsageFromOpenAIUsage(oaiUsage) +} + func NormalizeCacheCreationSplit(totalTokens int, tokens5m int, tokens1h int) (int, int) { remainder := lo.Max([]int{totalTokens - tokens5m - tokens1h, 0}) return tokens5m + remainder, tokens1h @@ -468,10 +475,36 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon oaiUsage := openAIResponse.Usage if oaiUsage == nil { oaiUsage = info.ClaudeConvertInfo.Usage - // Some upstreams emit finish_reason first, then send a final usage-only chunk. - // Defer closing until usage is available so the final message_delta carries it. - return claudeResponses } + // Emit closing events immediately. Some OpenAI-compatible upstreams + // (e.g. LiteLLM) send finish_reason without usage and never follow up + // with a usage-only chunk; deferring would leave Claude Code hanging. + stopOpenBlocks() + stopReason := stopReasonOpenAI2Claude(info.FinishReason) + if stopReason == "" { + stopReason = "end_turn" + } + if oaiUsage != nil { + claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ + Type: "message_delta", + Usage: buildClaudeUsageFromOpenAIUsage(oaiUsage), + Delta: &dto.ClaudeMediaMessage{ + StopReason: common.GetPointer[string](stopReason), + }, + }) + } else { + claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ + Type: "message_delta", + Delta: &dto.ClaudeMediaMessage{ + StopReason: common.GetPointer[string](stopReason), + }, + }) + } + claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ + Type: "message_stop", + }) + info.ClaudeConvertInfo.Done = true + return claudeResponses } var claudeResponse dto.ClaudeResponse @@ -647,6 +680,12 @@ func stopReasonOpenAI2Claude(reason string) string { return reasonmap.OpenAIFinishReasonToClaudeStopReason(reason) } +// StopReasonOpenAI2Claude is the exported variant for cross-package callers +// (e.g. fallback closing events in relay/channel/openai/helper.go). +func StopReasonOpenAI2Claude(reason string) string { + return stopReasonOpenAI2Claude(reason) +} + func toJSONString(v interface{}) string { b, err := json.Marshal(v) if err != nil { From b0707ce04046d0662080b6bc70f1c06f6a371f15 Mon Sep 17 00:00:00 2001 From: liyang Date: Wed, 24 Jun 2026 22:19:34 +0800 Subject: [PATCH 2/2] fix(claude-stream): address CodeRabbit review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two issues raised by CodeRabbit review on #5345: 1. **fast-path missed terminal events when usage is absent** `service/convert.go` SendResponseCount==1 branch previously only emitted message_delta when usage was available, so a stream that started and finished in a single chunk with finish_reason but no usage would also hang Claude clients. Now emits message_delta (with or without usage) and message_stop in both paths, matching the multi-chunk doneChunk branch. 2. **fallback path didn't close open content blocks** `relay/channel/openai/helper.go` HandleFinalResponse fallback was sending message_delta / message_stop without first closing any open content_block, breaking the "content_block_start … content_block_stop" pairing required by Claude protocol. Added a new exported helper `service.GenerateClaudeStopBlocksForOpenInfo` that reuses the existing block-tracking state to emit the correct content_block_stop events. The fallback now closes open blocks before sending terminal events, keeping the stream sequence valid. Both fixes maintain the same protocol contract: any valid OpenAI stream input produces a fully spec-compliant Claude SSE output. Co-Authored-By: Claude Haiku 4.5 --- relay/channel/openai/helper.go | 7 ++++++ service/convert.go | 40 +++++++++++++++++++++++++++++++++- 2 files changed, 46 insertions(+), 1 deletion(-) diff --git a/relay/channel/openai/helper.go b/relay/channel/openai/helper.go index 3254a0562314..739263ae7a29 100644 --- a/relay/channel/openai/helper.go +++ b/relay/channel/openai/helper.go @@ -175,6 +175,13 @@ func HandleFinalResponse(c *gin.Context, info *relaycommon.RelayInfo, lastStream // truncate the stream without proper terminators, and Claude Code hangs // indefinitely without message_stop. if !info.ClaudeConvertInfo.Done { + // Close any open content blocks first so the event sequence stays + // valid (content_block_start … content_block_stop pairs). + for _, stopBlock := range service.GenerateClaudeStopBlocksForOpenInfo(info) { + _ = helper.ClaudeData(c, *stopBlock) + } + info.ClaudeConvertInfo.LastMessagesType = relaycommon.LastMessageTypeNone + stopReason := service.StopReasonOpenAI2Claude(info.FinishReason) if stopReason == "" { stopReason = "end_turn" diff --git a/service/convert.go b/service/convert.go index 4cd6e7e53986..e00cf7c0b9bd 100644 --- a/service/convert.go +++ b/service/convert.go @@ -223,6 +223,29 @@ func generateStopBlock(index int) *dto.ClaudeResponse { } } +// GenerateClaudeStopBlocksForOpenInfo returns the content_block_stop events +// required to close any open Claude content blocks tracked in +// info.ClaudeConvertInfo. Used by fallback paths that need to emit a valid +// terminal event sequence without re-implementing the block-tracking logic. +// +// Returns an empty slice if no block is currently open. +func GenerateClaudeStopBlocksForOpenInfo(info *relaycommon.RelayInfo) []*dto.ClaudeResponse { + if info == nil { + return nil + } + var responses []*dto.ClaudeResponse + switch info.ClaudeConvertInfo.LastMessagesType { + case relaycommon.LastMessageTypeText, relaycommon.LastMessageTypeThinking: + responses = append(responses, generateStopBlock(info.ClaudeConvertInfo.Index)) + case relaycommon.LastMessageTypeTools: + base := info.ClaudeConvertInfo.ToolCallBaseIndex + for offset := 0; offset <= info.ClaudeConvertInfo.ToolCallMaxIndexOffset; offset++ { + responses = append(responses, generateStopBlock(base+offset)) + } + } + return responses +} + func buildClaudeUsageFromOpenAIUsage(oaiUsage *dto.Usage) *dto.ClaudeUsage { if oaiUsage == nil { return nil @@ -425,12 +448,27 @@ func StreamResponseOpenAI2Claude(openAIResponse *dto.ChatCompletionsStreamRespon if oaiUsage == nil { oaiUsage = info.ClaudeConvertInfo.Usage } + // Always emit message_delta + message_stop, even when usage is + // missing. Some OpenAI-compatible upstreams (e.g. LiteLLM) send + // finish_reason without usage; skipping the terminal events + // would leave Claude clients hanging. + stopReason := stopReasonOpenAI2Claude(info.FinishReason) + if stopReason == "" { + stopReason = "end_turn" + } if oaiUsage != nil { claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ Type: "message_delta", Usage: buildClaudeUsageFromOpenAIUsage(oaiUsage), Delta: &dto.ClaudeMediaMessage{ - StopReason: common.GetPointer[string](stopReasonOpenAI2Claude(info.FinishReason)), + StopReason: common.GetPointer[string](stopReason), + }, + }) + } else { + claudeResponses = append(claudeResponses, &dto.ClaudeResponse{ + Type: "message_delta", + Delta: &dto.ClaudeMediaMessage{ + StopReason: common.GetPointer[string](stopReason), }, }) }