oci-portal/internal/service/anthresponses.go

package service

import (
	"encoding/json"
	"fmt"
	"strings"

	"oci-portal/internal/aiwire"
)

// Anthropic Messages ↔ OCI OpenAI 兼容面(/actions/v1/responses)直通转换。
// typed chat 面剔除后 Messages 入口的唯一上游通路。语义损失(README 已披露):
// stop_sequences / top_k / metadata / thinking 无对应字段,忽略;上游 reasoning
// 输出项与增量事件丢弃(Anthropic thinking 块含签名语义,不伪造)。

// ErrAiUnsupportedBlock 表示请求含网关无法承接的内容块(仅支持文本/图片/工具块)。
var ErrAiUnsupportedBlock = fmt.Errorf("暂不支持文本与图片以外的内容块")

// AnthropicToResponsesBody 把 Messages 请求转为直通 body(强制 store:false)。
func AnthropicToResponsesBody(req aiwire.MessagesRequest) ([]byte, error) {
	input, err := anthInputItems(req.Messages)
	if err != nil {
		return nil, err
	}
	body := map[string]any{"model": req.Model, "max_output_tokens": req.MaxTokens,
		"input": input, "store": false}
	if sys := req.SystemText(); sys != "" {
		body["instructions"] = sys
	}
	if req.Temperature != nil {
		body["temperature"] = *req.Temperature
	}
	if req.TopP != nil {
		body["top_p"] = *req.TopP
	}
	if tools := anthRespTools(req.Tools); tools != nil {
		body["tools"] = tools
	}
	if tc := anthRespToolChoice(req.ToolChoice); tc != nil {
		body["tool_choice"] = tc
	}
	if req.OutputConfig != nil && req.OutputConfig.Effort != "" {
		body["reasoning"] = map[string]string{"effort": strings.ToLower(req.OutputConfig.Effort)}
	}
	if req.Stream {
		body["stream"] = true
	}
	return json.Marshal(body)
}

// anthInputItems 把消息序列展开为 Responses input 项:tool_use / tool_result 为
// 独立 function_call / function_call_output 项,其余聚合为 message 项;遇独立项时
// 先冲刷已聚合部件,保持块间相对顺序。
func anthInputItems(messages []aiwire.AnthMessage) ([]any, error) {
	var items []any
	for _, m := range messages {
		var parts []map[string]any
		flush := func() {
			if len(parts) > 0 {
				items = append(items, map[string]any{"role": m.Role, "content": parts})
				parts = nil
			}
		}
		for _, b := range m.Content.AllBlocks() {
			item, part, err := anthBlockItem(m.Role, b)
			if err != nil {
				return nil, err
			}
			if part != nil {
				parts = append(parts, part)
			}
			if item != nil {
				flush()
				items = append(items, item)
			}
		}
		flush()
	}
	return items, nil
}

// anthBlockItem 把单个内容块转为独立项或消息部件(thinking 忽略,未知块拒绝)。
func anthBlockItem(role string, b aiwire.AnthBlock) (item, part map[string]any, err error) {
	switch b.Type {
	case "text":
		return nil, respTextPart(role, b.Text), nil
	case "image":
		part, err = anthImageInput(b.Source)
		return nil, part, err
	case "tool_use":
		return map[string]any{"type": "function_call", "call_id": b.ID,
			"name": b.Name, "arguments": string(b.Input)}, nil, nil
	case "tool_result":
		return map[string]any{"type": "function_call_output",
			"call_id": b.ToolUseID, "output": b.ResultText()}, nil, nil
	case "thinking", "redacted_thinking":
		return nil, nil, nil
	default:
		return nil, nil, ErrAiUnsupportedBlock
	}
}

// respTextPart 按角色选择 Responses 文本部件类型(assistant 历史为 output_text);
// Anthropic 与 Chat Completions 两条转换链共用。
func respTextPart(role, text string) map[string]any {
	if role == "assistant" {
		return map[string]any{"type": "output_text", "text": text}
	}
	return map[string]any{"type": "input_text", "text": text}
}

