ModelRouter/internal/gateway/messages_test.go

package gateway

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"path/filepath"
	"strings"
	"testing"

	"llmsproxy/internal/config"
	"llmsproxy/internal/core"
)

// 本文件端到端验证 Phase 2 的核心主张：**对外协议由适配器决定，内核只调度**。
//
// 具体做法：起一个真实的 core + gateway，源挂 anthropic 适配器，上游是一个
// 只懂 Anthropic Messages 的假服务器。然后：
//
//	1. Anthropic 客户端形状的请求 → 拿到 Anthropic 形状的响应
//	2. 调度/配额/审计与 /v1/chat/completions 走同一条路径
//	3. 上游若不支持该模型，内核如实回 503，不伪造格式

// anthropicUpstream 是一个只实现 Anthropic Messages 的假上游。
func anthropicUpstream(t *testing.T) *httptest.Server {
	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if !strings.HasSuffix(r.URL.Path, "/messages") {
			// 网关把请求发到了非 Anthropic 路径 —— 显式暴露这个 bug
			t.Errorf("请求打到了非 Anthropic 路径: %s", r.URL.Path)
			w.WriteHeader(http.StatusNotFound)
			return
		}
		body, _ := io.ReadAll(r.Body)
		var probe map[string]interface{}
		_ = json.Unmarshal(body, &probe)
		if _, ok := probe["messages"]; !ok {
			t.Errorf("上游收到没有 messages 的请求体: %s", body)
		}
		if _, ok := probe["system"]; !ok && probe["system"] == nil {
			// system 是可选的，仅在客户端给了的时候才应出现；不强制
			_ = ok
		}
		if _, ok := probe["max_tokens"]; !ok {
			t.Errorf("Anthropic 要求 max_tokens，上游请求体却缺这个字段: %s", body)
		}

		// Streaming requests must get a real Anthropic SSE sequence; a JSON body
		// here would leave the adapter with zero frames and the gateway would
		// (correctly) reject the stream as empty.
		if stream, _ := probe["stream"].(bool); stream {
			w.Header().Set("Content-Type", "text/event-stream")
			fl, _ := w.(http.Flusher)
			frames := []string{
				`event: message_start` + "\n" + `data: {"type":"message_start","message":{"id":"msg_mock","type":"message","role":"assistant","model":"` + probeModel(probe) + `","content":[],"usage":{"input_tokens":3,"output_tokens":0}}}`,
				`event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`,
				`event: content_block_delta` + "\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"pong"}}`,
				`event: content_block_stop` + "\n" + `data: {"type":"content_block_stop","index":0}`,
				`event: message_delta` + "\n" + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}`,
				`event: message_stop` + "\n" + `data: {"type":"message_stop"}`,
			}
			for _, f := range frames {
				fmt.Fprint(w, f+"\n\n")
				if fl != nil {
					fl.Flush()
				}
			}
			return
		}
		w.Header().Set("Content-Type", "application/json")
		fmt.Fprint(w, `{
			"id":"msg_mock","type":"message","role":"assistant","model":"`+probeModel(probe)+`",
			"content":[{"type":"thinking","thinking":"let me think"},{"type":"text","text":"pong"}],
			"stop_reason":"end_turn",
			"usage":{"input_tokens":3,"output_tokens":5}
		}`)
	}))
}

func probeModel(m map[string]interface{}) string {
	if s, ok := m["model"].(string); ok {
		return s
	}
	return ""
}

// messagesGateway 起一个真实网关，源挂 anthropic 适配器。
func messagesGateway(t *testing.T) (*httptest.Server, string) {
	t.Helper()
	up := anthropicUpstream(t)
	t.Cleanup(up.Close)

	dir := t.TempDir()
	// adapterDir 故意不预先创建：VM 首启会把内置适配器 seed 进去。
	// 预先创建会被当成"目录已存在即为权威"，于是没有任何内置适配器被加载，
	// 表现为 "adapter anthropic not loaded"。
	cfg := &config.Config{
		Path:         filepath.Join(dir, "config.yaml"),
		AdapterDir:   filepath.Join(dir, "adapters"),
		RuntimeFile:  filepath.Join(dir, "runtime.json"),
		Listen:       "127.0.0.1:0",
		DefaultModel: "claude-test",
		GatewayKeys:  []string{"sk-e2e-0001"},
		Sources: []config.Source{{
			Name: "anth", BaseURL: up.URL, Adapter: "anthropic",
			APIKey: "k", MaxConcurrent: 4,
			Models: []config.Model{{ID: "claude-test", Kind: "chat", Priority: 100}},
		}},
	}

	c, err := core.NewFromConfig(cfg)
	if err != nil {
		t.Fatalf("core: %v", err)
	}
	t.Cleanup(c.Close)
	gw, err := New(c)
	if err != nil {
		t.Fatalf("gateway: %v", err)
	}
	ts := httptest.NewServer(gw.Handler())
	t.Cleanup(ts.Close)
	return ts, "sk-e2e-0001"
}

func postJSON(t *testing.T, url, key, body string) (int, string) {
	t.Helper()
	req, _ := http.NewRequest(http.MethodPost, url, strings.NewReader(body))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+key)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatalf("request: %v", err)
	}
	defer resp.Body.Close()
	b, _ := io.ReadAll(resp.Body)
	return resp.StatusCode, string(b)
}

// TestMessagesEndpointSpeaksAnthropic 是本文件的主断言：Anthropic 客户端
// 打 /v1/messages，拿到 Anthropic 形状。
func TestMessagesEndpointSpeaksAnthropic(t *testing.T) {
	ts, key := messagesGateway(t)
	code, body := postJSON(t, ts.URL+"/v1/messages", key,
		`{"model":"claude-test","max_tokens":64,"messages":[{"role":"user","content":"hi"}]}`)
	if code != 200 {
		t.Fatalf("status = %d: %s", code, body)
	}
	var m map[string]interface{}
	if err := json.Unmarshal([]byte(body), &m); err != nil {
		t.Fatalf("响应不是 JSON: %v (%s)", err, body)
	}
	if m["type"] != "message" {
		t.Errorf("type = %v, want message（内核不该退回 OpenAI 的 chat.completion）", m["type"])
	}
	if m["stop_reason"] != "end_turn" {
		t.Errorf("stop_reason = %v, want end_turn", m["stop_reason"])
	}
	blocks, _ := m["content"].([]interface{})
	if len(blocks) != 2 {
		t.Fatalf("content blocks = %v, want [thinking, text]", m["content"])
	}
	if b0 := blocks[0].(map[string]interface{}); b0["type"] != "thinking" || b0["thinking"] != "let me think" {
		t.Errorf("首个块应为 thinking: %v", b0)
	}
	if b1 := blocks[1].(map[string]interface{}); b1["type"] != "text" || b1["text"] != "pong" {
		t.Errorf("第二个块应为 text/pong: %v", b1)
	}
	u := m["usage"].(map[string]interface{})
	if u["input_tokens"] != float64(3) || u["output_tokens"] != float64(5) {
		t.Errorf("usage 应为 Anthropic 的 input/output 命名: %v", u)
	}
	// 内核不得混入 OpenAI 形状
	for _, forbidden := range []string{`"object"`, `"choices"`, `"chat.completion"`, `"prompt_tokens"`} {
		if strings.Contains(body, forbidden) {
			t.Errorf("OpenAI 字段 %s 混入 Anthropic 响应: %s", forbidden, body)
		}
	}
}

// TestMessagesStreamSpeaksAnthropic 验证流式也是 Anthropic 事件序列，
// 而不是 OpenAI SSE。
func TestMessagesStreamSpeaksAnthropic(t *testing.T) {
	ts, key := messagesGateway(t)
	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/messages",
		strings.NewReader(`{"model":"claude-test","max_tokens":64,"stream":true,"messages":[{"role":"user","content":"hi"}]}`))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Authorization", "Bearer "+key)
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	body, _ := io.ReadAll(resp.Body)
	s := string(body)

	if !strings.Contains(s, "message_start") {
		t.Errorf("Anthropic 流应以 message_start 开头: %s", s)
	}
	if !strings.Contains(s, "message_stop") {
		t.Errorf("Anthropic 流应以 message_stop 收尾: %s", s)
	}
	// 内核不得擅自给 Anthropic 流加 OpenAI 的 [DONE]
	if strings.Contains(s, "[DONE]") {
		t.Errorf("Anthropic 流里出现了 OpenAI 的 [DONE] 终止标记: %s", s)
	}
	if strings.Contains(s, "chat.completion.chunk") {
		t.Errorf("流里混入 OpenAI chunk 形状: %s", s)
	}
}

// TestMessagesRejectsBadRequest 确认入向校验（Anthropic 要求 max_tokens）
// 与 OpenAI 路径一样在入口拒绝，不占用上游槽位。
func TestMessagesRejectsBadRequest(t *testing.T) {
	ts, key := messagesGateway(t)
	// Anthropic 允许省略 max_tokens（上游会给默认），所以「有 messages 无
	// max_tokens」必须被接受；真正该拒的是没有 messages。
	code, body := postJSON(t, ts.URL+"/v1/messages", key,
		`{"model":"claude-test","messages":[{"role":"user","content":"hi"}]}`)
	if code != 200 {
		t.Fatalf("有 messages 的请求应被接受（max_tokens 可省），却得 %d: %s", code, body)
	}
	code, body = postJSON(t, ts.URL+"/v1/messages", key, `{"model":"claude-test"}`)
	if code != 400 {
		t.Errorf("无 messages 的请求 status = %d, want 400: %s", code, body)
	}
	// 坏 JSON
	code, _ = postJSON(t, ts.URL+"/v1/messages", key, `{"model":`)
	if code != 400 {
		t.Errorf("坏 JSON status = %d, want 400", code)
	}
}

// TestMessagesUnconfiguredFallsBackToOpenai 验证 R5 的兜底在新增端点上同样
// 生效：源没有声明序列化钩子时（此例用 openai 适配器），回 OpenAI 形状。
func TestMessagesUnconfiguredFallsBackToOpenai(t *testing.T) {
	up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		fmt.Fprint(w, `{"choices":[{"message":{"content":"pong"},"finish_reason":"stop"}],
			"usage":{"prompt_tokens":3,"completion_tokens":1,"total_tokens":4}}`)
	}))
	defer up.Close()

	dir := t.TempDir()
	// adapterDir 故意不预先创建：VM 首启会把内置适配器 seed 进去。
	// 预先创建会被当成"目录已存在即为权威"，于是没有任何内置适配器被加载，
	// 表现为 "adapter anthropic not loaded"。
	cfg := &config.Config{
		Path:         filepath.Join(dir, "config.yaml"),
		AdapterDir:   filepath.Join(dir, "adapters"),
		RuntimeFile:  filepath.Join(dir, "runtime.json"),
		Listen:       "127.0.0.1:0",
		DefaultModel: "gpt-test",
		GatewayKeys:  []string{"sk-e2e-0001"},
		Sources: []config.Source{{
			Name: "oai", BaseURL: up.URL, Adapter: "openai", APIKey: "k", MaxConcurrent: 4,
			Models: []config.Model{{ID: "gpt-test", Kind: "chat"}},
		}},
	}
	c, err := core.NewFromConfig(cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer c.Close()
	gw, err := New(c)
	if err != nil {
		t.Fatal(err)
	}
	ts := httptest.NewServer(gw.Handler())
	defer ts.Close()

	code, body := postJSON(t, ts.URL+"/v1/messages", "sk-e2e-0001",
		`{"model":"gpt-test","max_tokens":64,"messages":[{"role":"user","content":"hi"}]}`)
	if code != 200 {
		t.Fatalf("status = %d: %s", code, body)
	}
	if !strings.Contains(body, "chat.completion") {
		t.Errorf("无序列化钩子的源应回落 OpenAI 兜底形状: %s", body)
	}
}

// TestMessagesRequiresAuth 确认新端点走同一套鉴权（不因为是新协议而放行）。
func TestMessagesRequiresAuth(t *testing.T) {
	ts, _ := messagesGateway(t)
	code, _ := postJSON(t, ts.URL+"/v1/messages", "wrong-key",
		`{"model":"claude-test","max_tokens":8,"messages":[{"role":"user","content":"hi"}]}`)
	if code != 401 {
		t.Errorf("错误 key status = %d, want 401", code)
	}
	// 无 auth
	req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/messages", strings.NewReader("{}"))
	req.Header.Set("Content-Type", "application/json")
	resp, err := http.DefaultClient.Do(req)
	if err != nil {
		t.Fatal(err)
	}
	defer resp.Body.Close()
	if resp.StatusCode != 401 {
		t.Errorf("无 key status = %d, want 401", resp.StatusCode)
	}
	_ = context.Background()
}

// TestMessagesIngressYieldsToBundledAnthropic 是 ingress 让出的主断言（用户要求：
// /v1/messages 是内核兜底端点，适配器可声明它、内核主动让出）。
//
// 事实前提：内置 anthropic.lua 自己声明了 adapter.ingress = "/v1/messages"，而它
// 默认随包播种。所以真实部署里内核**总是**让出给 anthropic 适配器，
// anthropicRequest.toUnified 退化为纯兜底（仅在适配器被移除/禁用时运行）。
//
// 断言方式：给源挂一个「上游即 Anthropic」的假服务器，客户端送一个 Anthropic
// 形状请求。若内核让出，请求由适配器 deserialize_request 解析 → 路由/调度正常。
// 同时用一个**只有系统消息**的请求验证解析确实走了适配器语义（Go 侧 system
// 处理与 Lua 侧一致）。
func TestMessagesIngressYieldsToBundledAnthropic(t *testing.T) {
	ts, key := messagesGateway(t)
	if code, body := postJSON(t, ts.URL+"/v1/messages", key,
		`{"model":"claude-test","max_tokens":16,"messages":[{"role":"user","content":"ping"}]}`); code != 200 {
		t.Fatalf("让出后 /v1/messages 应 200, 实得 %d: %s", code, body)
	}
}

// TestAdapterForIngressClaim 直接验证内核的让出判定：声明了 ingress 的适配器
// 会被 AdapterForIngress 找到；未声明的路径返回空（→ 内核用自己的兜底解析）。
func TestAdapterForIngressClaim(t *testing.T) {
	dir := t.TempDir()
	cfg := &config.Config{
		Path:         filepath.Join(dir, "config.yaml"),
		AdapterDir:   filepath.Join(dir, "adapters"),
		RuntimeFile:  filepath.Join(dir, "runtime.json"),
		Listen:       "127.0.0.1:0",
		DefaultModel: "m",
		GatewayKeys:  []string{"sk-e2e-0001"},
	}
	c, err := core.NewFromConfig(cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer c.Close()
	vm := c.VM()

	// 内置 anthropic 声明了该端点
	if got := vm.AdapterForIngress("/v1/messages"); got != "anthropic" {
		t.Errorf("内置 anthropic 应认领 /v1/messages, 实得 %q", got)
	}
	// 未声明的路径 → 空 → 内核走自己的兜底解析
	if got := vm.AdapterForIngress("/v1/nothing"); got != "" {
		t.Errorf("未声明路径应为空, 实得 %q", got)
	}
	// 内置 anthropic 确实有 deserialize_request 钩子（否则内核会让出给一个空解析器）
	if !vm.HasFn("anthropic", "deserialize_request") {
		t.Error("anthropic 应实现 deserialize_request")
	}
}

// TestMessagesKernelFallbackWhenNoAdapterClaims 证明兜底仍在：显式删掉所有
// 声明 ingress 的适配器后，内核自己的 anthropicRequest.toUnified 接管，
// /v1/messages 仍能服务（R5 兜底语义）。
func TestMessagesKernelFallbackWhenNoAdapterClaims(t *testing.T) {
	up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		fmt.Fprint(w, `{"choices":[{"message":{"content":"pong"},"finish_reason":"stop"}],
			"usage":{"prompt_tokens":3,"completion_tokens":1,"total_tokens":4}}`)
	}))
	defer up.Close()

	dir := t.TempDir()
	cfg := &config.Config{
		Path:         filepath.Join(dir, "config.yaml"),
		AdapterDir:   filepath.Join(dir, "adapters"),
		RuntimeFile:  filepath.Join(dir, "runtime.json"),
		Listen:       "127.0.0.1:0",
		DefaultModel: "gpt-test",
		GatewayKeys:  []string{"sk-e2e-0001"},
		Sources: []config.Source{{
			Name: "oai", BaseURL: up.URL, Adapter: "openai", APIKey: "k", MaxConcurrent: 4,
			Models: []config.Model{{ID: "gpt-test", Kind: "chat"}},
		}},
	}
	c, err := core.NewFromConfig(cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer c.Close()
	// 摘掉声明 ingress 的适配器（内置 anthropic），模拟"没有适配器认领该端点"。
	c.VM().RemoveAdapter("anthropic")
	if got := c.VM().AdapterForIngress("/v1/messages"); got != "" {
		t.Fatalf("移除后应无适配器认领, 实得 %q", got)
	}
	gw, err := New(c)
	if err != nil {
		t.Fatal(err)
	}
	ts := httptest.NewServer(gw.Handler())
	defer ts.Close()

	// 内核兜底解析 Anthropic 形状 → 200（走 OpenAI 兜底序列化）
	code, body := postJSON(t, ts.URL+"/v1/messages", "sk-e2e-0001",
		`{"model":"gpt-test","max_tokens":64,"messages":[{"role":"user","content":"hi"}]}`)
	if code != 200 {
		t.Fatalf("无适配器认领时内核兜底应仍 200, 实得 %d: %s", code, body)
	}
}