feat(relay):response协议转claude协议

relay/channel/claude/adaptor.go

@@ -31,7 +31,9 @@ func (a *Adaptor) GetRequestURL(info *common.RelayInfo) (string, error) {
 	baseURL := strings.TrimSuffix(info.ChannelMeta.BaseURL, "/")
 
 	switch constant.RelayMode(info.RelayMode) {
-	case constant.RelayModeChatCompletions, constant.RelayModeClaudeMessages:
+	case constant.RelayModeChatCompletions, constant.RelayModeClaudeMessages,
+		constant.RelayModeResponses, constant.RelayModeResponsesCompact:
+		// Responses 入站：请求转 Claude Messages 格式打 /v1/messages，响应转回 Responses 格式
 		return baseURL + "/v1/messages", nil
 	default:
 		return "", fmt.Errorf("unsupported relay mode for Claude: %d", info.RelayMode)
@@ -209,6 +211,12 @@ func (a *Adaptor) DoResponse(ctx context.Context, resp *http.Response, info *com
 	switch clientFormat {
 	case constant.RelayFormatClaude:
 		return a.handleClaudeNativeResponse(ctx, resp, info, writer)
+	case constant.RelayFormatResponses:
+		// Responses 入站（codex 等）：Claude 响应转换为 Responses 格式
+		if info.IsStream {
+			return a.handleStreamToResponses(ctx, resp, info, writer)
+		}
+		return a.handleNonStreamToResponses(ctx, resp, info, writer)
 	case constant.RelayFormatOpenAI:
 		if info.IsStream {
 			return a.handleStreamToOpenAI(ctx, resp, info, writer)

relay/channel/claude/responses_bridge.go

@@ -0,0 +1,499 @@
+package claude
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/gogf/gf/v2/frame/g"
+
+	"github.com/qianfree/team-api/relay/channel/openai"
+	"github.com/qianfree/team-api/relay/common"
+	"github.com/qianfree/team-api/relay/constant"
+	"github.com/qianfree/team-api/relay/dto"
+	"github.com/qianfree/team-api/relay/helper"
+)
+
+// ========== Responses 入站桥接：Claude Messages → OpenAI Responses ==========
+//
+// 请求侧由 ConvertResponsesToClaude 完成（Responses → OpenAI → Claude），
+// 这里做响应侧：把 Claude 上游的响应/SSE 转回 Responses 格式，
+// 事件发射器复用 openai 包（EmitResponsesSSE / BuildResponsesObjectMap），
+// 与 chat→responses 桥接（openai/responses.go）的事件序列保持一致。
+
+// claudeToolCallState Responses 桥接中聚合中的工具调用（Claude tool_use 块）
+type claudeToolCallState struct {
+	id   string
+	name string
+	args strings.Builder
+}
+
+// handleNonStreamToResponses 将 Claude 非流式响应转换为 Responses 格式
+func (a *Adaptor) handleNonStreamToResponses(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) {
+	defer resp.Body.Close()
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, constant.NewUpstreamError(resp.StatusCode, "read response body failed", err).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header))
+	}
+	if resp.StatusCode != http.StatusOK {
+		return nil, constant.NewUpstreamError(resp.StatusCode, string(body), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header))
+	}
+
+	var claudeResp dto.ClaudeResponse
+	if err := json.Unmarshal(body, &claudeResp); err != nil {
+		return nil, constant.NewUpstreamError(resp.StatusCode, "invalid response body", err).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header))
+	}
+
+	// 构建 output：文本块 → message 项，tool_use 块 → function_call 项
+	var textParts []string
+	output := make([]map[string]any, 0)
+	for _, block := range claudeResp.Content {
+		switch block.Type {
+		case "text":
+			if block.Text != nil && *block.Text != "" {
+				textParts = append(textParts, *block.Text)
+			}
+		case "thinking", "redacted_thinking":
+			// 思考内容无 Responses 非流式对应物，跳过（流式侧以 reasoning summary 事件透出）
+		case "tool_use":
+			argsJSON, _ := json.Marshal(block.Input)
+			output = append(output, map[string]any{
+				"type":      "function_call",
+				"id":        block.ID,
+				"call_id":   block.ID,
+				"name":      block.Name,
+				"arguments": string(argsJSON),
+				"status":    "completed",
+			})
+		}
+	}
+	if len(textParts) > 0 {
+		msgItem := map[string]any{
+			"type":   "message",
+			"id":     fmt.Sprintf("msg_%s", claudeResp.ID),
+			"status": "completed",
+			"role":   "assistant",
+			"content": []map[string]any{{
+				"type":        "output_text",
+				"text":        strings.Join(textParts, "\n"),
+				"annotations": []any{},
+			}},
+		}
+		output = append([]map[string]any{msgItem}, output...)
+	}
+
+	modelName := claudeResp.Model
+	if modelName == "" || info.ChannelMeta.IsModelMapped {
+		modelName = info.OriginModelName
+	}
+	respID := fmt.Sprintf("resp_%s", claudeResp.ID)
+	if claudeResp.ID == "" {
+		respID = fmt.Sprintf("resp_%d", time.Now().UnixNano())
+	}
+	createdAt := int(time.Now().Unix())
+	completedAt := createdAt
+
+	// 客户端可见 usage 用 OpenAI 语义（input 含缓存，cached 为子集）
+	var visibleUsage *common.Usage
+	if claudeResp.Usage != nil {
+		promptTotal := claudeResp.Usage.InputTokens +
+			claudeResp.Usage.CacheReadInputTokens +
+			claudeResp.Usage.CacheCreationInputTokens
+		visibleUsage = &common.Usage{
+			PromptTokens:        promptTotal,
+			CompletionTokens:    claudeResp.Usage.OutputTokens,
+			TotalTokens:         promptTotal + claudeResp.Usage.OutputTokens,
+			PromptTokensDetails: claudeUsageToTokenDetails(claudeResp.Usage),
+		}
+	} else {
+		visibleUsage = &common.Usage{}
+	}
+
+	responsesBody, err := json.Marshal(openai.BuildResponsesObjectMap(respID, createdAt, "completed", modelName, output, openai.BuildResponsesUsageMap(visibleUsage), &completedAt, info))
+	if err != nil {
+		return nil, fmt.Errorf("marshal responses body failed: %w", err)
+	}
+	writer.Header().Set("Content-Type", "application/json")
+	writer.WriteHeader(http.StatusOK)
+	_, _ = writer.Write(responsesBody)
+
+	// 计费返回值按 Claude 口径（input 不含缓存，cache 独立列）
+	return buildUsageFromClaude(claudeResp.Usage), nil
+}
+
+// handleStreamToResponses 将 Claude 流式响应转换为 Responses 格式的 SSE
+func (a *Adaptor) handleStreamToResponses(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) {
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		resp.Body.Close()
+		return nil, constant.NewUpstreamError(resp.StatusCode, string(body), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header))
+	}
+	defer resp.Body.Close()
+
+	helper.SetEventStreamHeaders(writer)
+	writer = helper.NewSafeWriter(writer)
+	defer helper.PingTicker(writer, 15*time.Second)()
+
+	scanner := bufio.NewScanner(resp.Body)
+	buf := make([]byte, 0, 64*1024)
+	scanner.Buffer(buf, 10*1024*1024)
+
+	respID := fmt.Sprintf("resp_%d", time.Now().UnixNano())
+	msgID := fmt.Sprintf("msg_%d", time.Now().UnixNano())
+	createdAt := int(time.Now().Unix())
+	modelName := info.OriginModelName
+
+	var usage dto.ClaudeUsage
+	var textBuf strings.Builder
+	sentCreated := false
+	sentTextDone := false
+	sentCompleted := false
+	outputIndex := 0
+	contentIndex := 0
+	toolCalls := make([]*claudeToolCallState, 0) // 有序聚合，completed 的 output 数组按此顺序
+	toolIndexByID := make(map[string]int)        // callID → output_index
+	var currentTool *claudeToolCallState         // 正在接收参数增量的工具调用
+
+	// buildBillingUsage 计费口径（Claude：input 不含缓存），输出缺失按已转发文本估算
+	buildBillingUsage := func() *common.Usage {
+		u := buildUsageFromClaude(&usage)
+		if u.CompletionTokens == 0 && textBuf.Len() > 0 {
+			u.CompletionTokens = helper.EstimateStreamOutputTokens(info, textBuf.Len())
+			u.TotalTokens = u.PromptTokens + u.CompletionTokens
+		}
+		return u
+	}
+
+	// closeTextPart 关闭文本 content part（进入工具调用或流结束时调用）
+	closeTextPart := func() {
+		if sentTextDone {
+			return
+		}
+		finishedText := textBuf.String()
+		openai.EmitResponsesSSE(writer, "response.output_text.done", map[string]any{
+			"type":          "response.output_text.done",
+			"item_id":       msgID,
+			"output_index":  outputIndex,
+			"content_index": contentIndex,
+			"text":          finishedText,
+		})
+		openai.EmitResponsesSSE(writer, "response.content_part.done", map[string]any{
+			"type":          "response.content_part.done",
+			"item_id":       msgID,
+			"output_index":  outputIndex,
+			"content_index": contentIndex,
+			"part": map[string]any{
+				"type":        "output_text",
+				"text":        finishedText,
+				"annotations": []any{},
+			},
+		})
+		openai.EmitResponsesSSE(writer, "response.output_item.done", map[string]any{
+			"type":         "response.output_item.done",
+			"output_index": outputIndex,
+			"item": map[string]any{
+				"type":   "message",
+				"id":     msgID,
+				"status": "completed",
+				"role":   "assistant",
+				"content": []map[string]any{{
+					"type":        "output_text",
+					"text":        finishedText,
+					"annotations": []any{},
+				}},
+			},
+		})
+		sentTextDone = true
+		outputIndex++
+	}
+
+	// finish 发送每个工具调用的收尾事件 + response.completed
+	finish := func() {
+		if sentCompleted {
+			return
+		}
+		closeTextPart()
+
+		// 输出数组：文本消息（closeTextPart 已置 sentTextDone，此处等价于有文本才保留，
+		// 与 chat→responses 桥接的 finalOutput 构建口径一致）
+		finalOutput := make([]map[string]any, 0)
+		if textBuf.Len() > 0 {
+			finalOutput = append(finalOutput, map[string]any{
+				"type":   "message",
+				"id":     msgID,
+				"status": "completed",
+				"role":   "assistant",
+				"content": []map[string]any{{
+					"type":        "output_text",
+					"text":        textBuf.String(),
+					"annotations": []any{},
+				}},
+			})
+		}
+
+		for _, tc := range toolCalls {
+			openai.EmitResponsesSSE(writer, "response.function_call_arguments.done", map[string]any{
+				"type":         "response.function_call_arguments.done",
+				"item_id":      tc.id,
+				"output_index": toolIndexByID[tc.id],
+				"arguments":    tc.args.String(),
+			})
+			openai.EmitResponsesSSE(writer, "response.output_item.done", map[string]any{
+				"type":         "response.output_item.done",
+				"output_index": toolIndexByID[tc.id],
+				"item": map[string]any{
+					"type":      "function_call",
+					"id":        tc.id,
+					"call_id":   tc.id,
+					"name":      tc.name,
+					"arguments": tc.args.String(),
+					"status":    "completed",
+				},
+			})
+			finalOutput = append(finalOutput, map[string]any{
+				"type":      "function_call",
+				"id":        tc.id,
+				"call_id":   tc.id,
+				"name":      tc.name,
+				"arguments": tc.args.String(),
+				"status":    "completed",
+			})
+		}
+
+		// 客户端可见 usage 用 OpenAI 语义（input 含缓存，cached 为子集）
+		promptTotal := usage.InputTokens + usage.CacheReadInputTokens + usage.CacheCreationInputTokens
+		visibleUsage := &common.Usage{
+			PromptTokens:        promptTotal,
+			CompletionTokens:    usage.OutputTokens,
+			TotalTokens:         promptTotal + usage.OutputTokens,
+			PromptTokensDetails: claudeUsageToTokenDetails(&usage),
+		}
+		completedAt := int(time.Now().Unix())
+		openai.EmitResponsesSSE(writer, "response.completed", map[string]any{
+			"type":     "response.completed",
+			"response": openai.BuildResponsesObjectMap(respID, createdAt, "completed", modelName, finalOutput, openai.BuildResponsesUsageMap(visibleUsage), &completedAt, info),
+		})
+		sentCompleted = true
+
+		if info.StreamStatus.GetEndReason() == "" {
+			info.StreamStatus.SetEndReason(common.StreamEndReasonDone, nil)
+		}
+	}
+
+	for scanner.Scan() {
+		select {
+		case <-ctx.Done():
+			info.StreamStatus.SetEndReason(common.StreamEndReasonClientGone, ctx.Err())
+			interruptedUsage := buildUsageFromClaude(&usage)
+			helper.ApplyInterruptedUsageFallback(info, interruptedUsage, textBuf.Len())
+			return interruptedUsage, common.ErrStreamInterrupted
+		default:
+		}
+
+		line := scanner.Text()
+		if line == "" || !strings.HasPrefix(line, "data:") {
+			continue
+		}
+		data, _ := helper.ExtractSSEData(line)
+		if data == "" || data == "[DONE]" {
+			continue
+		}
+		info.SetFirstResponseTime()
+
+		var event dto.ClaudeResponse
+		if err := json.Unmarshal([]byte(data), &event); err != nil {
+			continue
+		}
+
+		switch event.Type {
+		case "message_start":
+			if event.Message != nil {
+				if event.Message.Model != "" && !info.ChannelMeta.IsModelMapped {
+					modelName = event.Message.Model
+				}
+				if event.Message.Usage != nil {
+					usage = *event.Message.Usage
+				}
+			}
+			if sentCreated {
+				continue
+			}
+			openai.EmitResponsesSSE(writer, "response.created", map[string]any{
+				"type":     "response.created",
+				"response": openai.BuildResponsesObjectMap(respID, createdAt, "in_progress", modelName, []any{}, nil, nil, info),
+			})
+			openai.EmitResponsesSSE(writer, "response.output_item.added", map[string]any{
+				"type":         "response.output_item.added",
+				"output_index": outputIndex,
+				"item": map[string]any{
+					"type":    "message",
+					"id":      msgID,
+					"status":  "in_progress",
+					"role":    "assistant",
+					"content": []any{},
+				},
+			})
+			openai.EmitResponsesSSE(writer, "response.content_part.added", map[string]any{
+				"type":          "response.content_part.added",
+				"item_id":       msgID,
+				"output_index":  outputIndex,
+				"content_index": contentIndex,
+				"part": map[string]any{
+					"type":        "output_text",
+					"text":        "",
+					"annotations": []any{},
+				},
+			})
+			sentCreated = true
+
+		case "content_block_start":
+			if event.ContentBlock == nil {
+				continue
+			}
+			switch event.ContentBlock.Type {
+			case "tool_use":
+				// 先关闭文本 content part，再开 function_call 项
+				closeTextPart()
+				tc := &claudeToolCallState{id: event.ContentBlock.ID, name: event.ContentBlock.Name}
+				toolCalls = append(toolCalls, tc)
+				toolIndexByID[tc.id] = outputIndex
+				currentTool = tc
+				openai.EmitResponsesSSE(writer, "response.output_item.added", map[string]any{
+					"type":         "response.output_item.added",
+					"output_index": outputIndex,
+					"item": map[string]any{
+						"type":    "function_call",
+						"id":      tc.id,
+						"call_id": tc.id,
+						"name":    tc.name,
+						"status":  "in_progress",
+					},
+				})
+				outputIndex++
+			case "text", "thinking", "redacted_thinking":
+				// 文本/思考块：文本复用首个 content part，思考以 reasoning summary 事件透出
+			}
+
+		case "content_block_delta":
+			if event.Delta == nil {
+				continue
+			}
+			switch event.Delta.Type {
+			case "text_delta":
+				if event.Delta.Text != nil && *event.Delta.Text != "" {
+					textBuf.WriteString(*event.Delta.Text)
+					openai.EmitResponsesSSE(writer, "response.output_text.delta", map[string]any{
+						"type":          "response.output_text.delta",
+						"item_id":       msgID,
+						"output_index":  0,
+						"content_index": contentIndex,
+						"delta":         *event.Delta.Text,
+					})
+				}
+			case "thinking_delta":
+				if event.Delta.Thinking != nil && *event.Delta.Thinking != "" {
+					openai.EmitResponsesSSE(writer, "response.reasoning_summary_text.delta", map[string]any{
+						"type":          "response.reasoning_summary_text.delta",
+						"item_id":       msgID,
+						"output_index":  0,
+						"summary_index": 0,
+						"delta":         *event.Delta.Thinking,
+					})
+				}
+			case "input_json_delta":
+				if event.Delta.PartialJSON != nil && *event.Delta.PartialJSON != "" && currentTool != nil {
+					currentTool.args.WriteString(*event.Delta.PartialJSON)
+					openai.EmitResponsesSSE(writer, "response.function_call_arguments.delta", map[string]any{
+						"type":         "response.function_call_arguments.delta",
+						"item_id":      currentTool.id,
+						"output_index": toolIndexByID[currentTool.id],
+						"delta":        *event.Delta.PartialJSON,
+					})
+				}
+			case "signature_delta":
+				// 思考签名无 Responses 对应物，忽略
+			}
+
+		case "content_block_stop":
+			// 块级收尾统一延迟到 message_stop / finish，这里仅结束当前工具块的增量定向
+			if currentTool != nil {
+				currentTool = nil
+			}
+
+		case "message_delta":
+			if event.Usage != nil {
+				if event.Usage.InputTokens > 0 {
+					usage.InputTokens = event.Usage.InputTokens
+				}
+				usage.OutputTokens = event.Usage.OutputTokens
+				if event.Usage.CacheReadInputTokens > 0 {
+					usage.CacheReadInputTokens = event.Usage.CacheReadInputTokens
+				}
+				if event.Usage.CacheCreationInputTokens > 0 {
+					usage.CacheCreationInputTokens = event.Usage.CacheCreationInputTokens
+				}
+				if event.Usage.CacheCreation != nil {
+					usage.CacheCreation = event.Usage.CacheCreation
+				}
+			}
+
+		case "message_stop":
+			finish()
+			return buildBillingUsage(), nil
+
+		case "error":
+			errMsg := "claude stream error"
+			if event.Error != nil {
+				if b, err := json.Marshal(event.Error); err == nil {
+					errMsg = fmt.Sprintf("claude stream error: %s", string(b))
+				}
+			}
+			info.StreamStatus.SetEndReason(common.StreamEndReasonError, fmt.Errorf("%s", errMsg))
+			g.Log().Warningf(ctx, "[Claude.handleStreamToResponses] %v", errMsg)
+			if !sentCreated {
+				errBody, _ := json.Marshal(map[string]any{
+					"error": map[string]any{
+						"message": errMsg,
+						"type":    "upstream_error",
+						"param":   nil,
+						"code":    "upstream_stream_error",
+					},
+				})
+				writeResponsesErrorBody(writer, errBody)
+				upstreamErr := constant.NewUpstreamError(http.StatusBadGateway, string(errBody), nil)
+				upstreamErr.ResponseWritten = true
+				return buildBillingUsage(), upstreamErr
+			}
+			return buildBillingUsage(), constant.NewUpstreamError(http.StatusBadGateway, errMsg, nil)
+
+		case "ping":
+			// 保活事件，忽略
+		}
+	}
+
+	if err := scanner.Err(); err != nil && err != io.EOF && ctx.Err() == nil {
+		info.StreamStatus.SetEndReason(common.StreamEndReasonError, err)
+		return buildBillingUsage(), fmt.Errorf("stream scanner error: %w", err)
+	}
+
+	// 上游未发 message_stop 即断流：仍合成 completed，避免客户端挂起
+	finish()
+
+	// 计费返回值按 Claude 口径（input 不含缓存，cache 独立列）
+	return buildBillingUsage(), nil
+}
+
+// writeResponsesErrorBody 写入 Responses 兼容的错误响应体（SSE 头提交后状态码已固定 200）
+func writeResponsesErrorBody(writer http.ResponseWriter, body []byte) {
+	writer.Header().Set("Content-Type", "application/json")
+	_, _ = writer.Write(body)
+	if f, ok := writer.(http.Flusher); ok {
+		f.Flush()
+	}
+}

