ModelRouter/internal/gateway/messages.go

package gateway

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

	"modelrouter/internal/provider"
	"modelrouter/internal/scheduler"
	"modelrouter/internal/types"
)

// handleMessages serves the Anthropic Messages protocol (POST /v1/messages).
//
// The point of this endpoint is NOT to add a second scheduling path. It reuses
// everything /v1/chat/completions does — key auth, model scope, per-model
// quotas, the AUTO chain, cooldowns, audit records, plugin hooks — and differs
// in exactly two places:
//
//  1. the request body is parsed from Anthropic's shape into the unified format
//     (messages with content blocks, a top-level `system`, `max_tokens`
//     mandatory, `thinking` optional);
//  2. the response is rendered by the SERVING adapter's serialize_* hooks, so
//     the client sees Anthropic's shape back.
//
// The kernel therefore contains no Anthropic field names on this path: parsing
// is table-driven, and serialization belongs to the adapter.
//
// A source whose adapter does not declare serialize_* falls back to the
// OpenAI shape, which is what the kernel guarantees for an unconfigured
// deployment. That asymmetry is deliberate (see serializer.go): an Anthropic
// client talking to a source that cannot speak Anthropic gets a clear
// OpenAI-shaped answer rather than a malformed Anthropic one.
func (g *Gateway) handleMessages(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "use POST")
		return
	}
	body, err := io.ReadAll(r.Body)
	if err != nil {
		writeError(w, http.StatusBadRequest, "invalid_request", "read body: "+err.Error())
		return
	}
	// INGRESS: the kernel's built-in Anthropic parse is the FALLBACK (R5). If an
	// adapter has CLAIMED this ingress path (adapter.ingress == "/v1/messages"),
	// the kernel YIELDS: parsing is done by that adapter's deserialize_request
	// hook, so Anthropic protocol iteration lives in anthropic.lua, not here.
	// No adapter claims it → the built-in parser below runs (unconfigured
	// deployments keep working). This is the inbound dual of serialize_*.
	req, handled, perr := g.parseViaIngress(r.URL.Path, string(body), g.core.DefaultModel())
	if perr != nil {
		writeError(w, http.StatusBadRequest, "invalid_request", perr.Error())
		return
	}
	if !handled {
		var raw anthropicRequest
		if err := json.Unmarshal(body, &raw); err != nil {
			writeError(w, http.StatusBadRequest, "invalid_request", "invalid json: "+err.Error())
			return
		}
		req, err = raw.toUnified(g.core.DefaultModel())
		if err != nil {
			writeError(w, http.StatusBadRequest, "invalid_request", err.Error())
			return
		}
	}

	model := req.Model
	ctx := r.Context()
	g.fireStart(ctx, &chatRequest{Messages: req.Messages, Tools: nil},
		chatKind(req.Stream), model, len(req.Messages), 0)

	if isAuto(model) {
		chain, _ := g.core.AutoChainFor(reqKey(ctx))
		if chain == nil || len(chain.Tiers) == 0 {
			writeError(w, http.StatusServiceUnavailable, "no_provider", "no auto slot configured")
			return
		}
		if q := g.checkQuota(ctx, "AUTO"); q != nil {
			g.writeReject(w, q)
			return
		}
		g.serveUnified(ctx, w, chain, req, nil)
		return
	}
	if allow := g.allowedModels(ctx); allow != nil && !g.hasScopeModel(allow, model) {
		writeError(w, http.StatusForbidden, "model_not_allowed",
			"model \""+model+"\" is not allowed for this key")
		return
	}
	cands, effective := g.resolveCands(ctx, &chatRequest{Model: model})
	if len(cands) == 0 {
		writeError(w, http.StatusNotFound, "model_not_found",
			"model \""+model+"\" is not configured")
		return
	}
	if effective == "" {
		effective = firstModel(cands[0])
	}
	if q := g.checkQuota(ctx, effective); q != nil {
		g.writeReject(w, q)
		return
	}
	g.serveUnified(ctx, w, nil, req, &directRoute{cands: cands, effective: effective})
}

// directRoute carries the pre-resolved candidate list for a model-pinned
// request, so serveUnified can run both routes (direct / AUTO) from one body.
type directRoute struct {
	cands     []*provider.Provider
	effective string
}

