iop/apps/edge/internal/openai/provider_normalization.go

package openai

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

	edgeservice "iop/apps/edge/internal/service"
	"iop/packages/go/config"
)

// providerRequestRequirements is the caller-neutral request shape used for
// provider operation selection. Agent or SDK identity is intentionally absent.
type providerRequestRequirements struct {
	Effort           string
	HasTools         bool
	HasTokenBudget   bool
	Stream           bool
	StructuredOutput bool
}

type providerOperationPlan struct {
	Operation  config.ProtocolOperation
	Effort     string
	EffortWire string
}

type openAIResponsesBridgeResponse struct {
	ID     string `json:"id"`
	Model  string `json:"model"`
	Status string `json:"status"`
	Output []struct {
		Type      string `json:"type"`
		ID        string `json:"id"`
		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"`
		Summary []struct {
			Type string `json:"type"`
			Text string `json:"text"`
		} `json:"summary"`
	} `json:"output"`
	Usage struct {
		InputTokens  int `json:"input_tokens"`
		OutputTokens int `json:"output_tokens"`
		InputDetails struct {
			CachedTokens int `json:"cached_tokens"`
		} `json:"input_tokens_details"`
	} `json:"usage"`
	IncompleteDetails struct {
		Reason string `json:"reason"`
	} `json:"incomplete_details"`
}

func anthropicProviderRequirements(req anthropicMessageRequest) providerRequestRequirements {
	requirements := providerRequestRequirements{
		HasTools:         len(req.Tools) > 0,
		Stream:           req.Stream,
		HasTokenBudget:   req.Thinking != nil && req.Thinking.Type == "enabled",
		StructuredOutput: req.OutputConfig != nil && req.OutputConfig.Format != nil,
	}
	if req.OutputConfig != nil {
		requirements.Effort = strings.TrimSpace(req.OutputConfig.Effort)
	}
	return requirements
}

func decodeAnthropicProviderRequirements(body []byte) (providerRequestRequirements, error) {
	var request struct {
		Stream       bool                     `json:"stream"`
		Tools        []json.RawMessage        `json:"tools"`
		Thinking     *anthropicThinkingConfig `json:"thinking"`
		OutputConfig *anthropicOutputConfig   `json:"output_config"`
	}
	if err := json.Unmarshal(body, &request); err != nil {
		return providerRequestRequirements{}, fmt.Errorf("decode Messages request")
	}
	return anthropicProviderRequirements(anthropicMessageRequest{
		Stream: request.Stream, Tools: make([]anthropicTool, len(request.Tools)),
		Thinking: request.Thinking, OutputConfig: request.OutputConfig,
	}), nil
}

func decodeResponsesProviderRequirements(body []byte) (providerRequestRequirements, error) {
	var request struct {
		Tools     []json.RawMessage `json:"tools"`
		Reasoning *struct {
			Effort string `json:"effort"`
		} `json:"reasoning"`
	}
	if err := json.Unmarshal(body, &request); err != nil {
		return providerRequestRequirements{}, fmt.Errorf("decode Responses request")
	}
	requirements := providerRequestRequirements{HasTools: len(request.Tools) > 0}
	if request.Reasoning != nil {
		requirements.Effort = strings.TrimSpace(request.Reasoning.Effort)
	}
	return requirements, nil
}

func responsesCandidatePredicate(requirements providerRequestRequirements) edgeservice.ProviderPoolCandidatePredicate {
	return func(candidate edgeservice.ProviderPoolCandidate) bool {
		// A nil profile is the legacy tunnel contract: operation resolution is
		// deferred to the existing dispatch path, which must remain compatible.
		if candidate.ProtocolProfile == nil {
			return true
		}
		if candidate.ExecutionPath != string(edgeservice.ProviderPoolPathTunnel) {
			return requirements.Effort == "" && !requirements.HasTools && !requirements.HasTokenBudget
		}
		_, err := selectProviderOperation(*candidate.ProtocolProfile, config.OperationResponses, requirements)
		return err == nil
	}
}

