fix: harden codex responses stream conversion

internal/converter/coverage_misc_sse_test.go

@@ -42,3 +42,26 @@ func TestSSE_FormatStringData(t *testing.T) {
 		t.Fatalf("expected string data")
 	}
 }
+
+func TestSSE_FormatMultilineDataPrefixesEveryLine(t *testing.T) {
+	out := FormatSSE("response.completed", []byte("{\n  \"type\": \"response.completed\"\n}"))
+	outStr := string(out)
+	for _, want := range []string{
+		"event: response.completed",
+		"data: {",
+		"data:   \"type\": \"response.completed\"",
+		"data: }",
+	} {
+		if !strings.Contains(outStr, want) {
+			t.Fatalf("expected %q in %q", want, outStr)
+		}
+	}
+
+	events, remaining := ParseSSE(outStr)
+	if remaining != "" || len(events) != 1 {
+		t.Fatalf("expected one complete parsed event, remaining=%q events=%d", remaining, len(events))
+	}
+	if events[0].Event != "response.completed" || !strings.Contains(string(events[0].Data), "response.completed") {
+		t.Fatalf("unexpected parsed event: %+v", events[0])
+	}
+}

internal/converter/coverage_openai_stream_test.go

@@ -180,6 +180,219 @@ func TestOpenAIToCodexRequestAndStream(t *testing.T) {
 	if !strings.Contains(string(streamOut), "response.completed") {
 		t.Fatalf("missing completed")
 	}
+	if !strings.Contains(string(streamOut), "data: [DONE]") {
+		t.Fatalf("missing terminal done sentinel")
+	}
+}
+
+func TestOpenAIToCodexStreamForwardsDoneWhenSplitFromCompletion(t *testing.T) {
+	state := NewTransformState()
+	respConv := &openaiToCodexResponse{}
+	chunk := OpenAIStreamChunk{ID: "chat_1", Model: "gpt", Choices: []OpenAIChoice{{
+		Delta:        &OpenAIMessage{Content: "hi"},
+		FinishReason: "stop",
+	}}}
+	chunkBody, _ := json.Marshal(chunk)
+	firstOut, err := respConv.TransformChunk(FormatSSE("", json.RawMessage(chunkBody)), state)
+	if err != nil {
+		t.Fatalf("TransformChunk first: %v", err)
+	}
+	if !strings.Contains(string(firstOut), "response.completed") {
+		t.Fatalf("missing completed before done: %s", string(firstOut))
+	}
+	doneOut, err := respConv.TransformChunk(FormatDone(), state)
+	if err != nil {
+		t.Fatalf("TransformChunk done: %v", err)
+	}
+	if string(doneOut) != string(FormatDone()) {
+		t.Fatalf("expected terminal done sentinel, got: %q", string(doneOut))
+	}
+	duplicateOut, err := respConv.TransformChunk(FormatDone(), state)
+	if err != nil {
+		t.Fatalf("TransformChunk duplicate done: %v", err)
+	}
+	if len(duplicateOut) != 0 {
+		t.Fatalf("expected duplicate done to be suppressed, got: %q", string(duplicateOut))
+	}
+}
+
+func TestOpenAIToCodexStreamForwardsNativeResponsesEvents(t *testing.T) {
+	state := NewTransformState()
+	respConv := &openaiToCodexResponse{}
+
+	created := map[string]interface{}{
+		"type": "response.created",
+		"response": map[string]interface{}{
+			"id":         "resp_native",
+			"object":     "response",
+			"created_at": int64(1),
+			"status":     "in_progress",
+		},
+	}
+	delta := map[string]interface{}{
+		"type":          "response.output_text.delta",
+		"item_id":       "msg_resp_native_0",
+		"output_index":  0,
+		"content_index": 0,
+		"delta":         "hi",
+	}
+	completed := map[string]interface{}{
+		"type": "response.completed",
+		"response": map[string]interface{}{
+			"id":     "resp_native",
+			"object": "response",
+			"status": "completed",
+		},
+	}
+
+	stream := append(FormatSSE("", created), FormatSSE("", delta)...)
+	stream = append(stream, FormatSSE("", completed)...)
+	stream = append(stream, FormatDone()...)
+
+	out, err := respConv.TransformChunk(stream, state)
+	if err != nil {
+		t.Fatalf("TransformChunk: %v", err)
+	}
+	outStr := string(out)
+	for _, want := range []string{"event: response.created", "event: response.output_text.delta", "event: response.completed", "data: [DONE]"} {
+		if !strings.Contains(outStr, want) {
+			t.Fatalf("missing %s in forwarded responses stream: %s", want, outStr)
+		}
+	}
+	if strings.Count(outStr, "response.completed") != 2 {
+		t.Fatalf("expected one completed event/data pair, got: %s", outStr)
+	}
+}
+
+func TestOpenAIToCodexStreamNormalizesNativeResponsesEventName(t *testing.T) {
+	state := NewTransformState()
+	respConv := &openaiToCodexResponse{}
+
+	completed := map[string]interface{}{
+		"type": "response.completed",
+		"response": map[string]interface{}{
+			"id":     "resp_native",
+			"object": "response",
+			"status": "completed",
+		},
+	}
+
+	out, err := respConv.TransformChunk(FormatSSE("unexpected.event", completed), state)
+	if err != nil {
+		t.Fatalf("TransformChunk: %v", err)
+	}
+	outStr := string(out)
+	if !strings.Contains(outStr, "event: response.completed") {
+		t.Fatalf("expected event name normalized from data.type, got: %s", outStr)
+	}
+	if strings.Contains(outStr, "event: unexpected.event") {
+		t.Fatalf("unexpected mismatched event name preserved: %s", outStr)
+	}
+}
+
+func TestOpenAIToCodexStreamForwardsErrorEvents(t *testing.T) {
+	state := NewTransformState()
+	respConv := &openaiToCodexResponse{}
+
+	errorEvent := map[string]interface{}{
+		"error": map[string]interface{}{
+			"message": "boom",
+		},
+	}
+	typedError := map[string]interface{}{
+		"type":    "error",
+		"message": "typed boom",
+	}
+
+	stream := append(FormatSSE("error", errorEvent), FormatSSE("", typedError)...)
+	out, err := respConv.TransformChunk(stream, state)
+	if err != nil {
+		t.Fatalf("TransformChunk: %v", err)
+	}
+	outStr := string(out)
+	if strings.Count(outStr, "event: error") != 2 {
+		t.Fatalf("expected both error events forwarded, got: %s", outStr)
+	}
+	if !strings.Contains(outStr, "boom") || !strings.Contains(outStr, "typed boom") {
+		t.Fatalf("expected error payloads preserved, got: %s", outStr)
+	}
+}
+
+func TestOpenAIToCodexStreamForwardsEventOnlyNativeResponsesEvent(t *testing.T) {
+	state := NewTransformState()
+	respConv := &openaiToCodexResponse{}
+
+	completed := map[string]interface{}{
+		"response": map[string]interface{}{
+			"id":     "resp_event_only",
+			"object": "response",
+			"status": "completed",
+		},
+	}
+
+	out, err := respConv.TransformChunk(FormatSSE("response.completed", completed), state)
+	if err != nil {
+		t.Fatalf("TransformChunk: %v", err)
+	}
+	outStr := string(out)
+	if !strings.Contains(outStr, "event: response.completed") || !strings.Contains(outStr, `"type":"response.completed"`) {
+		t.Fatalf("expected event-only response.completed to be forwarded with data.type injected, got: %s", outStr)
+	}
+}
+
+func TestOpenAIToCodexStreamErrorEventWinsOverResponseType(t *testing.T) {
+	state := NewTransformState()
+	respConv := &openaiToCodexResponse{}
+
+	errorEvent := map[string]interface{}{
+		"type": "response.completed",
+		"error": map[string]interface{}{
+			"message": "boom",
+		},
+	}
+
+	out, err := respConv.TransformChunk(FormatSSE("error", errorEvent), state)
+	if err != nil {
+		t.Fatalf("TransformChunk: %v", err)
+	}
+	outStr := string(out)
+	if !strings.Contains(outStr, "event: error") || !strings.Contains(outStr, `"type":"error"`) {
+		t.Fatalf("expected error event to keep error semantics, got: %s", outStr)
+	}
+	if strings.Contains(outStr, `"type":"response.completed"`) {
+		t.Fatalf("error event leaked response.completed type: %s", outStr)
+	}
+}
+
+func TestOpenAIToCodexStreamSynthesizesCompletedOnDoneWithoutFinishReason(t *testing.T) {
+	state := NewTransformState()
+	respConv := &openaiToCodexResponse{}
+
+	chunk := OpenAIStreamChunk{ID: "chat_done_only", Model: "gpt", Choices: []OpenAIChoice{{
+		Delta: &OpenAIMessage{Content: "hi"},
+	}}}
+	chunkBody, _ := json.Marshal(chunk)
+	stream := append(FormatSSE("", json.RawMessage(chunkBody)), FormatDone()...)
+
+	out, err := respConv.TransformChunk(stream, state)
+	if err != nil {
+		t.Fatalf("TransformChunk: %v", err)
+	}
+	outStr := string(out)
+	if !strings.Contains(outStr, "response.output_text.done") || !strings.Contains(outStr, "response.completed") {
+		t.Fatalf("expected synthetic completion before done, got: %s", outStr)
+	}
+	if strings.Index(outStr, "response.completed") > strings.Index(outStr, "data: [DONE]") {
+		t.Fatalf("response.completed should be emitted before [DONE], got: %s", outStr)
+	}
+
+	duplicateOut, err := respConv.TransformChunk(FormatDone(), state)
+	if err != nil {
+		t.Fatalf("TransformChunk duplicate done: %v", err)
+	}
+	if len(duplicateOut) != 0 {
+		t.Fatalf("expected duplicate done to be suppressed, got: %q", string(duplicateOut))
+	}
 }
 
 func TestClaudeToOpenAIStreamToolCalls(t *testing.T) {