// serveUnified runs the request through the scheduler and writes the reply via
// the serving adapter's serializer. It is the shared tail of handleChat and
// handleMessages: by this point every gate (auth, scope, quota) has already run,
// so the only remaining questions are "which source serves it" and "what shape
// does the client get".
func (g *Gateway) serveUnified(ctx context.Context, w http.ResponseWriter, chain *scheduler.Chain, req *types.ChatRequest, direct *directRoute) {
	rec := &Req{Key: keyID(reqKey(ctx)), Type: "chat", OK: false}
	if direct != nil {
		rec.Model = direct.effective
		rec.Source = firstSource(direct.cands)
	} else {
		rec.Type = "stream"
	}
	if req.Stream {
		rec.Type = "stream"
	}

	done := g.stats.Begin()
	defer done()

	var (
		resp               *types.UnifiedResponse
		chunks             <-chan types.UnifiedChunk
		usedSrc, usedModel string
		err                error
	)
	t0 := time.Now()
	if direct != nil {
		cands := toScheduler(direct.cands)
		if req.Stream {
			chunks, usedSrc, usedModel, err = g.core.Scheduler().ChatStream(ctx, cands, req)
		} else {
			resp, usedSrc, usedModel, err = g.core.Scheduler().Chat(ctx, cands, req)
		}
	} else {
		if req.Stream {
			chunks, usedSrc, usedModel, err = g.core.Scheduler().ChainChatStream(ctx, chain, req, nil, nil)
		} else {
			resp, usedSrc, usedModel, err = g.core.Scheduler().ChainChat(ctx, chain, req, nil, nil)
		}
	}
	rec.LatMs = time.Since(t0).Milliseconds()
	if err != nil {
		g.failChat(w, rec, err)
		g.writeRec(rec)
		return
	}
	rec.OK = true
	rec.Status = http.StatusOK
	if usedModel != "" {
		rec.Model = usedModel
	}
	rec.Source = usedSrc
	ser := g.serializerFor(usedSrc, directProviderList(direct))

	if chunks != nil {
		g.fireRouted(ctx, "stream", usedSrc, usedModel, tierForRoute(direct), true)
		rec.Prompt = estimatePromptTokens(req)
		g.pumpStream(w, ser, rec, chunks, modelLabel(req, usedModel), t0)
		return
	}
	recordChatUsage(rec, req, resp)
	g.fireRouted(ctx, "chat", usedSrc, usedModel, tierForRoute(direct), false)
	rec.FirstByteMs = rec.LatMs
	g.writeRec(rec)
	writeChatCompletion(w, ser, resp, modelLabel(req, usedModel))
}

// ---- Anthropic request shape (in-bound only; the kernel holds no outbound
// Anthropic field names) ----

// anthropicRequest is the client-facing parse target for POST /v1/messages.
//
// Only the fields the gateway must understand to schedule and account for a
// request appear here. Anything else the client sent is preserved verbatim in
// raw so an adapter can read it (the same "kernel does not interpret" rule that
// governs tools).
type anthropicRequest struct {
	Model     string             `json:"model"`
	System    json.RawMessage    `json:"system,omitempty"`
	Messages  []anthropicMessage `json:"messages"`
	MaxTokens int                `json:"max_tokens"`
	Stream    bool               `json:"stream,omitempty"`

	Temperature *float64               `json:"temperature,omitempty"`
	TopP        *float64               `json:"top_p,omitempty"`
	TopK        int                    `json:"top_k,omitempty"`
	StopSeqs    []string               `json:"stop_sequences,omitempty"`
	Tools       json.RawMessage        `json:"tools,omitempty"`
	ToolChoice  json.RawMessage        `json:"tool_choice,omitempty"`
	Thinking    *types.ThinkingConfig  `json:"thinking,omitempty"`
	Metadata    map[string]interface{} `json:"metadata,omitempty"`

	// raw is the untouched body, so an adapter can read fields the gateway
	// has no reason to know (cache_control, service_tier, …).
	raw json.RawMessage
}

// anthropicMessage accepts Anthropic's content as either a plain string or an
// array of typed blocks; both normalize to the unified message shape.
type anthropicMessage struct {
	Role    string          `json:"role"`
	Content json.RawMessage `json:"content"`
}