relay/channel/claude/responses_bridge_test.go

@@ -0,0 +1,228 @@
+package claude
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/qianfree/team-api/relay/common"
+	"github.com/qianfree/team-api/relay/constant"
+)
+
+// responsesInboundInfo 构造 responses 入站（codex 等）+ Claude 上游渠道的 RelayInfo
+func responsesInboundInfo(isStream bool) *common.RelayInfo {
+	return &common.RelayInfo{
+		RelayMode:       int(constant.RelayModeResponses),
+		IsStream:        isStream,
+		InboundFormat:   constant.RelayFormatResponses,
+		ClientFormat:    constant.RelayFormatResponses,
+		OriginModelName: "glm-5.3",
+		StreamStatus:    common.NewStreamStatus(),
+		ChannelMeta: &common.ChannelMeta{
+			BaseURL:           "https://upstream.example.com",
+			UpstreamModelName: "glm-5.3",
+			IsModelMapped:     false,
+		},
+	}
+}
+
+// TestGetRequestURL_ResponsesMode responses/compact 模式同样打 /v1/messages
+// （请求侧转 Claude Messages 格式，响应侧转回 Responses）。
+func TestGetRequestURL_ResponsesMode(t *testing.T) {
+	for _, mode := range []constant.RelayMode{constant.RelayModeResponses, constant.RelayModeResponsesCompact} {
+		a := &Adaptor{}
+		info := responsesInboundInfo(false)
+		info.RelayMode = int(mode)
+		got, err := a.GetRequestURL(info)
+		if err != nil {
+			t.Fatalf("GetRequestURL(%v) error: %v", mode, err)
+		}
+		if want := "https://upstream.example.com/v1/messages"; got != want {
+			t.Errorf("GetRequestURL(%v) = %q, want %q", mode, got, want)
+		}
+	}
+}
+
+// TestDoResponse_ResponsesInboundStream_TextAndToolCall
+// Claude SSE（文本 + 工具调用）完整转换为 Responses SSE：
+// 事件序列、工具参数聚合、completed 的 usage 映射（OpenAI 语义）。
+func TestDoResponse_ResponsesInboundStream_TextAndToolCall(t *testing.T) {
+	ss := strings.Join([]string{
+		`data: {"type":"message_start","message":{"id":"msg_01","model":"glm-5.3","usage":{"input_tokens":10,"output_tokens":1}}}`,
+		``,
+		`data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`,
+		``,
+		`data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"你好"}}`,
+		``,
+		`data: {"type":"content_block_stop","index":0}`,
+		``,
+		`data: {"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"toolu_01","name":"shell","input":{}}}`,
+		``,
+		`data: {"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"cmd\":\"ls\"}"}}`,
+		``,
+		`data: {"type":"content_block_stop","index":1}`,
+		``,
+		`data: {"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"input_tokens":10,"cache_read_input_tokens":4,"output_tokens":7}}`,
+		``,
+		`data: {"type":"message_stop"}`,
+		``,
+	}, "\n")
+	resp := &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": {"text/event-stream"}},
+		Body:       io.NopCloser(strings.NewReader(ss)),
+	}
+
+	info := responsesInboundInfo(true)
+	rec := httptest.NewRecorder()
+	a := &Adaptor{}
+	usage, err := a.DoResponse(context.Background(), resp, info, rec)
+	if err != nil {
+		t.Fatalf("DoResponse error: %v", err)
+	}
+	out := rec.Body.String()
+
+	// 事件序列完整性
+	for _, want := range []string{
+		"event: response.created",
+		"event: response.output_item.added",
+		"event: response.content_part.added",
+		`event: response.output_text.delta`,
+		"event: response.output_text.done",
+		"event: response.content_part.done",
+		"event: response.output_item.done",
+		"event: response.function_call_arguments.delta",
+		"event: response.function_call_arguments.done",
+		"event: response.completed",
+	} {
+		if !strings.Contains(out, want+"\n") {
+			t.Errorf("missing event %q in output:\n%s", want, out)
+		}
+	}
+	if !strings.Contains(out, `"delta":"你好"`) {
+		t.Errorf("text delta missing:\n%s", out)
+	}
+	if !strings.Contains(out, `"delta":"{\"cmd\":\"ls\"}"`) {
+		t.Errorf("tool arguments delta missing:\n%s", out)
+	}
+	if !strings.Contains(out, `"name":"shell"`) {
+		t.Errorf("tool name missing:\n%s", out)
+	}
+
+	// completed 的 usage 用 OpenAI 语义：input 含缓存（10+4），cached_tokens 为子集
+	if !strings.Contains(out, `"input_tokens":14`) {
+		t.Errorf("completed usage input_tokens should be 14 (input+cache_read):\n%s", out)
+	}
+	if !strings.Contains(out, `"cached_tokens":4`) {
+		t.Errorf("completed usage cached_tokens should be 4:\n%s", out)
+	}
+	if !strings.Contains(out, `"output_tokens":7`) {
+		t.Errorf("completed usage output_tokens should be 7:\n%s", out)
+	}
+
+	// completed 的 output 数组：message + function_call
+	completedIdx := strings.LastIndex(out, "event: response.completed")
+	if completedIdx < 0 {
+		t.Fatalf("response.completed missing:\n%s", out)
+	}
+	completed := out[completedIdx:]
+	if !strings.Contains(completed, `"type":"function_call"`) {
+		t.Errorf("completed output should contain function_call item:\n%s", completed)
+	}
+
+	// 计费返回值按 Claude 口径（input 不含缓存，cache 独立）
+	if usage.PromptTokens != 10 || usage.CompletionTokens != 7 {
+		t.Errorf("billing usage = %+v, want prompt=10 completion=7", usage)
+	}
+	if usage.PromptTokensDetails == nil || usage.PromptTokensDetails.CachedTokens != 4 {
+		t.Errorf("billing usage cache detail = %+v, want cached_tokens=4", usage.PromptTokensDetails)
+	}
+}
+
+// TestDoResponse_ResponsesInboundStream_UpstreamErrorBeforeEvents
+// 首个事件即 error：返回 upstream error 并写入错误体，不合成假成功的 response.completed。
+func TestDoResponse_ResponsesInboundStream_UpstreamErrorBeforeEvents(t *testing.T) {
+	ss := strings.Join([]string{
+		`data: {"type":"error","error":{"type":"overloaded_error","message":"Overloaded"}}`,
+		``,
+	}, "\n")
+	resp := &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": {"text/event-stream"}},
+		Body:       io.NopCloser(strings.NewReader(ss)),
+	}
+
+	info := responsesInboundInfo(true)
+	rec := httptest.NewRecorder()
+	a := &Adaptor{}
+	_, err := a.DoResponse(context.Background(), resp, info, rec)
+	if err == nil {
+		t.Fatal("upstream error event should return an upstream error, got nil")
+	}
+	if !strings.Contains(rec.Body.String(), "overloaded_error") {
+		t.Errorf("error body not written to client: %s", rec.Body.String())
+	}
+	if strings.Contains(rec.Body.String(), "response.completed") {
+		t.Errorf("should not synthesize response.completed on upstream error: %s", rec.Body.String())
+	}
+}
+
+// TestDoResponse_ResponsesInboundNonStream Claude 非流式 JSON 转换为 Responses 格式。
+func TestDoResponse_ResponsesInboundNonStream(t *testing.T) {
+	body := `{"id":"msg_02","type":"message","role":"assistant","model":"glm-5.3","content":[` +
+		`{"type":"text","text":"答案是42"},` +
+		`{"type":"tool_use","id":"toolu_02","name":"read","input":{"path":"a.go"}}` +
+		`],"stop_reason":"end_turn","usage":{"input_tokens":10,"cache_read_input_tokens":4,"output_tokens":5}}`
+	resp := &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": {"application/json"}},
+		Body:       io.NopCloser(strings.NewReader(body)),
+	}
+
+	info := responsesInboundInfo(false)
+	rec := httptest.NewRecorder()
+	a := &Adaptor{}
+	usage, err := a.DoResponse(context.Background(), resp, info, rec)
+	if err != nil {
+		t.Fatalf("DoResponse error: %v", err)
+	}
+
+	var m map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &m); err != nil {
+		t.Fatalf("response body not json: %v\n%s", err, rec.Body.String())
+	}
+	if m["object"] != "response" || m["status"] != "completed" {
+		t.Errorf("response envelope wrong: %v", m)
+	}
+	output, _ := m["output"].([]any)
+	if len(output) != 2 {
+		t.Fatalf("output = %v, want 2 items (message + function_call)", output)
+	}
+	msg, _ := output[0].(map[string]any)
+	if msg["type"] != "message" {
+		t.Errorf("output[0] = %v, want message", msg)
+	}
+	fn, _ := output[1].(map[string]any)
+	if fn["type"] != "function_call" || fn["name"] != "read" {
+		t.Errorf("output[1] = %v, want function_call read", fn)
+	}
+
+	// usage 用 OpenAI 语义：input 含缓存
+	u, _ := m["usage"].(map[string]any)
+	if u["input_tokens"] != float64(14) || u["output_tokens"] != float64(5) {
+		t.Errorf("usage = %v, want input=14 output=5", u)
+	}
+	d, _ := u["input_tokens_details"].(map[string]any)
+	if d["cached_tokens"] != float64(4) {
+		t.Errorf("input_tokens_details = %v, want cached_tokens=4", d)
+	}
+
+	// 计费口径按 Claude（input 不含缓存）
+	if usage.PromptTokens != 10 || usage.CompletionTokens != 5 {
+		t.Errorf("billing usage = %+v, want prompt=10 completion=5", usage)
+	}
+}