// anthImageInput 把 Anthropic image source 转为 Responses input_image 部件。
func anthImageInput(source json.RawMessage) (map[string]any, error) {
	var src struct {
		Type      string `json:"type"`
		MediaType string `json:"media_type"`
		Data      string `json:"data"`
		URL       string `json:"url"`
	}
	if err := json.Unmarshal(source, &src); err != nil {
		return nil, fmt.Errorf("image source 解析失败: %w", err)
	}
	switch src.Type {
	case "base64":
		if src.MediaType == "" || src.Data == "" {
			return nil, fmt.Errorf("image source 缺少 media_type 或 data")
		}
		return map[string]any{"type": "input_image",
			"image_url": "data:" + src.MediaType + ";base64," + src.Data}, nil
	case "url":
		if src.URL == "" {
			return nil, fmt.Errorf("image source 缺少 url")
		}
		return map[string]any{"type": "input_image", "image_url": src.URL}, nil
	}
	return nil, fmt.Errorf("不支持的 image source 类型 %q", src.Type)
}

func anthRespTools(tools []aiwire.AnthTool) []map[string]any {
	if len(tools) == 0 {
		return nil
	}
	out := make([]map[string]any, 0, len(tools))
	for _, t := range tools {
		out = append(out, map[string]any{"type": "function", "name": t.Name,
			"description": t.Description, "parameters": t.InputSchema})
	}
	return out
}

// anthRespToolChoice 映射 {type:auto|any|tool|none,name} → Responses 形态。
func anthRespToolChoice(raw json.RawMessage) any {
	if len(raw) == 0 {
		return nil
	}
	var tc struct {
		Type string `json:"type"`
		Name string `json:"name"`
	}
	if json.Unmarshal(raw, &tc) != nil {
		return nil
	}
	switch tc.Type {
	case "auto":
		return "auto"
	case "any":
		return "required"
	case "none":
		return "none"
	case "tool":
		return map[string]string{"type": "function", "name": tc.Name}
	}
	return nil
}

// respPayload 是直通响应中本转换关心的子集(未知字段忽略)。
type respPayload struct {
	Model             string             `json:"model"`
	Status            string             `json:"status"`
	IncompleteDetails *respIncomplete    `json:"incomplete_details"`
	Output            []respOutputItem   `json:"output"`
	Usage             *aiwire.RespUsage  `json:"usage"`
	Error             *map[string]string `json:"error"`
}

type respIncomplete struct {
	Reason string `json:"reason"`
}

type respOutputItem struct {
	Type      string `json:"type"`
	CallID    string `json:"call_id"`
	Name      string `json:"name"`
	Arguments string `json:"arguments"`
	Content   []struct {
		Type string `json:"type"`
		Text string `json:"text"`
	} `json:"content"`
}

// ResponsesToAnthropic 把直通非流式响应转为 Anthropic Messages 响应。
func ResponsesToAnthropic(payload []byte, msgID string) (*aiwire.MessagesResponse, error) {
	var resp respPayload
	if err := json.Unmarshal(payload, &resp); err != nil {
		return nil, fmt.Errorf("解析上游响应: %w", err)
	}
	out := &aiwire.MessagesResponse{ID: msgID, Type: "message", Role: "assistant",
		Model: resp.Model, Content: []aiwire.AnthBlock{}, StopReason: "end_turn"}
	for _, item := range resp.Output {
		switch item.Type {
		case "message":
			for _, part := range item.Content {
				if part.Type == "output_text" && part.Text != "" {
					out.Content = append(out.Content, aiwire.AnthBlock{Type: "text", Text: part.Text})
				}
			}
		case "function_call":
			out.Content = append(out.Content, aiwire.AnthBlock{Type: "tool_use", ID: item.CallID,
				Name: item.Name, Input: argsToJSON(item.Arguments)})
			out.StopReason = "tool_use"
		}
	}
	if respTruncated(&resp) {
		out.StopReason = "max_tokens"
	}
	out.Usage = anthUsageFromResp(resp.Usage)
	return out, nil
}