func rewriteResponsesProviderControls(body []byte, target string, plan providerOperationPlan) ([]byte, error) {
	patches := make([]topLevelJSONPatch, 0, 2)
	if strings.TrimSpace(target) != "" {
		modelJSON, err := json.Marshal(target)
		if err != nil {
			return nil, err
		}
		patches = append(patches, topLevelJSONPatch{name: "model", value: modelJSON})
	}
	if plan.Effort != "" {
		var root map[string]json.RawMessage
		if err := json.Unmarshal(body, &root); err != nil {
			return nil, fmt.Errorf("decode Responses request")
		}
		var reasoning map[string]any
		if raw := root["reasoning"]; len(raw) > 0 && string(raw) != "null" {
			if err := json.Unmarshal(raw, &reasoning); err != nil {
				return nil, fmt.Errorf("reasoning must be an object")
			}
		}
		if reasoning == nil {
			reasoning = make(map[string]any)
		}
		reasoning["effort"] = plan.Effort
		reasoningJSON, err := json.Marshal(reasoning)
		if err != nil {
			return nil, err
		}
		patches = append(patches, topLevelJSONPatch{name: "reasoning", value: reasoningJSON})
	}
	if len(patches) == 0 {
		return body, nil
	}
	patchPlan, err := planTopLevelJSONPatches(body, patches)
	if err != nil {
		return nil, err
	}
	return patchPlan.apply(), nil
}

// selectProviderOperation chooses an operation solely from normalized request
// requirements and the selected provider profile. The order prefers the
// closest wire surface, but only an operation that preserves every declared
// requirement is eligible.
func selectProviderOperation(profile config.ConcreteProtocolProfile, ingress config.ProtocolOperation, requirements providerRequestRequirements) (providerOperationPlan, error) {
	operations := []config.ProtocolOperation{ingress}
	if ingress == config.OperationMessages {
		switch profile.Driver {
		case config.ProtocolDriverAnthropicMessages:
			operations = []config.ProtocolOperation{config.OperationMessages}
		case config.ProtocolDriverOpenAIChat:
			operations = []config.ProtocolOperation{config.OperationChatCompletions, config.OperationResponses}
		case config.ProtocolDriverOpenAIResponses:
			operations = []config.ProtocolOperation{config.OperationResponses}
		default:
			operations = nil
		}
	}

	for _, operation := range operations {
		if _, ok := profile.Operations[string(operation)]; !ok {
			continue
		}
		if required := operationRequiredCapability(operation); required != "" && !profile.HasCapability(required) {
			continue
		}
		if requirements.Stream && !profile.HasCapability("streaming") {
			continue
		}
		if requirements.HasTools && !profile.HasCapability("tool_calling") {
			continue
		}

		plan := providerOperationPlan{Operation: operation}
		mapping, hasMapping := profile.EffortMapping(operation)
		supportsTokenBudget := hasMapping && mapping.TokenBudget
		if operation == config.OperationChatCompletions && profileSupportsAnthropicThinking(profile) {
			supportsTokenBudget = true
		}
		if requirements.HasTokenBudget && !supportsTokenBudget {
			continue
		}
		if requirements.Effort != "" {
			mapped, ok := profile.MapReasoningEffort(operation, requirements.Effort, requirements.HasTools)
			if !ok {
				continue
			}
			plan.Effort = mapped
			plan.EffortWire = mapping.Wire
		}
		return plan, nil
	}

	return providerOperationPlan{}, fmt.Errorf("protocol profile %q cannot preserve the requested operation, tools, and reasoning controls", profile.ID)
}

func operationRequiredCapability(operation config.ProtocolOperation) string {
	switch operation {
	case config.OperationMessages:
		return "messages"
	case config.OperationChatCompletions:
		return "chat"
	case config.OperationResponses:
		return "responses"
	case config.OperationCountTokens:
		return "count_tokens"
	default:
		return ""
	}
}

