Cyrene_v2/internal/runtime/model/openai.go

package model

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"strings"
	"time"

	"git.yeij.top/AskaEth/Cyrene_v2/internal/config"
)

// Wire constraints for the OpenAI-compatible /chat/completions protocol.
const (
	// sseDataPrefix marks an SSE event line carrying a JSON payload.
	sseDataPrefix = "data: "
	// sseDone is the terminal SSE sentinel.
	sseDone = "[DONE]"
	// sseBufInit is the initial bufio.Scanner read buffer size.
	sseBufInit = 64 * 1024
	// maxSSELineBytes caps a single SSE event line.
	maxSSELineBytes = 1024 * 1024
	// errBodyCap bounds how much of an error body is read for the message.
	errBodyCap = 512
	// defaultHTTPTimeout bounds a single provider call when Options.Timeout is 0.
	defaultHTTPTimeout = 180 * time.Second
)

// openAIChatRequest is the JSON body sent to /chat/completions. Field names
// match the provider protocol so the struct marshals directly.
type openAIChatRequest struct {
	Model         string            `json:"model"`
	Messages      []openAIMessage   `json:"messages"`
	Temperature   *float64          `json:"temperature,omitempty"`
	MaxTokens     int               `json:"max_tokens,omitempty"`
	Stream        bool              `json:"stream,omitempty"`
	StreamOptions *openAIStreamOpts `json:"stream_options,omitempty"`
	Tools         []openAITool      `json:"tools,omitempty"`
}

// openAIStreamOpts asks the provider to include usage in the final stream chunk.
type openAIStreamOpts struct {
	IncludeUsage bool `json:"include_usage"`
}

// openAIMessage is one request message in the wire body. Content is either a
// plain string (text-only) or an array of {type,text|image_url} parts.
type openAIMessage struct {
	Role    string `json:"role"`
	Content any    `json:"content"`
}

// openAITool is the wire form of a function tool schema (M2 placeholder).
type openAITool struct {
	Type     string       `json:"type"`
	Function openAIToolFn `json:"function"`
}

// openAIToolFn carries the function definition.
type openAIToolFn struct {
	Name        string `json:"name"`
	Description string `json:"description,omitempty"`
	Parameters  any    `json:"parameters,omitempty"`
}

// openAIChatResponse is the JSON body returned by /chat/completions, for both
// the non-streaming payload and each SSE stream chunk.
type openAIChatResponse struct {
	ID      string         `json:"id"`
	Model   string         `json:"model"`
	Choices []openAIChoice `json:"choices"`
	Usage   *openAIUsage   `json:"usage,omitempty"`
}

// openAIChoice is one candidate completion.
type openAIChoice struct {
	Index        int                   `json:"index"`
	Message      openAIResponseMessage `json:"message"`
	Delta        *openAIDelta          `json:"delta,omitempty"`
	FinishReason string                `json:"finish_reason,omitempty"`
}

// openAIResponseMessage is the assistant message inside a non-streaming choice.
type openAIResponseMessage struct {
	Role    string `json:"role"`
	Content string `json:"content"`
}

// openAIDelta is the incremental payload inside a stream chunk. Reasoning content
// (OpenAI o-series / DeepSeek / DashScope) arrives in ReasoningContent; streaming
// function calls arrive in ToolCalls, accumulated per fragment index.
type openAIDelta struct {
	Role             string                `json:"role"`
	Content          string                `json:"content"`
	ReasoningContent string                `json:"reasoning_content"`
	ToolCalls        []openAIToolCallDelta `json:"tool_calls"`
}

// openAIToolCallDelta is one streaming function-call fragment. OpenAI emits a
// sequence of these sharing the same Index: the ID and Function.Name usually
// arrive once (in the first fragment) while Function.Arguments streams across
// fragments.
type openAIToolCallDelta struct {
	Index    int              `json:"index"`
	ID       string           `json:"id"`
	Function openAIToolCallFn `json:"function"`
}

// openAIToolCallFn is the function part of a streaming tool-call fragment.
type openAIToolCallFn struct {
	Name      string `json:"name"`
	Arguments string `json:"arguments"`
}

// openAIToolAcc accumulates the partial state of one streaming tool call (keyed
// by the provider's tool_calls[].index) so its arguments re-assemble across the
// delta fragments. started guards a single ToolCallStart emission per block.
type openAIToolAcc struct {
	id      string
	name    string
	args    strings.Builder
	started bool
}

// idOrName resolves the block identifier used to correlate start/delta/end: the
// provider tool-call id when present, else the function name.
func (a *openAIToolAcc) idOrName() string {
	if a.id != "" {
		return a.id
	}
	return a.name
}