func anthUsageFromResp(u *aiwire.RespUsage) aiwire.AnthUsage {
	if u == nil {
		return aiwire.AnthUsage{}
	}
	return aiwire.AnthUsage{InputTokens: u.InputTokens, OutputTokens: u.OutputTokens,
		CacheReadInputTokens: u.InputTokensDetails.CachedTokens}
}

// argsToJSON 保证 tool_use.input 是合法 JSON 对象(模型可能产出非法片段)。
func argsToJSON(args string) json.RawMessage {
	trimmed := strings.TrimSpace(args)
	if trimmed == "" {
		return json.RawMessage(`{}`)
	}
	if json.Valid([]byte(trimmed)) {
		return json.RawMessage(trimmed)
	}
	b, _ := json.Marshal(map[string]string{"_raw": args})
	return b
}

// ---- Anthropic 流式桥:Responses SSE 事件 → Anthropic 事件序列 ----

// AnthEvent 是一条待写出的 Anthropic SSE 事件。
type AnthEvent struct {
	Event string
	Data  any
}

// AnthRespBridge 把直通 SSE 事件流桥接为 Anthropic 事件序列:
// message_start → content_block_start/delta/stop(text 与 tool_use 分块)→ message_delta → message_stop。
// reasoning 系列事件丢弃;上游 error / response.failed 转 Anthropic error 事件透传。
type AnthRespBridge struct {
	id, model   string
	started     bool
	blockOpen   bool
	blockIsTool bool
	blockIndex  int
	stopReason  string
	usage       aiwire.AnthUsage
	// sawTerminal 标记收到过终态事件(completed/incomplete/failed);
	// 上游流提前 EOF 时据此发 error 事件而非伪装正常结束
	sawTerminal bool
	// errMsg 记录上游错误消息,非空即本流已失败(供调用日志)
	errMsg string
}

// NewAnthRespBridge 构造桥;id 为响应消息 ID。
func NewAnthRespBridge(id, model string) *AnthRespBridge {
	return &AnthRespBridge{id: id, model: model, blockIndex: -1, stopReason: "end_turn"}
}

// respStreamEvent 是直通 SSE data JSON 中桥关心的子集。
type respStreamEvent struct {
	Type     string          `json:"type"`
	Delta    string          `json:"delta"`
	Message  string          `json:"message"`
	Item     *respOutputItem `json:"item"`
	Response *respPayload    `json:"response"`
}

// Feed 消费一行 SSE data JSON,返回应立即写出的事件。
func (st *AnthRespBridge) Feed(data []byte) []AnthEvent {
	var ev respStreamEvent
	if json.Unmarshal(data, &ev) != nil {
		return nil
	}
	if ev.Type == "error" || ev.Type == "response.failed" {
		return st.failWith(ev)
	}
	// message_start 延迟到首个可见输出事件:created / reasoning 阶段不向客户端
	// 写任何字节,上游此段断流时 handler 才有降级非流式重做的无感窗口
	var events []AnthEvent
	switch ev.Type {
	case "response.output_item.added":
		if ev.Item != nil && ev.Item.Type == "function_call" {
			events = append(events, st.ensureStarted()...)
			events = append(events, st.openBlock(true, ev.Item.CallID, ev.Item.Name)...)
			st.stopReason = "tool_use"
		}
	case "response.output_text.delta":
		events = append(events, st.ensureStarted()...)
		events = append(events, st.textDelta(ev.Delta)...)
	case "response.function_call_arguments.delta":
		events = append(events, st.argsDelta(ev.Delta)...)
	case "response.completed", "response.incomplete":
		st.sawTerminal = true
		st.finishFrom(ev.Response)
	}
	return events
}

// ensureStarted 在首个可见输出事件前补发 message_start(仅一次)。
func (st *AnthRespBridge) ensureStarted() []AnthEvent {
	if st.started {
		return nil
	}
	st.started = true
	return []AnthEvent{st.startEvent()}
}