relay/channel/openai/adaptor_responses_test.go

@@ -353,7 +353,7 @@ func TestBuildResponsesObjectMap_StoreFalseEcho(t *testing.T) {
 	info := &common.RelayInfo{
 		ResponsesRequest: &dto.OpenAIResponsesRequest{Temperature: &temp},
 	}
-	m := buildResponsesObjectMap("resp_1", 1700000000, "completed", "gpt-4o", []any{}, nil, nil, info)
+	m := BuildResponsesObjectMap("resp_1", 1700000000, "completed", "gpt-4o", []any{}, nil, nil, info)
 	if m["store"] != false {
 		t.Errorf("store = %v, want false", m["store"])
 	}

relay/channel/openai/responses.go

@@ -316,13 +316,13 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 			}
 
 			// response.created
-			emitResponsesSSE(writer, "response.created", map[string]any{
+			EmitResponsesSSE(writer, "response.created", map[string]any{
 				"type":     "response.created",
-				"response": buildResponsesObjectMap(respID, createdAt, "in_progress", modelName, []any{}, nil, nil, info),
+				"response": BuildResponsesObjectMap(respID, createdAt, "in_progress", modelName, []any{}, nil, nil, info),
 			})
 
 			// response.output_item.added
-			emitResponsesSSE(writer, "response.output_item.added", map[string]any{
+			EmitResponsesSSE(writer, "response.output_item.added", map[string]any{
 				"type":         "response.output_item.added",
 				"output_index": outputIndex,
 				"item": map[string]any{
@@ -335,7 +335,7 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 			})
 
 			// response.content_part.added
-			emitResponsesSSE(writer, "response.content_part.added", map[string]any{
+			EmitResponsesSSE(writer, "response.content_part.added", map[string]any{
 				"type":          "response.content_part.added",
 				"item_id":       msgID,
 				"output_index":  outputIndex,
@@ -368,7 +368,7 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 				}
 				if deltaText != "" {
 					contentBuilder.WriteString(deltaText)
-					emitResponsesSSE(writer, "response.output_text.delta", map[string]any{
+					EmitResponsesSSE(writer, "response.output_text.delta", map[string]any{
 						"type":          "response.output_text.delta",
 						"item_id":       msgID,
 						"output_index":  outputIndex,
@@ -380,7 +380,7 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 
 			// 推理内容
 			if choice.Delta.ReasoningContent != nil && *choice.Delta.ReasoningContent != "" {
-				emitResponsesSSE(writer, "response.reasoning_summary_text.delta", map[string]any{
+				EmitResponsesSSE(writer, "response.reasoning_summary_text.delta", map[string]any{
 					"type":          "response.reasoning_summary_text.delta",
 					"item_id":       msgID,
 					"output_index":  outputIndex,
@@ -401,14 +401,14 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 					// 先关闭文本 content part
 					if !sentTextDone {
 						finishedText := contentBuilder.String()
-						emitResponsesSSE(writer, "response.output_text.done", map[string]any{
+						EmitResponsesSSE(writer, "response.output_text.done", map[string]any{
 							"type":          "response.output_text.done",
 							"item_id":       msgID,
 							"output_index":  outputIndex,
 							"content_index": contentIndex,
 							"text":          finishedText,
 						})
-						emitResponsesSSE(writer, "response.content_part.done", map[string]any{
+						EmitResponsesSSE(writer, "response.content_part.done", map[string]any{
 							"type":          "response.content_part.done",
 							"item_id":       msgID,
 							"output_index":  outputIndex,
@@ -419,7 +419,7 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 								"annotations": []any{},
 							},
 						})
-						emitResponsesSSE(writer, "response.output_item.done", map[string]any{
+						EmitResponsesSSE(writer, "response.output_item.done", map[string]any{
 							"type":         "response.output_item.done",
 							"output_index": outputIndex,
 							"item": map[string]any{
@@ -444,7 +444,7 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 					toolCallNameByID[callID] = tc.Function.Name
 					toolCallArgsByID[callID] = ""
 
-					emitResponsesSSE(writer, "response.output_item.added", map[string]any{
+					EmitResponsesSSE(writer, "response.output_item.added", map[string]any{
 						"type":         "response.output_item.added",
 						"output_index": outputIndex,
 						"item": map[string]any{
@@ -469,7 +469,7 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 				// 工具调用 arguments 增量
 				if tc.Function.Arguments != "" {
 					toolCallArgsByID[callID] += tc.Function.Arguments
-					emitResponsesSSE(writer, "response.function_call_arguments.delta", map[string]any{
+					EmitResponsesSSE(writer, "response.function_call_arguments.delta", map[string]any{
 						"type":         "response.function_call_arguments.delta",
 						"item_id":      callID,
 						"output_index": toolCallIndexByID[callID],
@@ -484,14 +484,14 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 
 				// 关闭文本 content part（如果尚未关闭）
 				if !sentTextDone {
-					emitResponsesSSE(writer, "response.output_text.done", map[string]any{
+					EmitResponsesSSE(writer, "response.output_text.done", map[string]any{
 						"type":          "response.output_text.done",
 						"item_id":       msgID,
 						"output_index":  outputIndex,
 						"content_index": contentIndex,
 						"text":          finishedText,
 					})
-					emitResponsesSSE(writer, "response.content_part.done", map[string]any{
+					EmitResponsesSSE(writer, "response.content_part.done", map[string]any{
 						"type":          "response.content_part.done",
 						"item_id":       msgID,
 						"output_index":  outputIndex,
@@ -502,7 +502,7 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 							"annotations": []any{},
 						},
 					})
-					emitResponsesSSE(writer, "response.output_item.done", map[string]any{
+					EmitResponsesSSE(writer, "response.output_item.done", map[string]any{
 						"type":         "response.output_item.done",
 						"output_index": outputIndex,
 						"item": map[string]any{
@@ -523,13 +523,13 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 
 				// 发送每个 tool call 的 function_call_arguments.done + output_item.done
 				for tcID, tcIdx := range toolCallIndexByID {
-					emitResponsesSSE(writer, "response.function_call_arguments.done", map[string]any{
+					EmitResponsesSSE(writer, "response.function_call_arguments.done", map[string]any{
 						"type":         "response.function_call_arguments.done",
 						"item_id":      tcID,
 						"output_index": tcIdx,
 						"arguments":    toolCallArgsByID[tcID],
 					})
-					emitResponsesSSE(writer, "response.output_item.done", map[string]any{
+					EmitResponsesSSE(writer, "response.output_item.done", map[string]any{
 						"type":         "response.output_item.done",
 						"output_index": tcIdx,
 						"item": map[string]any{
@@ -624,9 +624,9 @@ func (a *Adaptor) handleResponsesInboundStream(ctx context.Context, resp *http.R
 
 	// response.completed
 	completedAt := int(time.Now().Unix())
-	emitResponsesSSE(writer, "response.completed", map[string]any{
+	EmitResponsesSSE(writer, "response.completed", map[string]any{
 		"type":     "response.completed",
-		"response": buildResponsesObjectMap(respID, createdAt, "completed", modelName, finalOutput, buildResponsesUsageMap(&usage), &completedAt, info),
+		"response": BuildResponsesObjectMap(respID, createdAt, "completed", modelName, finalOutput, BuildResponsesUsageMap(&usage), &completedAt, info),
 	})
 
 	if info.StreamStatus.GetEndReason() == "" {
@@ -656,8 +656,8 @@ func extractStreamEmbeddedError(data []byte) (json.RawMessage, bool) {
 	return errBody, true
 }
 
-// emitResponsesSSE 发送一个 Responses API 格式的 SSE 事件
-func emitResponsesSSE(w http.ResponseWriter, eventType string, data any) {
+// EmitResponsesSSE 发送一个 Responses API 格式的 SSE 事件（导出供 claude 等适配器的 Responses 桥接复用）
+func EmitResponsesSSE(w http.ResponseWriter, eventType string, data any) {
 	jsonData, err := json.Marshal(data)
 	if err != nil {
 		return
@@ -671,7 +671,7 @@ func emitResponsesSSE(w http.ResponseWriter, eventType string, data any) {
 // buildResponsesObjectMap 构建 Responses API response 对象的完整字段 map。
 // 请求参数从 info.ResponsesRequest echo（快照缺失时回退默认值）；
 // store 恒为 false——合成响应不落上游存储，客户端不可经生命周期端点 retrieve。
-func buildResponsesObjectMap(respID string, createdAt int, status string, model string, output any, usageObj map[string]any, completedAt *int, info *common.RelayInfo) map[string]any {
+func BuildResponsesObjectMap(respID string, createdAt int, status string, model string, output any, usageObj map[string]any, completedAt *int, info *common.RelayInfo) map[string]any {
 	echo := extractResponsesRequestEcho(info)
 	m := map[string]any{
 		"id":                   respID,
@@ -707,7 +707,7 @@ func buildResponsesObjectMap(respID string, createdAt int, status string, model
 }
 
 // buildResponsesUsageMap 构建 Responses API usage 对象
-func buildResponsesUsageMap(usage *common.Usage) map[string]any {
+func BuildResponsesUsageMap(usage *common.Usage) map[string]any {
 	inputDetails := map[string]any{"cached_tokens": 0}
 	outputDetails := map[string]any{"reasoning_tokens": 0}
 	if usage.PromptTokensDetails != nil {