// toUnified converts the parsed request into the internal format.
func (a *anthropicRequest) toUnified(defaultModel string) (*types.ChatRequest, error) {
	if len(a.Messages) == 0 {
		return nil, errors.New("messages is required")
	}
	msgs := make([]types.ChatMessage, 0, len(a.Messages))
	for i, m := range a.Messages {
		u := types.ChatMessage{Role: m.Role, Content: normalizeAnthropicContent(m.Content)}
		if u.Role == "" {
			return nil, fmt.Errorf("messages[%d].role is required", i)
		}
		msgs = append(msgs, u)
	}
	model := a.Model
	if model == "" {
		model = defaultModel
	}
	th := a.Thinking
	if th == nil {
		// Anthropic's default is thinking OFF, and `thinking: {enabled:false}`
		// is the explicit spelling of that. Emitting it keeps the unified
		// request honest about the client's intent instead of leaving the
		// field absent and letting each adapter guess.
		no := false
		th = &types.ThinkingConfig{Enabled: &no}
	}
	return &types.ChatRequest{
		Model:         model,
		Messages:      msgs,
		Temperature:   a.Temperature,
		MaxTokens:     a.MaxTokens,
		Stream:        a.Stream,
		TopP:          a.TopP,
		TopK:          a.TopK,
		StopSequences: a.StopSeqs,
		Tools:         a.Tools,
		ToolChoice:    a.ToolChoice,
		System:        systemToString(a.System),
		Thinking:      th,
		ExtraBody:     map[string]interface{}{},
	}, nil
}

// normalizeAnthropicContent turns Anthropic content into the unified shape.
//
// A string stays a string. An array of blocks is preserved as-is: the unified
// format deliberately carries multimodal/tool blocks verbatim (ChatMessage
// .Content is json.RawMessage) rather than flattening them, because only an
// adapter knows which block types its upstream accepts.
func normalizeAnthropicContent(raw json.RawMessage) json.RawMessage {
	trimmed := strings.TrimSpace(string(raw))
	if trimmed == "" {
		return types.StringContent("")
	}
	if strings.HasPrefix(trimmed, `"`) {
		return raw
	}
	return raw
}

// systemToString accepts Anthropic's two spellings: a plain string, or an array
// of {type:"text", text} blocks.
func systemToString(raw json.RawMessage) string {
	trimmed := strings.TrimSpace(string(raw))
	if trimmed == "" || trimmed == "null" {
		return ""
	}
	if strings.HasPrefix(trimmed, `"`) {
		var s string
		if err := json.Unmarshal(raw, &s); err == nil {
			return s
		}
		return ""
	}
	var blocks []struct {
		Type string `json:"type"`
		Text string `json:"text"`
	}
	if err := json.Unmarshal(raw, &blocks); err != nil {
		return ""
	}
	var sb strings.Builder
	for _, b := range blocks {
		if b.Type == "text" {
			sb.WriteString(b.Text)
		}
	}
	return sb.String()
}

// parseViaIngress is the kernel's INBOUND yield point, the dual of
// serializerFor (which decides the outbound serializer).
//
// Design (user requirement): an inbound endpoint such as /v1/messages is the
// KERNEL'S FALLBACK — the built-in parser in handleMessages is what an
// unconfigured deployment gets. But an adapter may CLAIM that endpoint
// (adapter.ingress == path) and declare deserialize_request; when it does, the
// kernel YIELDS parsing to it, so client-protocol iteration lives in the
// adapter and the kernel's own copy is only the safety net (R5).
//
// Returns handled=false when no adapter claims the path (or one claims it
// without a deserialize_request hook) — the caller then runs its built-in
// parser. A hook that exists but FAILS is a hard error, not a silent fallback:
// the client spoke a protocol the adapter claimed to handle, and quietly
// parsing it as a different protocol would mis-schedule the request.
func (g *Gateway) parseViaIngress(path, body, defaultModel string) (*types.ChatRequest, bool, error) {
	vm := g.core.VM()
	if vm == nil {
		return nil, false, nil
	}
	adapter := vm.AdapterForIngress(path)
	if adapter == "" || !vm.HasFn(adapter, "deserialize_request") {
		return nil, false, nil
	}
	out, err := vm.Transform(adapter, "deserialize_request", body)
	if err != nil {
		return nil, true, fmt.Errorf("adapter %s deserialize_request(%s): %w", adapter, path, err)
	}
	var req types.ChatRequest
	if err := json.Unmarshal([]byte(out), &req); err != nil {
		return nil, true, fmt.Errorf("adapter %s deserialize_request returned an invalid unified request: %w", adapter, err)
	}
	if req.Model == "" {
		req.Model = defaultModel
	}
	return &req, true, nil
}

func chatKind(stream bool) string {
	if stream {
		return "stream"
	}
	return "chat"
}

// directProviderList hands the serving-source resolver its candidate list.
func directProviderList(direct *directRoute) []*provider.Provider {
	if direct == nil {
		return nil
	}
	return direct.cands
}

func tierForRoute(direct *directRoute) int {
	if direct != nil {
		return -1
	}
	return -2
}

func modelLabel(req *types.ChatRequest, usedModel string) string {
	if usedModel != "" {
		return usedModel
	}
	return req.Model
}