func prepareAnthropicResponsesBridge(body []byte, target string, profile config.ConcreteProtocolProfile, plan providerOperationPlan) ([]byte, anthropicMessageRequest, error) {
	req, err := decodeAnthropicMessageRequest(body, true)
	if err != nil {
		return nil, req, err
	}
	if plan.Operation != config.OperationResponses || (plan.Effort != "" && plan.EffortWire != config.ProtocolEffortWireOpenAIResponses) {
		return nil, req, fmt.Errorf("selected operation has incompatible Responses effort normalization")
	}
	if req.TopK != nil {
		return nil, req, fmt.Errorf("top_k is not supported by the Responses bridge")
	}
	if req.Thinking != nil && req.Thinking.Type == "enabled" {
		return nil, req, fmt.Errorf("selected Responses profile does not support an explicit thinking token budget")
	}

	input := make([]map[string]any, 0, len(req.Messages))
	for index, message := range req.Messages {
		blocks, err := decodeAnthropicContent(message.Content)
		if err != nil {
			return nil, req, fmt.Errorf("messages[%d].content: %w", index, err)
		}
		converted, err := anthropicMessageToResponses(message.Role, blocks)
		if err != nil {
			return nil, req, fmt.Errorf("messages[%d]: %w", index, err)
		}
		input = append(input, converted...)
	}

	responses := map[string]any{
		"model":             target,
		"input":             input,
		"max_output_tokens": *req.MaxTokens,
		"stream":            req.Stream,
	}
	if system, err := decodeAnthropicSystem(req.System); err != nil {
		return nil, req, err
	} else if len(system) > 0 {
		parts := make([]string, 0, len(system))
		for _, block := range system {
			parts = append(parts, block.Text)
		}
		responses["instructions"] = strings.Join(parts, "\n")
	}
	if plan.Effort != "" {
		responses["reasoning"] = map[string]any{"effort": plan.Effort}
	}
	if req.Temperature != nil {
		responses["temperature"] = *req.Temperature
	}
	if req.TopP != nil {
		responses["top_p"] = *req.TopP
	}
	if len(req.StopSequences) > 0 {
		return nil, req, fmt.Errorf("stop_sequences is not supported by the Responses bridge")
	}
	if len(req.Tools) > 0 {
		tools := make([]map[string]any, 0, len(req.Tools))
		for _, tool := range req.Tools {
			var schema map[string]any
			if err := json.Unmarshal(tool.InputSchema, &schema); err != nil {
				return nil, req, fmt.Errorf("tool %q input_schema is invalid", tool.Name)
			}
			converted := map[string]any{"type": "function", "name": tool.Name, "parameters": schema}
			if tool.Description != "" {
				converted["description"] = tool.Description
			}
			tools = append(tools, converted)
		}
		responses["tools"] = tools
	}
	if req.ToolChoice != nil {
		choice, parallel := anthropicToolChoiceToResponses(*req.ToolChoice)
		responses["tool_choice"] = choice
		if parallel != nil {
			responses["parallel_tool_calls"] = *parallel
		}
	}
	if req.OutputConfig != nil && req.OutputConfig.Format != nil {
		var schema map[string]any
		if err := json.Unmarshal(req.OutputConfig.Format.Schema, &schema); err != nil {
			return nil, req, fmt.Errorf("decode output_config.format.schema: %w", err)
		}
		responses["text"] = map[string]any{"format": map[string]any{
			"type": "json_schema", "name": "response", "strict": true, "schema": schema,
		}}
	}

	encoded, err := json.Marshal(responses)
	if err != nil {
		return nil, req, fmt.Errorf("encode Responses bridge request: %w", err)
	}
	return encoded, req, nil
}