// failWith 记录上游错误并产出 Anthropic error 事件(每流至多一次)。
func (st *AnthRespBridge) failWith(ev respStreamEvent) []AnthEvent {
	if st.errMsg != "" {
		return nil
	}
	st.sawTerminal = true
	msg := ev.Message
	if ev.Type == "response.failed" {
		st.finishFrom(ev.Response)
		if m := respErrorMsg(ev.Response); m != "" {
			msg = m
		}
	}
	if msg == "" {
		msg = "上游返回错误事件 " + ev.Type
	}
	st.errMsg = msg
	return []AnthEvent{st.errorEvent()}
}

// respErrorMsg 提取 response.failed 载荷中的错误消息。
func respErrorMsg(resp *respPayload) string {
	if resp == nil || resp.Error == nil {
		return ""
	}
	return (*resp.Error)["message"]
}

// errorEvent 按 Anthropic 流式协议产出 error 事件。
func (st *AnthRespBridge) errorEvent() AnthEvent {
	return AnthEvent{Event: "error", Data: map[string]any{
		"type":  "error",
		"error": map[string]string{"type": "api_error", "message": st.errMsg},
	}}
}

// Err 返回上游错误消息;空串表示流正常(供调用日志)。
func (st *AnthRespBridge) Err() string { return st.errMsg }

// SawTerminal 报告是否收到过终态事件;false 即上游流提前终止。
func (st *AnthRespBridge) SawTerminal() bool { return st.sawTerminal }

// AnthMessageEvents 把完整 Messages 响应展开为标准事件序列,
// 供流式上游断流后的非流式降级结果推送(客户端协议不变)。
func AnthMessageEvents(m *aiwire.MessagesResponse) []AnthEvent {
	events := []AnthEvent{{Event: "message_start", Data: map[string]any{
		"type": "message_start",
		"message": map[string]any{
			"id": m.ID, "type": "message", "role": m.Role, "model": m.Model,
			"content": []any{}, "stop_reason": nil,
			"usage": map[string]int{"input_tokens": 0, "output_tokens": 0},
		},
	}}}
	for i, b := range m.Content {
		events = append(events, anthBlockEvents(i, b)...)
	}
	usage := map[string]int{"input_tokens": m.Usage.InputTokens, "output_tokens": m.Usage.OutputTokens}
	if m.Usage.CacheReadInputTokens > 0 {
		usage["cache_read_input_tokens"] = m.Usage.CacheReadInputTokens
	}
	events = append(events,
		AnthEvent{Event: "message_delta", Data: map[string]any{
			"type":  "message_delta",
			"delta": map[string]any{"stop_reason": m.StopReason, "stop_sequence": nil},
			"usage": usage,
		}},
		AnthEvent{Event: "message_stop", Data: map[string]any{"type": "message_stop"}})
	return events
}

// anthBlockEvents 把一个内容块展开为 start / delta / stop 三事件。
func anthBlockEvents(index int, b aiwire.AnthBlock) []AnthEvent {
	var start, delta map[string]any
	if b.Type == "tool_use" {
		start = map[string]any{"type": "tool_use", "id": b.ID, "name": b.Name, "input": map[string]any{}}
		delta = map[string]any{"type": "input_json_delta", "partial_json": string(b.Input)}
	} else {
		start = map[string]any{"type": "text", "text": ""}
		delta = map[string]any{"type": "text_delta", "text": b.Text}
	}
	return []AnthEvent{
		{Event: "content_block_start", Data: map[string]any{
			"type": "content_block_start", "index": index, "content_block": start}},
		{Event: "content_block_delta", Data: map[string]any{
			"type": "content_block_delta", "index": index, "delta": delta}},
		{Event: "content_block_stop", Data: map[string]any{
			"type": "content_block_stop", "index": index}},
	}
}

func (st *AnthRespBridge) textDelta(delta string) []AnthEvent {
	var events []AnthEvent
	if !st.blockOpen || st.blockIsTool {
		events = append(events, st.openBlock(false, "", "")...)
	}
	events = append(events, AnthEvent{Event: "content_block_delta", Data: map[string]any{
		"type": "content_block_delta", "index": st.blockIndex,
		"delta": map[string]string{"type": "text_delta", "text": delta},
	}})
	return events
}