// openAIUsage reports token consumption.
type openAIUsage struct {
	PromptTokens     int `json:"prompt_tokens"`
	CompletionTokens int `json:"completion_tokens"`
	TotalTokens      int `json:"total_tokens"`
}

// OpenAIChatAdapter calls an OpenAI-compatible /chat/completions endpoint. It is
// constructed from a config.LLMConfig and implements Adapter for both
// non-streaming and token-level SSE streaming.
type OpenAIChatAdapter struct {
	name     string
	baseURL  string
	apiKey   string
	model    string
	client   *http.Client
	recorder Recorder
}

// OpenAIOption configures an OpenAIChatAdapter.
type OpenAIOption func(*OpenAIChatAdapter)

// WithHTTPClient overrides the default HTTP client (used to inject mocks or a
// tuned transport in tests and production).
func WithHTTPClient(c *http.Client) OpenAIOption {
	return func(a *OpenAIChatAdapter) { a.client = c }
}

// WithRecorder wires a per-call Recorder for CallRecord accounting.
func WithRecorder(r Recorder) OpenAIOption {
	return func(a *OpenAIChatAdapter) { a.recorder = r }
}

// WithModel overrides the adapter's default model name.
func WithModel(m string) OpenAIOption {
	return func(a *OpenAIChatAdapter) { a.model = m }
}

// NewOpenAIChat builds an adapter from a supplier config. baseURL is trimmed of
// any trailing slash before the /chat/completions path is appended.
func NewOpenAIChat(cfg config.LLMConfig, opts ...OpenAIOption) *OpenAIChatAdapter {
	a := &OpenAIChatAdapter{
		name:    "openai",
		baseURL: strings.TrimRight(cfg.BaseURL, "/"),
		apiKey:  cfg.APIKey,
		model:   cfg.Model,
		client: &http.Client{
			Transport: http.DefaultTransport,
			Timeout:   defaultHTTPTimeout,
		},
	}
	for _, o := range opts {
		o(a)
	}
	return a
}

// Name returns the adapter identifier.
func (a *OpenAIChatAdapter) Name() string { return a.name }

// Chat performs a non-streaming completion.
func (a *OpenAIChatAdapter) Chat(ctx context.Context, req Request, opts Options) (Response, error) {
	start := time.Now()
	if a.baseURL == "" {
		return Response{}, ErrUpstream
	}
	ctx, cancel := withCallTimeout(ctx, opts)
	defer cancel()
	model, err := a.resolveModel(req)
	if err != nil {
		return Response{}, err
	}
	req.Model = model
	body, err := a.buildRequest(req, false)
	if err != nil {
		return Response{}, err
	}
	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, a.endpoint(), bytes.NewReader(body))
	if err != nil {
		return Response{}, fmt.Errorf("model: %w: build request: %v", ErrUpstream, err)
	}
	resp, err := a.doRequest(httpReq, false)
	if err != nil {
		return Response{}, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return Response{}, a.upstreamError(resp)
	}
	var wire openAIChatResponse
	if err := json.NewDecoder(resp.Body).Decode(&wire); err != nil {
		return Response{}, fmt.Errorf("model: %w: decode response: %v", ErrUpstream, err)
	}
	out := responseFromWire(&wire)
	a.record(req, start, out.TotalTokens)
	return out, nil
}

// ChatStream streams structured events via onEvent and returns the aggregated
// response with usage when the provider reports it in the final chunk.
func (a *OpenAIChatAdapter) ChatStream(ctx context.Context, req Request, onEvent func(StreamEvent)) (Response, error) {
	start := time.Now()
	if a.baseURL == "" {
		return Response{}, ErrUpstream
	}
	model, err := a.resolveModel(req)
	if err != nil {
		return Response{}, err
	}
	req.Model = model
	body, err := a.buildRequest(req, true)
	if err != nil {
		return Response{}, err
	}
	httpReq, err := http.NewRequestWithContext(ctx, http.MethodPost, a.endpoint(), bytes.NewReader(body))
	if err != nil {
		return Response{}, fmt.Errorf("model: %w: build request: %v", ErrUpstream, err)
	}
	resp, err := a.doRequest(httpReq, true)
	if err != nil {
		return Response{}, err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return Response{}, a.upstreamError(resp)
	}
	out, err := scanSSE(resp.Body, onEvent)
	if err != nil {
		return Response{}, err
	}
	out.Model = model
	a.record(req, start, out.TotalTokens)
	return out, nil
}

// withCallTimeout wraps ctx with opts.Timeout when it is positive.
func withCallTimeout(parent context.Context, opts Options) (context.Context, context.CancelFunc) {
	if opts.Timeout > 0 {
		return context.WithTimeout(parent, opts.Timeout)
	}
	return parent, func() {}
}

