sub2api/backend/internal/pkg/apicompat/responses_stream_event_wire.go
visa2 f10bca8155 refactor(apicompat): redesign the Codex Responses ↔ Chat Completions bridge
Codex CLI speaks the OpenAI Responses protocol (streaming, store:false), while
many upstreams (e.g. DeepSeek in thinking mode) only expose Chat Completions.
The bridge that translates between the two had grown field by field and leaned
on Go's serialization defaults, which both the Responses client (Codex) and the
Chat upstream reject in ways the official OpenAI endpoints tolerate.

Problems this fixes (all observed running Codex CLI against a DeepSeek upstream):
  - Streaming reasoning was never shown in the Codex TUI (the answer appeared
    with no visible thinking): reasoning deltas were emitted before the reasoning
    item was opened, so the strict client discarded them.
  - A tool-using turn could wedge the session into a "no response" state: the
    function_call stream was never closed (no function_call_arguments.done /
    output_item.done), so Codex never saw the tool call complete.
  - Parallel tool calls were rejected upstream (400/502): each function_call
    became its own assistant message, producing consecutive assistant messages
    with mismatched tool replies.
  - A tool turn was rejected with "reasoning_content in the thinking mode must be
    passed back": the reasoning that produced the tool call was dropped instead
    of being returned on the assistant message.
  - Items with no Chat equivalent (web_search_call, ...) and Codex's
    command-approval notice landed between an assistant tool_calls message and
    its tool reply, triggering "An assistant message with 'tool_calls' must be
    followed by tool messages responding to each 'tool_call_id'".
  - Interrupt/reconnect left an unanswered or dangling tool_call in the history,
    triggering the same 400.

The shared root cause is reliance on serialization defaults — omitempty dropping
protocol-required zero values, and unrecognized item types falling through a
generic path — rather than deliberately reproducing the target protocol. The
bridge is reworked into two explicit layers.

Request direction (Responses input -> Chat messages): a parse -> build ->
normalize pipeline.
  - reasoning_content is carried back on the assistant message that produced a
    tool call (DeepSeek thinking mode requires it to continue the same thought)
  - consecutive function_call items (parallel tool calls) are merged into a
    single assistant message's tool_calls array
  - item types with no Chat equivalent are skipped instead of leaking through a
    generic path
  - normalizeChatMessages is the single invariant gate: it guarantees every
    assistant tool_calls message is immediately followed by one tool reply per
    tool_call_id — reordering any intervening message (such as a command-approval
    notice) to after the replies, dropping unanswered tool_calls and orphan tool
    replies, and preserving bare passthrough tool messages.

Response direction (Chat SSE -> Responses SSE): ResponsesStreamEvent.MarshalJSON
constructs each streamed event explicitly so protocol-required fields are always
present (output_index/content_index/summary_index at 0, message content:[],
reasoning summary:[], function_call call_id/name/arguments, output_text part
text/annotations/logprobs). This is a single source of truth that removes any
post-hoc JSON patching. Reasoning is emitted as its own output item, opened
before its deltas, and tool calls are fully closed
(function_call_arguments.done + output_item.done with complete arguments).

Tests cover request-direction message invariants against golden Codex request
shapes (parallel calls, unknown items, intervening messages, partial/dangling
calls), per-event wire completeness, and streaming lifecycle ordering.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-05-31 16:14:58 +08:00

200 lines
5.9 KiB
Go

