diff --git a/relay/channel/claude/adaptor.go b/relay/channel/claude/adaptor.go index aeea9c0..8119ee1 100644 --- a/relay/channel/claude/adaptor.go +++ b/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) diff --git a/relay/channel/claude/responses_bridge.go b/relay/channel/claude/responses_bridge.go new file mode 100644 index 0000000..3ce134b --- /dev/null +++ b/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() + } +} diff --git a/relay/channel/claude/responses_bridge_test.go b/relay/channel/claude/responses_bridge_test.go new file mode 100644 index 0000000..7c2666a --- /dev/null +++ b/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) + } +} diff --git a/relay/channel/openai/adaptor_responses_test.go b/relay/channel/openai/adaptor_responses_test.go index 87d9b90..7d62863 100644 --- a/relay/channel/openai/adaptor_responses_test.go +++ b/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"]) } diff --git a/relay/channel/openai/responses.go b/relay/channel/openai/responses.go index 84ee7e8..11f7165 100644 --- a/relay/channel/openai/responses.go +++ b/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 {