// resolveModel returns the effective model name.
func (a *OpenAIChatAdapter) resolveModel(req Request) (string, error) {
	if req.Model != "" {
		return req.Model, nil
	}
	if a.model != "" {
		return a.model, nil
	}
	return "", ErrNoModel
}

// endpoint returns the full /chat/completions URL.
func (a *OpenAIChatAdapter) endpoint() string {
	return a.baseURL + "/chat/completions"
}

func (a *OpenAIChatAdapter) buildRequest(req Request, stream bool) ([]byte, error) {
	msgs, err := buildMessages(req.Messages)
	if err != nil {
		return nil, err
	}
	wire := openAIChatRequest{
		Model:       req.Model,
		Messages:    msgs,
		Temperature: req.Temperature,
		MaxTokens:   req.MaxTokens,
		Stream:      stream,
		Tools:       buildTools(req.Tools),
	}
	if stream {
		wire.StreamOptions = &openAIStreamOpts{IncludeUsage: true}
	}
	buf, err := json.Marshal(wire)
	if err != nil {
		return nil, fmt.Errorf("model: %w: marshal request: %v", ErrBadRequest, err)
	}
	return buf, nil
}

func buildMessages(msgs []Message) ([]openAIMessage, error) {
	out := make([]openAIMessage, 0, len(msgs))
	for _, m := range msgs {
		content, err := encodeContent(m.Content)
		if err != nil {
			return nil, err
		}
		out = append(out, openAIMessage{Role: string(m.Role), Content: content})
	}
	return out, nil
}

func encodeContent(parts []Content) (any, error) {
	if len(parts) == 0 {
		return "", nil
	}
	if len(parts) == 1 && parts[0].Type == ContentTypeText {
		return parts[0].Text, nil
	}
	arr := make([]any, 0, len(parts))
	for _, p := range parts {
		switch p.Type {
		case ContentTypeText:
			arr = append(arr, map[string]any{"type": "text", "text": p.Text})
		case ContentTypeImage:
			arr = append(arr, map[string]any{
				"type":      "image_url",
				"image_url": map[string]any{"url": p.ImageURL},
			})
		default:
			return nil, fmt.Errorf("model: %w: unsupported content type %q", ErrBadRequest, p.Type)
		}
	}
	return arr, nil
}

func buildTools(tools []ToolSpec) []openAITool {
	if len(tools) == 0 {
		return nil
	}
	out := make([]openAITool, 0, len(tools))
	for _, t := range tools {
		out = append(out, openAITool{Type: "function", Function: openAIToolFn{
			Name:        t.Name,
			Description: t.Description,
			Parameters:  t.Parameters,
		}})
	}
	return out
}

func (a *OpenAIChatAdapter) doRequest(httpReq *http.Request, stream bool) (*http.Response, error) {
	if a.apiKey != "" {
		httpReq.Header.Set("Authorization", "Bearer "+a.apiKey)
	}
	httpReq.Header.Set("Content-Type", "application/json")
	if stream {
		httpReq.Header.Set("Accept", "text/event-stream")
	} else {
		httpReq.Header.Set("Accept", "application/json")
	}
	resp, err := a.client.Do(httpReq)
	if err != nil {
		return nil, fmt.Errorf("model: %w: %v", ErrUpstream, err)
	}
	return resp, nil
}

func (a *OpenAIChatAdapter) upstreamError(resp *http.Response) error {
	snippet, _ := io.ReadAll(io.LimitReader(resp.Body, errBodyCap))
	return fmt.Errorf("model: %w: status %d: %s", ErrUpstream, resp.StatusCode, strings.TrimSpace(string(snippet)))
}

func (a *OpenAIChatAdapter) record(req Request, start time.Time, tokens int) {
	if a.recorder == nil {
		return
	}
	a.recorder(CallRecord{
		Model:   req.Model,
		Purpose: req.Purpose,
		Tokens:  tokens,
		TookMs:  time.Since(start).Milliseconds(),
	})
}

func responseFromWire(wire *openAIChatResponse) Response {
	out := Response{Model: wire.Model}
	if len(wire.Choices) > 0 {
		text := wire.Choices[0].Message.Content
		out.Content = text
		out.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: text}}}
	}
	if u := wire.Usage; u != nil {
		out.PromptTokens = u.PromptTokens
		out.CompletionTokens = u.CompletionTokens
		out.TotalTokens = u.TotalTokens
	}
	return out
}

