Cyrene_v2/internal/runtime/model/openai.go

package model

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"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
)

// 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
	// reasoningKind is this endpoint's configured reasoning-switch shape
	// (models.json group reasoning.kind / LLM_REASONING_KIND, 裁决 03.4.2/08.4.1).
	// Empty means "auto": infer the vendor from the model name/base URL and send
	// nothing when the shape is ambiguous.
	reasoningKind string
	// log reports reasoning_effort validation and ignored thinking directives. It
	// is never nil: a discard logger tagged with this adapter's source is
	// installed when the caller supplies none, so a decision is always traceable.
	log *slog.Logger
}

// 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 }
}

// WithOpenAILogger attaches the logger used to report reasoning_effort
// validation and ignored thinking directives. A nil logger is ignored so the
// discard default stays in place.
func WithOpenAILogger(l *slog.Logger) OpenAIOption {
	return func(a *OpenAIChatAdapter) {
		if l != nil {
			a.log = l
		}
	}
}

// NewOpenAIChat builds an adapter from a supplier config. baseURL is trimmed of
// any trailing slash before the /chat/completions path is appended. The
// endpoint's configured reasoning shape (cfg.ReasoningKind) travels with the
// adapter so every chain built from the same config — the group's chain, its
// per-persona chain and the local rung — resolves its reasoning switches the
// same way.
func NewOpenAIChat(cfg config.LLMConfig, opts ...OpenAIOption) *OpenAIChatAdapter {
	a := &OpenAIChatAdapter{
		name:          "openai",
		baseURL:       strings.TrimRight(cfg.BaseURL, "/"),
		apiKey:        cfg.APIKey,
		model:         cfg.Model,
		reasoningKind: strings.TrimSpace(cfg.ReasoningKind),
		client: &http.Client{
			Transport: http.DefaultTransport,
			Timeout:   defaultHTTPTimeout,
		},
		log: discardLogger("runtime/model/openai"),
	}
	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) 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(),
	})
}