func (st *AnthRespBridge) argsDelta(delta string) []AnthEvent {
	if !st.blockOpen || !st.blockIsTool {
		return nil
	}
	return []AnthEvent{{Event: "content_block_delta", Data: map[string]any{
		"type": "content_block_delta", "index": st.blockIndex,
		"delta": map[string]string{"type": "input_json_delta", "partial_json": delta},
	}}}
}

// finishFrom 记录终态:usage 与 stop_reason(max_output_tokens 截断 → max_tokens)。
func (st *AnthRespBridge) finishFrom(resp *respPayload) {
	if resp == nil {
		return
	}
	st.usage = anthUsageFromResp(resp.Usage)
	if respTruncated(resp) {
		st.stopReason = "max_tokens"
	}
}

func (st *AnthRespBridge) startEvent() AnthEvent {
	return AnthEvent{Event: "message_start", Data: map[string]any{
		"type": "message_start",
		"message": map[string]any{
			"id": st.id, "type": "message", "role": "assistant", "model": st.model,
			"content": []any{}, "stop_reason": nil,
			"usage": map[string]int{"input_tokens": 0, "output_tokens": 0},
		},
	}}
}

// openBlock 关闭当前块并打开新块(text 或 tool_use)。
func (st *AnthRespBridge) openBlock(isTool bool, toolID, toolName string) []AnthEvent {
	var events []AnthEvent
	if st.blockOpen {
		events = append(events, st.closeBlockEvent())
	}
	st.blockOpen, st.blockIsTool = true, isTool
	st.blockIndex++
	block := map[string]any{"type": "text", "text": ""}
	if isTool {
		block = map[string]any{"type": "tool_use", "id": toolID, "name": toolName, "input": map[string]any{}}
	}
	events = append(events, AnthEvent{Event: "content_block_start", Data: map[string]any{
		"type": "content_block_start", "index": st.blockIndex, "content_block": block,
	}})
	return events
}

func (st *AnthRespBridge) closeBlockEvent() AnthEvent {
	return AnthEvent{Event: "content_block_stop", Data: map[string]any{
		"type": "content_block_stop", "index": st.blockIndex,
	}}
}

// Finish 在上游流结束后收尾:关块 → message_delta(stop_reason+usage)→ message_stop。
// 已发过 error 事件的流不再补终态;未见终态事件即 EOF 视为上游提前终止,
// 发 error 事件而非伪装正常结束(否则客户端拿到"成功的空消息")。
func (st *AnthRespBridge) Finish() []AnthEvent {
	if st.errMsg != "" {
		return nil
	}
	if !st.sawTerminal {
		st.errMsg = "上游流提前终止,未返回终态事件"
		return []AnthEvent{st.errorEvent()}
	}
	var events []AnthEvent
	if !st.started {
		st.started = true
		events = append(events, st.startEvent())
	}
	if st.blockOpen {
		events = append(events, st.closeBlockEvent())
		st.blockOpen = false
	}
	usage := map[string]int{"output_tokens": st.usage.OutputTokens}
	if st.usage.InputTokens > 0 {
		usage["input_tokens"] = st.usage.InputTokens
	}
	if st.usage.CacheReadInputTokens > 0 {
		usage["cache_read_input_tokens"] = st.usage.CacheReadInputTokens
	}
	events = append(events,
		AnthEvent{Event: "message_delta", Data: map[string]any{
			"type":  "message_delta",
			"delta": map[string]any{"stop_reason": st.stopReason, "stop_sequence": nil},
			"usage": usage,
		}},
		AnthEvent{Event: "message_stop", Data: map[string]any{"type": "message_stop"}},
	)
	return events
}

// Usage 返回聚合到的用量(供调用日志)。
func (st *AnthRespBridge) Usage() aiwire.AnthUsage { return st.usage }