diff --git a/internal/translator/codex/claude/codex_claude_response.go b/internal/translator/codex/claude/codex_claude_response.go index b60e7c3ff..69d495832 100644 --- a/internal/translator/codex/claude/codex_claude_response.go +++ b/internal/translator/codex/claude/codex_claude_response.go @@ -21,6 +21,10 @@ var ( dataTag = []byte("data:") ) +// codexThinkingSummaryPartSeparator joins consecutive reasoning summary parts inside +// the single thinking block that represents one Codex reasoning item. +const codexThinkingSummaryPartSeparator = "\n\n" + // ConvertCodexResponseToClaudeParams holds parameters for response conversion. type ConvertCodexResponseToClaudeParams struct { HasEmittedToolUse bool @@ -32,7 +36,6 @@ type ConvertCodexResponseToClaudeParams struct { HasTextDelta bool TextBlockOpen bool ThinkingBlockOpen bool - ThinkingStopPending bool ThinkingSignature string ThinkingSummarySeen bool WebSearchToolUseIDs map[string]struct{} @@ -80,12 +83,6 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa output := make([]byte, 0, 512) rootResult := gjson.ParseBytes(rawJSON) params := (*param).(*ConvertCodexResponseToClaudeParams) - if params.ThinkingBlockOpen && params.ThinkingStopPending { - switch rootResult.Get("type").String() { - case "response.content_part.added", "response.completed", "response.incomplete": - output = append(output, finalizeCodexThinkingBlock(params)...) - } - } typeResult := rootResult.Get("type") typeStr := typeResult.String() @@ -101,20 +98,24 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa output = translatorcommon.AppendSSEEventBytes(output, "message_start", template, 2) case "response.reasoning_summary_part.added": - if params.ThinkingBlockOpen && params.ThinkingStopPending { - output = append(output, finalizeCodexThinkingBlock(params)...) + // Codex splits a single reasoning item into several summary parts, but only + // output_item.done carries that item's final encrypted_content. Keep one + // thinking block open for the whole item and separate the parts with a blank + // line, so the only signature ever emitted is the final one. + if params.ThinkingBlockOpen { + output = append(output, appendCodexThinkingDelta(params, codexThinkingSummaryPartSeparator)...) + } else { + output = append(output, startCodexThinkingBlock(params)...) } params.ThinkingSummarySeen = true - output = append(output, startCodexThinkingBlock(params)...) 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) + output = append(output, startCodexThinkingBlock(params)...) + output = append(output, appendCodexThinkingDelta(params, rootResult.Get("delta").String())...) case "response.reasoning_summary_part.done": - params.ThinkingStopPending = true + // Intentionally does not close the thinking block: it stays open until + // output_item.done delivers the reasoning item's final encrypted_content. case "response.content_part.added": + output = append(output, finalizeCodexThinkingBlock(params)...) if rootResult.Get("part.type").String() == "output_text" { output = append(output, startCodexTextBlock(params)...) } @@ -177,7 +178,12 @@ func ConvertCodexResponseToClaude(_ context.Context, _ string, originalRequestRa params.FunctionCallBlockCallID = callID params.FunctionCallBlockIndex = blockIndex case "reasoning": + // A previous reasoning item that never reported output_item.done must not + // leak its still-open block into this one. + output = append(output, finalizeCodexThinkingBlock(params)...) params.ThinkingSummarySeen = false + // Kept only as a fallback for streams whose output_item.done omits + // encrypted_content; it is a pre-content snapshot, never the final value. params.ThinkingSignature = itemResult.Get("encrypted_content").String() case "web_search_call": // Defer server_tool_use until output_item.done carries action/query. @@ -859,11 +865,23 @@ func startCodexThinkingBlock(params *ConvertCodexResponseToClaudeParams) []byte template := []byte(`{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}`) template, _ = sjson.SetBytes(template, "index", params.BlockIndex) params.ThinkingBlockOpen = true - params.ThinkingStopPending = false return translatorcommon.AppendSSEEventBytes(nil, "content_block_start", template, 2) } +// appendCodexThinkingDelta emits a thinking_delta for the currently open thinking block. +func appendCodexThinkingDelta(params *ConvertCodexResponseToClaudeParams, text string) []byte { + if text == "" { + return nil + } + + 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", text) + + return translatorcommon.AppendSSEEventBytes(nil, "content_block_delta", template, 2) +} + func finalizeCodexSignatureOnlyThinkingBlock(params *ConvertCodexResponseToClaudeParams) []byte { if params.ThinkingSignature == "" { return nil @@ -893,7 +911,6 @@ func finalizeCodexThinkingBlock(params *ConvertCodexResponseToClaudeParams) []by params.BlockIndex++ params.ThinkingBlockOpen = false - params.ThinkingStopPending = false return output } diff --git a/internal/translator/codex/claude/codex_claude_response_test.go b/internal/translator/codex/claude/codex_claude_response_test.go index adae51487..bc59cec34 100644 --- a/internal/translator/codex/claude/codex_claude_response_test.go +++ b/internal/translator/codex/claude/codex_claude_response_test.go @@ -171,54 +171,83 @@ func TestConvertCodexResponseToClaude_StreamThinkingWithoutReasoningItemStillInc } } -func TestConvertCodexResponseToClaude_StreamThinkingFinalizesPendingBlockBeforeNextSummaryPart(t *testing.T) { +// codexThinkingStreamDigest collects the thinking-related events produced by a Codex +// stream so tests can assert block/signature counts and the reassembled thinking text. +type codexThinkingStreamDigest struct { + Starts int + Stops int + Signatures []string + Thinking string + Raw string +} + +func digestCodexThinkingStream(t *testing.T, chunks [][]byte) codexThinkingStreamDigest { + t.Helper() + ctx := context.Background() originalRequest := []byte(`{"messages":[]}`) var param any - chunks := [][]byte{ - []byte("data: {\"type\":\"response.reasoning_summary_part.added\"}"), - []byte("data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"First part\"}"), - []byte("data: {\"type\":\"response.reasoning_summary_part.done\"}"), - []byte("data: {\"type\":\"response.reasoning_summary_part.added\"}"), - } - var outputs [][]byte for _, chunk := range chunks { outputs = append(outputs, ConvertCodexResponseToClaude(ctx, "", originalRequest, nil, chunk, ¶m)...) } - startCount := 0 - stopCount := 0 + var digest codexThinkingStreamDigest + var thinking strings.Builder + var raw strings.Builder for _, out := range outputs { + raw.Write(out) for _, line := range strings.Split(string(out), "\n") { if !strings.HasPrefix(line, "data: ") { continue } data := gjson.Parse(strings.TrimPrefix(line, "data: ")) - if data.Get("type").String() == "content_block_start" && data.Get("content_block.type").String() == "thinking" { - startCount++ - } - if data.Get("type").String() == "content_block_stop" { - stopCount++ + switch data.Get("type").String() { + case "content_block_start": + if data.Get("content_block.type").String() == "thinking" { + digest.Starts++ + } + case "content_block_delta": + switch data.Get("delta.type").String() { + case "thinking_delta": + thinking.WriteString(data.Get("delta.thinking").String()) + case "signature_delta": + digest.Signatures = append(digest.Signatures, data.Get("delta.signature").String()) + } + case "content_block_stop": + digest.Stops++ } } } + digest.Thinking = thinking.String() + digest.Raw = raw.String() - if startCount != 2 { - t.Fatalf("expected 2 thinking block starts, got %d", startCount) + return digest +} + +func TestConvertCodexResponseToClaude_StreamThinkingKeepsSingleBlockAcrossSummaryParts(t *testing.T) { + digest := digestCodexThinkingStream(t, [][]byte{ + []byte("data: {\"type\":\"response.reasoning_summary_part.added\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"First part\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.done\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.added\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"Second part\"}"), + }) + + if digest.Starts != 1 { + t.Fatalf("expected a single thinking block start for one reasoning item, got %d", digest.Starts) } - if stopCount != 1 { - t.Fatalf("expected pending thinking block to be finalized before second start, got %d stops", stopCount) + if digest.Stops != 0 { + t.Fatalf("expected the thinking block to stay open until output_item.done, got %d stops", digest.Stops) + } + if want := "First part\n\nSecond part"; digest.Thinking != want { + t.Fatalf("thinking text = %q, want %q", digest.Thinking, want) } } -func TestConvertCodexResponseToClaude_StreamThinkingRetainsSignatureAcrossMultipartReasoning(t *testing.T) { - ctx := context.Background() - originalRequest := []byte(`{"messages":[]}`) - var param any - - chunks := [][]byte{ +func TestConvertCodexResponseToClaude_StreamThinkingEmitsSingleSignatureAcrossMultipartReasoning(t *testing.T) { + digest := digestCodexThinkingStream(t, [][]byte{ []byte("data: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"reasoning\",\"encrypted_content\":\"enc_sig_multipart\"}}"), []byte("data: {\"type\":\"response.reasoning_summary_part.added\"}"), []byte("data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"First part\"}"), @@ -227,31 +256,80 @@ func TestConvertCodexResponseToClaude_StreamThinkingRetainsSignatureAcrossMultip []byte("data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"Second part\"}"), []byte("data: {\"type\":\"response.reasoning_summary_part.done\"}"), []byte("data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"reasoning\"}}"), - } + }) - var outputs [][]byte - for _, chunk := range chunks { - outputs = append(outputs, ConvertCodexResponseToClaude(ctx, "", originalRequest, nil, chunk, ¶m)...) + if digest.Starts != 1 || digest.Stops != 1 { + t.Fatalf("expected exactly one thinking block, got %d starts and %d stops", digest.Starts, digest.Stops) } - - signatureDeltaCount := 0 - for _, out := range outputs { - for _, line := range strings.Split(string(out), "\n") { - if !strings.HasPrefix(line, "data: ") { - continue - } - data := gjson.Parse(strings.TrimPrefix(line, "data: ")) - if data.Get("type").String() == "content_block_delta" && data.Get("delta.type").String() == "signature_delta" { - signatureDeltaCount++ - if got := data.Get("delta.signature").String(); got != "enc_sig_multipart" { - t.Fatalf("unexpected signature delta: %q", got) - } - } - } + if len(digest.Signatures) != 1 { + t.Fatalf("expected one signature_delta for one reasoning item, got %d: %v", len(digest.Signatures), digest.Signatures) } + // output_item.done omitted encrypted_content here, so the pre-content fallback is expected. + if digest.Signatures[0] != "enc_sig_multipart" { + t.Fatalf("unexpected signature delta: %q", digest.Signatures[0]) + } + if want := "First part\n\nSecond part"; digest.Thinking != want { + t.Fatalf("thinking text = %q, want %q", digest.Thinking, want) + } +} - if signatureDeltaCount != 2 { - t.Fatalf("expected signature_delta for both multipart thinking blocks, got %d", signatureDeltaCount) +// TestConvertCodexResponseToClaude_StreamThinkingNeverEmitsPreContentEncryptedContent guards the +// real-world shape earlier tests missed: output_item.added carries a fixed-size pre-content +// snapshot of encrypted_content that always differs from the final value on output_item.done. +// Emitting that snapshot makes the client replay bogus reasoning items for the rest of the session. +func TestConvertCodexResponseToClaude_StreamThinkingNeverEmitsPreContentEncryptedContent(t *testing.T) { + digest := digestCodexThinkingStream(t, [][]byte{ + []byte("data: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"reasoning\",\"encrypted_content\":\"enc_sig_pre_content_snapshot\"}}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.added\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"Part A\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.done\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.added\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"Part B\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.done\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.added\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"Part C\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.done\"}"), + []byte("data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"reasoning\",\"encrypted_content\":\"enc_sig_final\"}}"), + }) + + if digest.Starts != 1 || digest.Stops != 1 { + t.Fatalf("expected one thinking block for one reasoning item with three summary parts, got %d starts and %d stops", digest.Starts, digest.Stops) + } + if len(digest.Signatures) != 1 || digest.Signatures[0] != "enc_sig_final" { + t.Fatalf("expected exactly one signature_delta carrying the final encrypted_content, got %v", digest.Signatures) + } + if strings.Contains(digest.Raw, "enc_sig_pre_content_snapshot") { + t.Fatal("pre-content encrypted_content snapshot leaked into the Claude stream") + } + if want := "Part A\n\nPart B\n\nPart C"; digest.Thinking != want { + t.Fatalf("thinking text = %q, want %q", digest.Thinking, want) + } +} + +// TestConvertCodexResponseToClaude_StreamThinkingEmitsOneBlockPerReasoningItem checks that two +// consecutive reasoning items stay separate blocks, each signed with its own final value. +func TestConvertCodexResponseToClaude_StreamThinkingEmitsOneBlockPerReasoningItem(t *testing.T) { + digest := digestCodexThinkingStream(t, [][]byte{ + []byte("data: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"reasoning\",\"encrypted_content\":\"enc_pre_1\"}}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.added\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"First item\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.done\"}"), + []byte("data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"reasoning\",\"encrypted_content\":\"enc_final_1\"}}"), + []byte("data: {\"type\":\"response.output_item.added\",\"item\":{\"type\":\"reasoning\",\"encrypted_content\":\"enc_pre_2\"}}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.added\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_text.delta\",\"delta\":\"Second item\"}"), + []byte("data: {\"type\":\"response.reasoning_summary_part.done\"}"), + []byte("data: {\"type\":\"response.output_item.done\",\"item\":{\"type\":\"reasoning\",\"encrypted_content\":\"enc_final_2\"}}"), + }) + + if digest.Starts != 2 || digest.Stops != 2 { + t.Fatalf("expected two thinking blocks for two reasoning items, got %d starts and %d stops", digest.Starts, digest.Stops) + } + if len(digest.Signatures) != 2 || digest.Signatures[0] != "enc_final_1" || digest.Signatures[1] != "enc_final_2" { + t.Fatalf("expected each block signed with its own final encrypted_content, got %v", digest.Signatures) + } + if strings.Contains(digest.Raw, "enc_pre_1") || strings.Contains(digest.Raw, "enc_pre_2") { + t.Fatal("pre-content encrypted_content snapshot leaked into the Claude stream") } }