Merge pull request #3770 from sususu98/fix/codex-claude-stream-error-upstream-dev

fix: translate codex stream errors for claude
This commit is contained in:
sususu98
2026-06-09 11:43:43 +08:00
committed by GitHub
2 changed files with 118 additions and 17 deletions
@@ -77,47 +77,50 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa
typeStr := typeResult.String()
var template []byte
if typeStr == "response.created" {
switch typeStr {
case "error":
output = append(output, codexStreamErrorToClaudeError(rootResult)...)
case "response.created":
template = []byte(`{"type":"message_start","message":{"id":"","type":"message","role":"assistant","model":"claude-opus-4-1-20250805","stop_sequence":null,"usage":{"input_tokens":0,"output_tokens":0},"content":[],"stop_reason":null}}`)
template, _ = sjson.SetBytes(template, "message.model", rootResult.Get("response.model").String())
template, _ = sjson.SetBytes(template, "message.id", rootResult.Get("response.id").String())
output = translatorcommon.AppendSSEEventBytes(output, "message_start", template, 2)
} else if typeStr == "response.reasoning_summary_part.added" {
case "response.reasoning_summary_part.added":
if params.ThinkingBlockOpen && params.ThinkingStopPending {
output = append(output, finalizeCodexThinkingBlock(params)...)
}
params.ThinkingSummarySeen = true
output = append(output, startCodexThinkingBlock(params)...)
} else if typeStr == "response.reasoning_summary_text.delta" {
case "response.reasoning_summary_text.delta":
template = []byte(`{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":""}}`)
template, _ = sjson.SetBytes(template, "index", params.BlockIndex)
template, _ = sjson.SetBytes(template, "delta.thinking", rootResult.Get("delta").String())
output = translatorcommon.AppendSSEEventBytes(output, "content_block_delta", template, 2)
} else if typeStr == "response.reasoning_summary_part.done" {
case "response.reasoning_summary_part.done":
params.ThinkingStopPending = true
} else if typeStr == "response.content_part.added" {
case "response.content_part.added":
template = []byte(`{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`)
template, _ = sjson.SetBytes(template, "index", params.BlockIndex)
params.TextBlockOpen = true
output = translatorcommon.AppendSSEEventBytes(output, "content_block_start", template, 2)
} else if typeStr == "response.output_text.delta" {
case "response.output_text.delta":
params.HasTextDelta = true
template = []byte(`{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":""}}`)
template, _ = sjson.SetBytes(template, "index", params.BlockIndex)
template, _ = sjson.SetBytes(template, "delta.text", rootResult.Get("delta").String())
output = translatorcommon.AppendSSEEventBytes(output, "content_block_delta", template, 2)
} else if typeStr == "response.content_part.done" {
case "response.content_part.done":
template = []byte(`{"type":"content_block_stop","index":0}`)
template, _ = sjson.SetBytes(template, "index", params.BlockIndex)
params.TextBlockOpen = false
params.BlockIndex++
output = translatorcommon.AppendSSEEventBytes(output, "content_block_stop", template, 2)
} else if typeStr == "response.completed" || typeStr == "response.incomplete" {
case "response.completed", "response.incomplete":
template = []byte(`{"type":"message_delta","delta":{"stop_reason":"tool_use","stop_sequence":null},"usage":{"input_tokens":0,"output_tokens":0}}`)
responseData := rootResult.Get("response")
template, _ = sjson.SetBytes(template, "delta.stop_reason", mapCodexStopReasonToClaude(codexStopReason(responseData), params.HasToolCall))
@@ -131,10 +134,11 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa
output = translatorcommon.AppendSSEEventBytes(output, "message_delta", template, 2)
output = translatorcommon.AppendSSEEventBytes(output, "message_stop", []byte(`{"type":"message_stop"}`), 2)
} else if typeStr == "response.output_item.added" {
case "response.output_item.added":
itemResult := rootResult.Get("item")
itemType := itemResult.Get("type").String()
if itemType == "function_call" {
switch itemType {
case "function_call":
output = append(output, finalizeCodexThinkingBlock(params)...)
params.HasToolCall = true
params.HasReceivedArgumentsDelta = false
@@ -156,14 +160,15 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa
template, _ = sjson.SetBytes(template, "index", params.BlockIndex)
output = translatorcommon.AppendSSEEventBytes(output, "content_block_delta", template, 2)
} else if itemType == "reasoning" {
case "reasoning":
params.ThinkingSummarySeen = false
params.ThinkingSignature = itemResult.Get("encrypted_content").String()
}
} else if typeStr == "response.output_item.done" {
case "response.output_item.done":
itemResult := rootResult.Get("item")
itemType := itemResult.Get("type").String()
if itemType == "message" {
switch itemType {
case "message":
if params.HasTextDelta {
return [][]byte{output}
}
@@ -205,13 +210,13 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa
params.BlockIndex++
params.HasTextDelta = true
output = translatorcommon.AppendSSEEventBytes(output, "content_block_stop", template, 2)
} else if itemType == "function_call" {
case "function_call":
template = []byte(`{"type":"content_block_stop","index":0}`)
template, _ = sjson.SetBytes(template, "index", params.BlockIndex)
params.BlockIndex++
output = translatorcommon.AppendSSEEventBytes(output, "content_block_stop", template, 2)
} else if itemType == "reasoning" {
case "reasoning":
if signature := itemResult.Get("encrypted_content").String(); signature != "" {
params.ThinkingSignature = signature
}
@@ -223,14 +228,14 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa
params.ThinkingSignature = ""
params.ThinkingSummarySeen = false
}
} else if typeStr == "response.function_call_arguments.delta" {
case "response.function_call_arguments.delta":
params.HasReceivedArgumentsDelta = true
template = []byte(`{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":""}}`)
template, _ = sjson.SetBytes(template, "index", params.BlockIndex)
template, _ = sjson.SetBytes(template, "delta.partial_json", rootResult.Get("delta").String())
output = translatorcommon.AppendSSEEventBytes(output, "content_block_delta", template, 2)
} else if typeStr == "response.function_call_arguments.done" {
case "response.function_call_arguments.done":
if !params.HasReceivedArgumentsDelta {
if args := rootResult.Get("arguments").String(); args != "" {
template = []byte(`{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":""}}`)
@@ -245,6 +250,38 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa
return [][]byte{output}
}
func codexStreamErrorToClaudeError(rootResult gjson.Result) []byte {
errorResult := rootResult.Get("error")
errType := strings.TrimSpace(errorResult.Get("type").String())
if errType == "" {
errType = strings.TrimSpace(rootResult.Get("error_type").String())
}
if errType == "" {
errType = "api_error"
}
code := strings.TrimSpace(errorResult.Get("code").String())
message := strings.TrimSpace(errorResult.Get("message").String())
if message == "" {
message = strings.TrimSpace(rootResult.Get("message").String())
}
if message == "" {
message = code
}
if message == "" {
message = errType
}
if code == "cyber_policy" || errType == "invalid_request" {
errType = "invalid_request_error"
}
out := []byte(`{"type":"error","error":{"type":"api_error","message":""}}`)
out, _ = sjson.SetBytes(out, "error.type", errType)
out, _ = sjson.SetBytes(out, "error.message", message)
return translatorcommon.AppendSSEEventBytes(nil, "error", out, 2)
}
// ConvertCodexResponseToClaudeNonStream converts a non-streaming Codex response to a non-streaming Claude Code response.
// This function processes the complete Codex response and transforms it into a single Claude Code-compatible
// JSON response. It handles message content, tool calls, reasoning content, and usage metadata, combining all
@@ -68,6 +68,55 @@ func TestConvertCodexResponseToClaude_StreamThinkingIncludesSignature(t *testing
}
}
func TestConvertCodexResponseToClaude_StreamCyberPolicyError(t *testing.T) {
ctx := context.Background()
var param any
outputs := ConvertCodexResponseToClaude(ctx, "", []byte(`{"messages":[]}`), nil, []byte(`data: {"type":"error","error":{"type":"invalid_request","code":"cyber_policy","message":"This content was flagged for possible cybersecurity risk.","param":null},"sequence_number":3}`), &param)
if len(outputs) != 1 {
t.Fatalf("expected one error chunk, got %d: %q", len(outputs), outputs)
}
out := string(outputs[0])
if !strings.Contains(out, "event: error\n") {
t.Fatalf("expected Claude SSE error event, got: %q", out)
}
payload, ok := firstClaudeStreamPayloadForEvent(out, "error")
if !ok {
t.Fatalf("missing error event payload: %q", out)
}
if got := payload.Get("type").String(); got != "error" {
t.Fatalf("type = %q, want error. Payload: %s", got, payload.Raw)
}
if got := payload.Get("error.type").String(); got != "invalid_request_error" {
t.Fatalf("error.type = %q, want invalid_request_error. Payload: %s", got, payload.Raw)
}
if got := payload.Get("error.message").String(); got != "This content was flagged for possible cybersecurity risk." {
t.Fatalf("error.message = %q. Payload: %s", got, payload.Raw)
}
}
func TestConvertCodexResponseToClaude_StreamErrorTypeFallbackMessage(t *testing.T) {
ctx := context.Background()
var param any
outputs := ConvertCodexResponseToClaude(ctx, "", []byte(`{"messages":[]}`), nil, []byte(`data: {"type":"error","error":{},"error_type":"overloaded_error"}`), &param)
if len(outputs) != 1 {
t.Fatalf("expected one error chunk, got %d: %q", len(outputs), outputs)
}
payload, ok := firstClaudeStreamPayloadForEvent(string(outputs[0]), "error")
if !ok {
t.Fatalf("missing error event payload: %q", outputs[0])
}
if got := payload.Get("error.type").String(); got != "overloaded_error" {
t.Fatalf("error.type = %q, want overloaded_error. Payload: %s", got, payload.Raw)
}
if got := payload.Get("error.message").String(); got != "overloaded_error" {
t.Fatalf("error.message = %q, want overloaded_error. Payload: %s", got, payload.Raw)
}
}
func TestConvertCodexResponseToClaude_StreamThinkingWithoutReasoningItemStillIncludesSignatureField(t *testing.T) {
ctx := context.Background()
originalRequest := []byte(`{"messages":[]}`)
@@ -726,3 +775,18 @@ func findClaudeStreamMessageDelta(outputs [][]byte) (gjson.Result, bool) {
}
return gjson.Result{}, false
}
func firstClaudeStreamPayloadForEvent(output, event string) (gjson.Result, bool) {
var currentEvent string
for _, line := range strings.Split(output, "\n") {
if strings.HasPrefix(line, "event: ") {
currentEvent = strings.TrimPrefix(line, "event: ")
continue
}
if currentEvent != event || !strings.HasPrefix(line, "data: ") {
continue
}
return gjson.Parse(strings.TrimPrefix(line, "data: ")), true
}
return gjson.Result{}, false
}