package apicompat
import "encoding/json"
// MarshalJSON renders a ResponsesStreamEvent into its wire form.
//
// The OpenAI Responses streaming protocol requires several fields to be present
// even when they hold a zero value: output_index/content_index/summary_index are
// meaningful at 0, a function_call item must always carry call_id/name/arguments
// (arguments may be ""), a message item must carry content:[] and an output_text
// part must carry text/annotations/logprobs. Go's `omitempty` drops exactly those
// zero values, and strict clients (Codex CLI) reject items/deltas whose required
// fields are missing.
//
// Rather than marshalling with omitempty and patching the JSON afterwards, every
// streamed event type is constructed explicitly here — the Go analogue of the
// reference gateways' (cc-switch, CCX) per-event object construction. This is the
// single source of truth for Responses SSE field presence and applies uniformly
// to every emitter (Chat→Responses bridge and Anthropic→Responses converter).
//
// Event types not listed fall back to the default struct marshalling, which
// bounds the blast radius of this method to the streamed item/part/text/tool
// events.
func (e ResponsesStreamEvent) MarshalJSON() ([]byte, error) {
switch e.Type {
case "response.output_text.delta", "response.output_text.done":
m := e.wireBase()
e.putItemID(m)
m["output_index"] = e.OutputIndex
m["content_index"] = e.ContentIndex
if e.Type == "response.output_text.done" {
m["text"] = e.Text
} else {
m["delta"] = e.Delta
}
return json.Marshal(m)
case "response.content_part.added", "response.content_part.done":
m := e.wireBase()
e.putItemID(m)
m["output_index"] = e.OutputIndex
m["content_index"] = e.ContentIndex
m["part"] = outputTextPartWire(e.Part)
return json.Marshal(m)
case "response.reasoning_summary_text.delta", "response.reasoning_summary_text.done":
m := e.wireBase()
e.putItemID(m)
m["output_index"] = e.OutputIndex
m["summary_index"] = e.SummaryIndex
if e.Type == "response.reasoning_summary_text.done" {
m["text"] = e.Text
} else {
m["delta"] = e.Delta
}
return json.Marshal(m)
case "response.reasoning_summary_part.added", "response.reasoning_summary_part.done":
m := e.wireBase()
e.putItemID(m)
m["output_index"] = e.OutputIndex
m["summary_index"] = e.SummaryIndex
m["part"] = summaryTextPartWire(e.Part)
return json.Marshal(m)
case "response.output_item.added", "response.output_item.done":
m := e.wireBase()
m["output_index"] = e.OutputIndex
m["item"] = responsesItemWire(e.Item)
return json.Marshal(m)
case "response.function_call_arguments.delta", "response.function_call_arguments.done":
m := e.wireBase()
e.putItemID(m)
m["output_index"] = e.OutputIndex
if e.CallID != "" {
m["call_id"] = e.CallID
}
if e.Name != "" {
m["name"] = e.Name
}
if e.Type == "response.function_call_arguments.done" {
m["arguments"] = e.Arguments
} else {
m["delta"] = e.Delta
}
return json.Marshal(m)
default:
// response.created / completed / done / failed / incomplete and any
// event type not shaped above keep the default struct marshalling.
type alias ResponsesStreamEvent
return json.Marshal(alias(e))
}
}
func (e ResponsesStreamEvent) wireBase() map[string]any {
m := map[string]any{
"type": e.Type,
"sequence_number": e.SequenceNumber,
}
return m
}
func (e ResponsesStreamEvent) putItemID(m map[string]any) {
if e.ItemID != "" {
m["item_id"] = e.ItemID
}
}
// outputTextPartWire renders a content part for a message's output_text, always
// carrying text/annotations/logprobs (matching cc-switch's push_text_delta).
func outputTextPartWire(part *ResponsesContentPart) map[string]any {
text := ""
if part != nil {
text = part.Text
}
return map[string]any{
"type": "output_text",
"text": text,
"annotations": []any{},
"logprobs": []any{},
}
}
// summaryTextPartWire renders a reasoning summary part.
func summaryTextPartWire(part *ResponsesContentPart) map[string]any {
text := ""
if part != nil {
text = part.Text
}
return map[string]any{
"type": "summary_text",
"text": text,
}
}
// responsesItemWire renders an output_item with every field the item's type
// requires to be present, including the empty arrays/strings that omitempty
// would otherwise drop. Mirrors cc-switch's response_function_call_item and the
// message/reasoning item shapes codex expects.
func responsesItemWire(item *ResponsesOutput) map[string]any {
if item == nil {
return map[string]any{}
}
m := map[string]any{
"type": item.Type,
"id": item.ID,
}
if item.Status != "" {
m["status"] = item.Status
}
switch item.Type {
case "message":
role := item.Role
if role == "" {
role = "assistant"
}
m["role"] = role
m["content"] = messageContentWire(item.Content)
case "reasoning":
m["summary"] = reasoningSummaryWire(item.Summary)
if item.EncryptedContent != "" {
m["encrypted_content"] = item.EncryptedContent
}
case "function_call":
m["call_id"] = item.CallID
m["name"] = item.Name
m["arguments"] = item.Arguments
}
return m
}
// messageContentWire renders a message item's content array; always an array
// (never null), with each output_text part carrying its text.
func messageContentWire(parts []ResponsesContentPart) []map[string]any {
out := make([]map[string]any, 0, len(parts))
for _, p := range parts {
typ := p.Type
if typ == "" {
typ = "output_text"
}
out = append(out, map[string]any{"type": typ, "text": p.Text})
}
return out
}
// reasoningSummaryWire renders a reasoning item's summary array; always an array.
func reasoningSummaryWire(summary []ResponsesSummary) []map[string]any {
out := make([]map[string]any, 0, len(summary))
for _, s := range summary {
typ := s.Type
if typ == "" {
typ = "summary_text"
}
out = append(out, map[string]any{"type": typ, "text": s.Text})
}
return out
}