func anthropicMessageToResponses(role string, blocks []anthropicContentBlock) ([]map[string]any, error) {
	if role == "assistant" {
		out := make([]map[string]any, 0, len(blocks))
		for _, block := range blocks {
			switch block.Type {
			case "text":
				out = append(out, map[string]any{"type": "message", "role": "assistant", "content": []map[string]any{{"type": "output_text", "text": block.Text}}})
			case "thinking":
				if block.Signature != "" {
					return nil, fmt.Errorf("signed thinking blocks cannot be represented by the Responses bridge")
				}
			case "tool_use":
				callID, _, _ := decodeAnthropicBridgeToolID(block.ID)
				out = append(out, map[string]any{"type": "function_call", "call_id": callID, "name": block.Name, "arguments": string(block.Input)})
			default:
				return nil, fmt.Errorf("content block %q is invalid for an assistant message", block.Type)
			}
		}
		return out, nil
	}

	out := make([]map[string]any, 0, len(blocks))
	content := make([]map[string]any, 0, len(blocks))
	flushContent := func() {
		if len(content) > 0 {
			out = append(out, map[string]any{"type": "message", "role": "user", "content": content})
			content = nil
		}
	}
	for _, block := range blocks {
		switch block.Type {
		case "text":
			content = append(content, map[string]any{"type": "input_text", "text": block.Text})
		case "image":
			imageURL := block.Source.URL
			if block.Source.Type == "base64" {
				imageURL = "data:" + block.Source.MediaType + ";base64," + block.Source.Data
			}
			content = append(content, map[string]any{"type": "input_image", "image_url": imageURL})
		case "tool_result":
			flushContent()
			result, err := anthropicToolResultText(block.Content)
			if err != nil {
				return nil, err
			}
			if block.IsError {
				result = "Error: " + result
			}
			callID, _, _ := decodeAnthropicBridgeToolID(block.ToolUseID)
			out = append(out, map[string]any{"type": "function_call_output", "call_id": callID, "output": result})
		default:
			return nil, fmt.Errorf("content block %q is invalid for a user message", block.Type)
		}
	}
	flushContent()
	if len(out) == 0 {
		return nil, fmt.Errorf("user message content is empty")
	}
	return out, nil
}

func anthropicToolChoiceToResponses(choice anthropicToolChoice) (any, *bool) {
	parallel := !choice.DisableParallelToolUse
	switch choice.Type {
	case "auto":
		return "auto", &parallel
	case "any":
		return "required", &parallel
	case "none":
		return "none", &parallel
	default:
		return map[string]any{"type": "function", "name": choice.Name}, &parallel
	}
}

func convertResponsesResponseToAnthropic(body []byte, requestModel string) (anthropicMessageResponse, error) {
	var response openAIResponsesBridgeResponse
	if err := json.Unmarshal(body, &response); err != nil {
		return anthropicMessageResponse{}, fmt.Errorf("decode Responses response: %w", err)
	}
	if strings.TrimSpace(response.ID) == "" {
		return anthropicMessageResponse{}, fmt.Errorf("Responses response has no id")
	}
	content := make([]map[string]any, 0, len(response.Output))
	hasTools := false
	for _, item := range response.Output {
		switch item.Type {
		case "message":
			for _, part := range item.Content {
				if part.Type == "output_text" && part.Text != "" {
					content = append(content, map[string]any{"type": "text", "text": part.Text})
				}
			}
		case "reasoning":
			for _, part := range item.Summary {
				if part.Text != "" {
					content = append(content, map[string]any{"type": "thinking", "thinking": part.Text, "signature": ""})
				}
			}
		case "function_call":
			if item.CallID == "" || item.Name == "" || !json.Valid([]byte(item.Arguments)) {
				return anthropicMessageResponse{}, fmt.Errorf("Responses function call has invalid call_id, name, or arguments")
			}
			var input any
			if err := json.Unmarshal([]byte(item.Arguments), &input); err != nil {
				return anthropicMessageResponse{}, fmt.Errorf("decode Responses function arguments: %w", err)
			}
			content = append(content, map[string]any{
				"type": "tool_use", "id": encodeAnthropicBridgeToolID(item.CallID, openAIChatToolExtraContent{}),
				"name": item.Name, "input": input,
			})
			hasTools = true
		}
	}
	stopReason := "end_turn"
	if hasTools {
		stopReason = "tool_use"
	} else if response.Status == "incomplete" && response.IncompleteDetails.Reason == "max_output_tokens" {
		stopReason = "max_tokens"
	}
	model := requestModel
	if model == "" {
		model = response.Model
	}
	return anthropicMessageResponse{
		ID: response.ID, Type: "message", Role: "assistant", Model: model, Content: content,
		StopReason: &stopReason,
		Usage: anthropicUsage{
			InputTokens: response.Usage.InputTokens, OutputTokens: response.Usage.OutputTokens,
			CacheReadInputTokens: response.Usage.InputDetails.CachedTokens,
		},
	}, nil
}