// scanSSE incrementally decodes stream chunks, accumulates the full body text
// with a strings.Builder (never reading the whole body at once) and captures
// usage from the final chunk. It splits each delta into its segment and brackets
// them: reasoning_content as thinking_start/delta/end, tool_calls fragments as
// tool_call_start/delta/end, and body text as text_start/delta/end. The
// accumulated tool-call arguments re-assemble across fragments into a structured
// Args map, and are attached to the returned Response alongside the text.
func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) {
	scanner := bufio.NewScanner(r)
	scanner.Buffer(make([]byte, 0, sseBufInit), maxSSELineBytes)
	agg := Response{}
	var text strings.Builder
	textStarted := false
	thinkingStarted := false
	toolCalls := map[int]*openAIToolAcc{}
	var toolOrder []int

	endThinking := func() {
		if onEvent != nil && thinkingStarted {
			thinkingStarted = false
			onEvent(StreamEvent{Kind: StreamKindThinkingEnd})
		}
	}

	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" || !strings.HasPrefix(line, sseDataPrefix) {
			continue
		}
		data := strings.TrimSpace(strings.TrimPrefix(line, sseDataPrefix))
		if data == sseDone {
			break
		}
		var chunk openAIChatResponse
		if err := json.Unmarshal([]byte(data), &chunk); err != nil {
			return Response{}, fmt.Errorf("model: %w: decode stream chunk: %v", ErrUpstream, err)
		}
		if u := chunk.Usage; u != nil {
			agg.PromptTokens = u.PromptTokens
			agg.CompletionTokens = u.CompletionTokens
			agg.TotalTokens = u.TotalTokens
		}
		if len(chunk.Choices) == 0 || chunk.Choices[0].Delta == nil {
			continue
		}
		delta := chunk.Choices[0].Delta

		// Reasoning / thinking segment (OpenAI o-series, DeepSeek, DashScope).
		if delta.ReasoningContent != "" {
			if onEvent != nil && !thinkingStarted {
				thinkingStarted = true
				onEvent(StreamEvent{Kind: StreamKindThinkingStart})
			}
			if onEvent != nil {
				onEvent(StreamEvent{Kind: StreamKindThinkingDelta, Delta: delta.ReasoningContent})
			}
			continue
		}

		// Streaming tool-call fragments.
		if len(delta.ToolCalls) > 0 {
			endThinking()
			for _, tc := range delta.ToolCalls {
				acc := toolCalls[tc.Index]
				if acc == nil {
					acc = &openAIToolAcc{}
					toolCalls[tc.Index] = acc
					toolOrder = append(toolOrder, tc.Index)
				}
				if acc.id == "" {
					acc.id = tc.ID
				}
				if acc.name == "" {
					acc.name = tc.Function.Name
				}
				if onEvent != nil && !acc.started && acc.idOrName() != "" {
					acc.started = true
					onEvent(StreamEvent{Kind: StreamKindToolCallStart, Index: tc.Index, ID: acc.idOrName()})
				}
				if tc.Function.Arguments != "" {
					acc.args.WriteString(tc.Function.Arguments)
					if onEvent != nil {
						onEvent(StreamEvent{Kind: StreamKindToolCallDelta, Index: tc.Index, ID: acc.idOrName(), Delta: tc.Function.Arguments})
					}
				}
			}
			continue
		}

		// Body-text segment.
		if delta.Content == "" {
			continue
		}
		endThinking()
		text.WriteString(delta.Content)
		if onEvent != nil {
			if !textStarted {
				textStarted = true
				onEvent(StreamEvent{Kind: StreamKindTextStart})
			}
			onEvent(StreamEvent{Kind: StreamKindTextDelta, Delta: delta.Content})
		}
	}
	if err := scanner.Err(); err != nil {
		return Response{}, fmt.Errorf("model: %w: read stream: %v", ErrUpstream, err)
	}

	if onEvent != nil {
		if thinkingStarted {
			onEvent(StreamEvent{Kind: StreamKindThinkingEnd})
		}
		if textStarted {
			onEvent(StreamEvent{Kind: StreamKindTextEnd})
		}
		// OpenAI has no per-block stop event, so each tool_call block is closed
		// when the stream terminates.
		for _, idx := range toolOrder {
			acc := toolCalls[idx]
			if acc == nil || !acc.started {
				continue
			}
			onEvent(StreamEvent{Kind: StreamKindToolCallEnd, Index: idx, ID: acc.idOrName()})
		}
	}

	for _, idx := range toolOrder {
		acc := toolCalls[idx]
		if acc == nil {
			continue
		}
		args := map[string]any{}
		if acc.args.Len() > 0 {
			if err := json.Unmarshal([]byte(acc.args.String()), &args); err != nil {
				args = map[string]any{}
			}
		}
		agg.ToolCalls = append(agg.ToolCalls, ToolCall{Name: acc.name, Args: args})
	}
	agg.Content = text.String()
	agg.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: agg.Content}}}
	return agg, nil
}