fix(translator/codex): keep single thinking block per reasoning item in streaming translator (#4581)

Keep a single Claude thinking block open across multiple reasoning summary parts for one Codex reasoning item, and finalize it only when output_item.done delivers the item's final encrypted_content. Previously, each summary part closed the preceding thinking block with the pre-content encrypted_content snapshot captured at output_item.added. This emitted N thinking blocks with placeholder signatures for one reasoning item, causing client replay to append redundant placeholder reasoning items into conversation history. Add regression tests verifying that multi-part summary reasoning items produce exactly one thinking block signed with the final encrypted_content.
2026-09-07 00:24:17 +08:00 · 2026-07-26 12:21:37 +08:00
parent f6c32ec3ff
commit 94cf4674d7
2 changed files with 159 additions and 64 deletions
--- 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
 }
--- 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, &param)...)
 	}

-	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, &param)...)
+	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")
 	}
 }