diff --git a/internal/gateway/api.go b/internal/gateway/api.go index f7996d9..4dc75f0 100644 --- a/internal/gateway/api.go +++ b/internal/gateway/api.go @@ -470,8 +470,8 @@ func (g *Gateway) handleStatsAPI(w http.ResponseWriter, r *http.Request) { // the all-time view showed the truth, which read as "the gateway is // idle" precisely when it was busiest. A calendar window cannot // filter it, so it travels with every response unfiltered. - "active": g.stats.Active(), - "records": page.Records, + "active": g.stats.Active(), + "records": page.Records, // The paging cursor has to travel with the first page, otherwise // the scroll handler has nothing to page from and the table is // stuck at one screen forever. diff --git a/internal/gateway/discover.go b/internal/gateway/discover.go index 3c5db64..7a5efa4 100644 --- a/internal/gateway/discover.go +++ b/internal/gateway/discover.go @@ -535,11 +535,11 @@ type modelCaps struct { } type upstreamModel struct { - ID string `json:"id"` - Object string `json:"object"` - Kind string `json:"kind"` - Type string `json:"type"` - Modality string `json:"modality"` + ID string `json:"id"` + Object string `json:"object"` + Kind string `json:"kind"` + Type string `json:"type"` + Modality string `json:"modality"` Architecture struct { InputModalities []string `json:"input_modalities"` OutputModalities []string `json:"output_modalities"` @@ -699,8 +699,8 @@ func modelMeta(m upstreamModel) map[string]interface{} { } } - copyFields(m.Metadata) // nested first - copyFields(mustJSON(m)) // top level wins + copyFields(m.Metadata) // nested first + copyFields(mustJSON(m)) // top level wins if len(meta) == 0 { return nil } diff --git a/internal/gateway/discover_api.go b/internal/gateway/discover_api.go index 0840954..257dc77 100644 --- a/internal/gateway/discover_api.go +++ b/internal/gateway/discover_api.go @@ -5,10 +5,10 @@ import ( "encoding/json" "errors" "io" + "llmsproxy/internal/config" "net/http" "strings" "time" - "llmsproxy/internal/config" ) // HTTP surface for model discovery. @@ -68,9 +68,9 @@ type discoverResponse struct { OK bool `json:"ok"` // Normalized source fields, ready to be echoed back to /api/sources. - BaseURL string `json:"base_url"` - Adapter string `json:"adapter"` - Models []string `json:"models"` + BaseURL string `json:"base_url"` + Adapter string `json:"adapter"` + Models []string `json:"models"` ImageModels []string `json:"image_models"` // Endpoint is set only when discovery had to move a version segment. @@ -152,18 +152,18 @@ func (g *Gateway) handleSourceDiscoverAPI(w http.ResponseWriter, r *http.Request } out := &discoverResponse{ - OK: true, - BaseURL: res.BaseURL, - Adapter: req.Adapter, - Models: chat, - ImageModels: image, + OK: true, + BaseURL: res.BaseURL, + Adapter: req.Adapter, + Models: chat, + ImageModels: image, Endpoint: res.Endpoint, ImageEndpoint: res.ImageEndpoint, ModelsPath: res.ModelsPath, - Tried: res.Tried, - LatencyMS: res.LatencyMS, - Note: res.Note, - Warnings: warns, + Tried: res.Tried, + LatencyMS: res.LatencyMS, + Note: res.Note, + Warnings: warns, } if len(chat) == 0 && len(image) == 0 { out.Warnings = append(out.Warnings, @@ -207,14 +207,14 @@ func (g *Gateway) handleSourceDiscoverAPI(w http.ResponseWriter, r *http.Request // keep the stored one" rule, and going through a second decoder here would // duplicate (and eventually diverge from) that logic. src := config.Source{ - Name: name, - BaseURL: res.BaseURL, - APIKey: g.resolveAPIKey(name, req.APIKey, exists), - Adapter: req.Adapter, + Name: name, + BaseURL: res.BaseURL, + APIKey: g.resolveAPIKey(name, req.APIKey, exists), + Adapter: req.Adapter, Endpoint: res.Endpoint, ImageEndpoint: res.ImageEndpoint, Models: modelsToConfig(chat, image), - Temperature: req.Temperature, + Temperature: req.Temperature, } if req.MaxConcurrent > 0 { src.MaxConcurrent = req.MaxConcurrent diff --git a/internal/gateway/discover_api_test.go b/internal/gateway/discover_api_test.go index 1fa6983..71d6f03 100644 --- a/internal/gateway/discover_api_test.go +++ b/internal/gateway/discover_api_test.go @@ -32,12 +32,12 @@ func catalogUpstream(t *testing.T) *httptest.Server { func TestDefaultSourceName(t *testing.T) { cases := map[string]string{ - "https://api.vendor.com/v1": "vendor.com", - "https://api.openai.com/v1": "openai.com", - "https://www.vendor.com/v1": "vendor.com", - "http://127.0.0.1:8080/v1": "127.0.0.1", + "https://api.vendor.com/v1": "vendor.com", + "https://api.openai.com/v1": "openai.com", + "https://www.vendor.com/v1": "vendor.com", + "http://127.0.0.1:8080/v1": "127.0.0.1", "https://amr-link.open-design.ai/v1": "amr-link.open-design.ai", - "not a url at all": "", + "not a url at all": "", } for in, want := range cases { if got := defaultSourceName(in); got != want { @@ -122,16 +122,16 @@ func TestSelectModels_WarnsThatImagesAreNotInAUTO(t *testing.T) { // operator previews one set of values and saves another. func TestDiscoveryResultAndConfigRoundTrip(t *testing.T) { res := &discoveryResult{ - BaseURL: "https://h", - Endpoint: "/v1/chat/completions", - Models: []string{"m1", "m2"}, + BaseURL: "https://h", + Endpoint: "/v1/chat/completions", + Models: []string{"m1", "m2"}, ImageModels: []string{"img1"}, } src := config.Source{ - Name: "x", - BaseURL: res.BaseURL, - Endpoint: res.Endpoint, - Models: modelsToConfig(res.Models, res.ImageModels), + Name: "x", + BaseURL: res.BaseURL, + Endpoint: res.Endpoint, + Models: modelsToConfig(res.Models, res.ImageModels), } if len(src.Models) != 3 { t.Fatalf("models = %d, want 3", len(src.Models)) diff --git a/internal/gateway/discover_endpoint_test.go b/internal/gateway/discover_endpoint_test.go index 8865921..4801b6a 100644 --- a/internal/gateway/discover_endpoint_test.go +++ b/internal/gateway/discover_endpoint_test.go @@ -31,10 +31,10 @@ func TestComposeImageEndpoint_NoVersionBaseLeavesDefault(t *testing.T) { // TestDiscovery_BareHostNeedsVersionOnEndpointWhenCatalogIsVersioned // distinguishes two cases that look identical from the outside. // -// bare host + catalog at "/models" → chat is "/chat/completions" (the -// adapter default, no endpoint written) -// bare host + catalog at "/v1/models" → chat is "/v1/chat/completions", -// so an endpoint MUST be written +// bare host + catalog at "/models" → chat is "/chat/completions" (the +// adapter default, no endpoint written) +// bare host + catalog at "/v1/models" → chat is "/v1/chat/completions", +// so an endpoint MUST be written // // The catalog path is the only evidence of whether the upstream wants the // version segment, and getting this backwards sends every chat request to a @@ -105,15 +105,15 @@ func TestDiscovery_BareHostNeedsVersionOnEndpointWhenCatalogIsVersioned(t *testi // silent, expensive mistakes, so they are asserted literally. func TestComposeEndpoints_BothFormsProduceWorkingURLs(t *testing.T) { cases := []struct { - name string - base string - wantChat string - wantImage string + name string + base string + wantChat string + wantImage string }{ { - name: "bare host: adapter and image defaults compose", - base: "https://h", - wantChat: "https://h/chat/completions", + name: "bare host: adapter and image defaults compose", + base: "https://h", + wantChat: "https://h/chat/completions", // openai's adapter endpoint is /chat/completions (verified in // internal/lua/adapters/openai.lua: adapter.endpoint). wantImage: "https://h/v1/images/generations", diff --git a/internal/gateway/discover_test.go b/internal/gateway/discover_test.go index d465b43..2d18d0d 100644 --- a/internal/gateway/discover_test.go +++ b/internal/gateway/discover_test.go @@ -356,10 +356,10 @@ func TestDiscoveryCandidates_DedupesAndPrefersTyped(t *testing.T) { func TestVersionSegment(t *testing.T) { cases := map[string]string{ - "/v1/models": "/v1", + "/v1/models": "/v1", "/v1beta/models": "/v1beta", - "/models": "", - "/api/models": "", + "/models": "", + "/api/models": "", "/vision/models": "", } for in, want := range cases { @@ -371,19 +371,19 @@ func TestVersionSegment(t *testing.T) { func TestTrimVersionSuffix(t *testing.T) { cases := map[string]string{ - "https://h/v1": "https://h", - "https://h/v2": "https://h", - "https://h/v1beta": "https://h", - "https://h/api/v1": "https://h/api", - "https://h/vision": "https://h/vision", // not numeric — a real path segment - "https://h": "https://h", - "https://h/openai": "https://h/openai", - "https://h/v1/extra": "https://h/v1/extra", + "https://h/v1": "https://h", + "https://h/v2": "https://h", + "https://h/v1beta": "https://h", + "https://h/api/v1": "https://h/api", + "https://h/vision": "https://h/vision", // not numeric — a real path segment + "https://h": "https://h", + "https://h/openai": "https://h/openai", + "https://h/v1/extra": "https://h/v1/extra", // A segment that merely STARTS with v must not be stripped: stripping // it would rewrite base "https://h/vision" to "https://h" and, worse, // would treat a real vendor path as a version. - "https://h/vertex": "https://h/vertex", - "https://h/volcengine": "https://h/volcengine", + "https://h/vertex": "https://h/vertex", + "https://h/volcengine": "https://h/volcengine", "https://h/vertex-ai/v1": "https://h/vertex-ai", } for in, want := range cases { @@ -463,7 +463,7 @@ func TestComposeEndpoint_ExactlyOneVersionSegment(t *testing.T) { // Whichever way it resolved, base + endpoint must never carry the // same version segment twice. onBase := trimVersionSuffix(tc.persistedBase) != tc.persistedBase - onEndpoint := versionSegment(strings.TrimSuffix(got, "/chat/completions") + "/models") != "" + onEndpoint := versionSegment(strings.TrimSuffix(got, "/chat/completions")+"/models") != "" if onBase && onEndpoint { t.Errorf("version segment duplicated: base %q + endpoint %q", tc.persistedBase, got) } diff --git a/internal/gateway/key_auto_api_test.go b/internal/gateway/key_auto_api_test.go index c2ab771..7bb62cd 100644 --- a/internal/gateway/key_auto_api_test.go +++ b/internal/gateway/key_auto_api_test.go @@ -4,8 +4,8 @@ import ( "encoding/json" "net/http" "net/http/httptest" - "path/filepath" "os" + "path/filepath" "strings" "testing" @@ -33,11 +33,11 @@ func keyAutoGateway(t *testing.T) *Gateway { t.Fatal(err) } cfg := &config.Config{ - Path: cfgPath, - AdapterDir: filepath.Join(td, "adapters"), - RuntimeFile: filepath.Join(td, "runtime.json"), + Path: cfgPath, + AdapterDir: filepath.Join(td, "adapters"), + RuntimeFile: filepath.Join(td, "runtime.json"), DefaultModel: "AUTO", - GatewayKeys: []string{"sk-admin"}, + GatewayKeys: []string{"sk-admin"}, Keys: []config.GWKey{ {Key: "sk-admin", Role: "admin", Name: "admin"}, {Key: "sk-user", Role: "user", Name: "user"}, @@ -97,7 +97,7 @@ func TestKeyAutoAPICrudAndInherit(t *testing.T) { // Initially inherits. var got struct { - Inherits bool `json:"inherits"` + Inherits bool `json:"inherits"` Auto []config.ModelScope `json:"auto"` } rr := doReqAs(t, g, "sk-admin", "GET", "/api/keys/sk-user/auto", "") @@ -254,4 +254,4 @@ func TestKeyAutoAPIDoesNotShadowKeyRoutes(t *testing.T) { if rr := doReqAs(t, g, "sk-user", "GET", "/api/keys/me", ""); rr.Code != http.StatusOK { t.Fatalf("GET /api/keys/me as user = %d, want 200", rr.Code) } -} \ No newline at end of file +} diff --git a/internal/gateway/messages.go b/internal/gateway/messages.go new file mode 100644 index 0000000..0201831 --- /dev/null +++ b/internal/gateway/messages.go @@ -0,0 +1,322 @@ +package gateway + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "time" + + "llmsproxy/internal/provider" + "llmsproxy/internal/scheduler" + "llmsproxy/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 + } + var raw anthropicRequest + if err := json.NewDecoder(r.Body).Decode(&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() +} + +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 +} diff --git a/internal/gateway/messages_test.go b/internal/gateway/messages_test.go new file mode 100644 index 0000000..ec4a0e5 --- /dev/null +++ b/internal/gateway/messages_test.go @@ -0,0 +1,303 @@ +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() +} diff --git a/internal/gateway/plugin_call_api_test.go b/internal/gateway/plugin_call_api_test.go index 1f9f328..7f4afa1 100644 --- a/internal/gateway/plugin_call_api_test.go +++ b/internal/gateway/plugin_call_api_test.go @@ -147,4 +147,4 @@ func TestPluginCallIsAdminOnly(t *testing.T) { t.Fatalf("a non-admin key reached the call endpoint: HTTP %d %s", rr.Code, rr.Body.String()) } -} \ No newline at end of file +} diff --git a/internal/gateway/plugin_http_routes.go b/internal/gateway/plugin_http_routes.go index 71e6e46..d7f8ce1 100644 --- a/internal/gateway/plugin_http_routes.go +++ b/internal/gateway/plugin_http_routes.go @@ -204,4 +204,4 @@ type bodyTooLarge struct{} func (bodyTooLarge) Error() string { return "request body exceeds 4 MiB" } // limitBody is kept for the numeric form some callers want in a log line. -var _ = strconv.Itoa \ No newline at end of file +var _ = strconv.Itoa diff --git a/internal/gateway/plugins_api_test.go b/internal/gateway/plugins_api_test.go index 85cbcd7..8c4e69f 100644 --- a/internal/gateway/plugins_api_test.go +++ b/internal/gateway/plugins_api_test.go @@ -222,7 +222,7 @@ func TestPluginsListCarriesAccessTables(t *testing.T) { t.Fatalf("GET /api/plugins = %d", rr.Code) } var out struct { - Visibility map[string]string `json:"visibility"` + Visibility map[string]string `json:"visibility"` HTTPAccess []map[string]interface{} `json:"http_access"` } if err := json.Unmarshal(rr.Body.Bytes(), &out); err != nil { diff --git a/internal/gateway/serializer_test.go b/internal/gateway/serializer_test.go index 6b4403b..ca49326 100644 --- a/internal/gateway/serializer_test.go +++ b/internal/gateway/serializer_test.go @@ -32,15 +32,17 @@ func luaVMFor(t *testing.T) *lua.VM { return vm } -// TestFallbackUsedWhenAdapterDeclaresNothing 是「兜底」本身的测试: -// 没有任何内置适配器声明 serialize_*,所以默认路径必须是 OpenAI。 +// TestFallbackUsedWhenAdapterDeclaresNothing 是「兜底」本身的测试。 +// +// 注:anthropic 现在**确实**声明了 serialize_*(Phase 2 给了它 Anthropic +// 形状),所以本测试只在仍未声明钩子的那些内置适配器上验证兜底行为。 +// 这正是设计意图:谁声明谁接管,没声明的继续用 OpenAI 兜底。 func TestFallbackUsedWhenAdapterDeclaresNothing(t *testing.T) { vm := luaVMFor(t) - // 先证明前提:内置适配器确实没声明这些钩子(否则下面测的是别的东西) - for _, name := range []string{"openai", "anthropic", "gemini", "ollama", "deepseek"} { + for _, name := range []string{"openai", "gemini", "ollama", "deepseek", "github", "groq"} { for _, hook := range []string{"serialize_response", "serialize_chunk"} { if vm.HasFn(name, hook) { - t.Fatalf("内置适配器 %s 声明了 %s —— 本测试的前提(默认走兜底)已不成立", name, hook) + t.Errorf("内置适配器 %s 声明了 %s —— 兜底测试的前提已不成立", name, hook) } } } @@ -48,6 +50,11 @@ func TestFallbackUsedWhenAdapterDeclaresNothing(t *testing.T) { if _, isFallback := ser.(openAISerializer); !isFallback { t.Fatalf("未声明钩子的适配器应得到兜底序列化器,得到 %T", ser) } + + // 对照:声明了钩子的适配器不走兜底 + if _, isFallback := serializerFor(vm, "anthropic").(openAISerializer); isFallback { + t.Error("anthropic 声明了 serialize_response,不应走兜底") + } } // TestAdapterSerializerWinsOverFallback 锁定核心规则:适配器声明钩子即接管, @@ -323,19 +330,30 @@ func TestSerializerForConcurrent(t *testing.T) { errs <- "nil serializer" return } - // 结果必须自洽:同一适配器每次拿到同一类型 + // 结果必须自洽:同一适配器每次拿到同一类型,且形状与该适配器 + // 声明的协议一致(openai→OpenAI,anthropic→Anthropic)。 _, isFallback := ser.(openAISerializer) - if w%2 == 0 && !isFallback { - errs <- "openai 未走兜底" - return + wantShape := "chat.completion" + switch name { + case "anthropic": + if isFallback { + errs <- "anthropic 并发下退回兜底" + return + } + wantShape = `"type":"message"` + default: + if !isFallback { + errs <- name + " 并发下未走兜底" + return + } } body, err := ser.NonStream(&types.UnifiedResponse{Content: "x"}, "m", "id", 1) if err != nil { errs <- "serialize: " + err.Error() return } - if !strings.Contains(string(body), "chat.completion") { - errs <- "并发下输出形状不一致" + if !strings.Contains(string(body), wantShape) { + errs <- "并发下输出形状不一致: " + name + " -> " + string(body) return } } diff --git a/internal/gateway/server.go b/internal/gateway/server.go index 3fe9cb8..32bc1d4 100644 --- a/internal/gateway/server.go +++ b/internal/gateway/server.go @@ -216,6 +216,14 @@ func (g *Gateway) routes(w http.ResponseWriter, r *http.Request) { switch { case r.URL.Path == "/v1/chat/completions": g.handleChat(w, r) + case r.URL.Path == "/v1/messages": + // Anthropic Messages protocol. Scheduling, quotas, cooldowns, AUTO + // chain and accounting are EXACTLY the ones /v1/chat/completions uses — + // only the wire shapes differ, and those come from the serving + // adapter's serialize_* hooks. The kernel holds no Anthropic field + // names (see serializer.go: the fallback is OpenAI, and any adapter + // that declares serialize_* replaces it wholesale). + g.handleMessages(w, r) case r.URL.Path == "/v1/images/generations": g.handleImage(w, r) case r.URL.Path == "/v1/models": diff --git a/internal/gateway/stats.go b/internal/gateway/stats.go index 8aeff2a..68d6f1b 100644 --- a/internal/gateway/stats.go +++ b/internal/gateway/stats.go @@ -53,7 +53,7 @@ type Req struct { // 0 means "not reported", which is NOT the same as "no reasoning happened" — // three of the sources measured return no such field at all. Reasoning int64 `json:"reasoning_tokens,omitempty"` - OK bool `json:"ok"` + OK bool `json:"ok"` // Status http status code Status int `json:"status"` // Err short error message diff --git a/internal/gateway/stats_period_records_test.go b/internal/gateway/stats_period_records_test.go index 912e08f..465ebf5 100644 --- a/internal/gateway/stats_period_records_test.go +++ b/internal/gateway/stats_period_records_test.go @@ -74,7 +74,7 @@ func TestStatsPeriodRecordsMatchPager(t *testing.T) { rr := doReq(t, g, "GET", "/api/stats?period=day&limit=100", "") var st struct { Records []struct { - Time int64 `json:"time"` + Time int64 `json:"time"` Model string `json:"model"` } `json:"records"` } @@ -84,7 +84,7 @@ func TestStatsPeriodRecordsMatchPager(t *testing.T) { rr2 := doReq(t, g, "GET", "/api/stats/records?limit=100", "") var pg struct { Records []struct { - Time int64 `json:"time"` + Time int64 `json:"time"` Model string `json:"model"` } `json:"records"` } diff --git a/internal/gateway/stats_unknown_key_test.go b/internal/gateway/stats_unknown_key_test.go index 30a281b..62eff24 100644 --- a/internal/gateway/stats_unknown_key_test.go +++ b/internal/gateway/stats_unknown_key_test.go @@ -137,4 +137,4 @@ func TestStatsAPIUnknownKeyReturnsJSON(t *testing.T) { if resp.StatusCode != http.StatusOK { t.Fatalf("status = %d, want 200", resp.StatusCode) } -} \ No newline at end of file +} diff --git a/internal/gateway/ui_image_page_test.go b/internal/gateway/ui_image_page_test.go index a1b6bfe..5f533ba 100644 --- a/internal/gateway/ui_image_page_test.go +++ b/internal/gateway/ui_image_page_test.go @@ -48,11 +48,11 @@ func TestUI_HasDedicatedImagePage(t *testing.T) { ui := uiSource(t) for _, want := range []string{ - `data-tab="image"`, // sidebar entry - `id="tab-image"`, // pane - `function renderImage`, // renderer registered with refresh() - `tab === "image"`, // dispatched - `/v1/images/generations`, // the endpoint the page actually calls + `data-tab="image"`, // sidebar entry + `id="tab-image"`, // pane + `function renderImage`, // renderer registered with refresh() + `tab === "image"`, // dispatched + `/v1/images/generations`, // the endpoint the page actually calls } { if !strings.Contains(ui, want) { t.Errorf("image page is missing %q", want) @@ -99,8 +99,8 @@ func TestUI_ImagePageCallsTheImageEndpoint(t *testing.T) { // contract: the field the UI reads must actually be serialized. func TestSourceStatus_ExposesModelKindsToTheUI(t *testing.T) { st := provider.SourceStatus{ - Name: "vela", - Models: []string{"claude-sonnet-5.5", "gpt-image-2"}, + Name: "vela", + Models: []string{"claude-sonnet-5.5", "gpt-image-2"}, ModelKinds: []provider.ModelKindInfo{ {ID: "claude-sonnet-5.5", Kind: "chat"}, {ID: "gpt-image-2", Kind: "image"}, @@ -192,4 +192,4 @@ func TestImageEndpointComposesUnderRealConditions(t *testing.T) { "default image path doubles the version segment; if the upstream ever "+ "accepts it, composeImageEndpoint's rule needs rechecking", doubled, resp2.StatusCode) } -} \ No newline at end of file +} diff --git a/internal/gateway/ui_image_wire_test.go b/internal/gateway/ui_image_wire_test.go index 56efb32..3364ddf 100644 --- a/internal/gateway/ui_image_wire_test.go +++ b/internal/gateway/ui_image_wire_test.go @@ -106,4 +106,4 @@ func sendImageSource(t *testing.T) string { body = body[:end+1] } return body -} \ No newline at end of file +} diff --git a/internal/gateway/ui_key_auto_test.go b/internal/gateway/ui_key_auto_test.go index c49fea9..3d131cb 100644 --- a/internal/gateway/ui_key_auto_test.go +++ b/internal/gateway/ui_key_auto_test.go @@ -126,6 +126,7 @@ func TestUIStatsShowsPeriodWindow(t *testing.T) { t.Fatalf("i18n key periodRange appears %d time(s), want both zh and en", n) } } + // Both canvases can exist in the DOM at once: the modal is mounted on top of // the settings tab, whose #scr-canvas stays alive behind it. Resolving the // canvas by document order (whichever comes first) painted the per-key chain diff --git a/internal/gateway/ui_plugin_test.go b/internal/gateway/ui_plugin_test.go index 9414097..8e258a8 100644 --- a/internal/gateway/ui_plugin_test.go +++ b/internal/gateway/ui_plugin_test.go @@ -396,15 +396,15 @@ func TestPluginSidebarEntryCarriesATitle(t *testing.T) { func TestPluginPageHasAccessEditor(t *testing.T) { src := uiSource(t) for _, needle := range []string{ - `renderPluginAccess`, // called after rendering the plugin list - `savePluginAccess`, // wired to a button - `/access`, // hits the per-endpoint access API - `id="pl-access"`, // container exists - `http_access`, // reads the table from /api/plugins - `data-ep=`, // one input per endpoint - `data-plvis=`, // one visibility selector per plugin - `/http/*/access`, // whole-plugin visibility shares the access API - `plVisCell`, // renders the declared-vs-override distinction + `renderPluginAccess`, // called after rendering the plugin list + `savePluginAccess`, // wired to a button + `/access`, // hits the per-endpoint access API + `id="pl-access"`, // container exists + `http_access`, // reads the table from /api/plugins + `data-ep=`, // one input per endpoint + `data-plvis=`, // one visibility selector per plugin + `/http/*/access`, // whole-plugin visibility shares the access API + `plVisCell`, // renders the declared-vs-override distinction } { if !strings.Contains(src, needle) { t.Errorf("plugin page lacks %q", needle) diff --git a/internal/lua/adapters/anthropic.lua b/internal/lua/adapters/anthropic.lua index 578ebb8..459eb41 100644 --- a/internal/lua/adapters/anthropic.lua +++ b/internal/lua/adapters/anthropic.lua @@ -456,4 +456,124 @@ function adapter.transform_error(status, body) return nil end +-- ============================================================ +-- 对外序列化(统一格式 → Anthropic 形状) +-- ============================================================ +-- +-- 内核只持统一格式;对外发什么形状由适配器决定(Phase 2)。 +-- 这几个钩子让以 anthropic 适配器为服务端的源,直接对 Anthropic 协议客户端 +-- 提供 /v1/messages 语义,而内核里不再有任何 Anthropic 字段名。 +-- +-- 与入向的 transform_* 互为逆运算:那里的映射(end_turn→stop)在这里反着 +-- 做回去(stop→end_turn),且默认 stop_sequence 是最保守的选择——它语义上 +-- 最接近"正常结束"。 + +-- 统一 finish_reason → Anthropic stop_reason +local function to_stop_reason(finish) + if finish == "tool_calls" then return "tool_use" end + if finish == "length" then return "max_tokens" end + if finish == "content_filter" then return "refusal" end + -- stop / 空 / 未知 → end_turn(Anthropic 的"正常说完") + return "end_turn" +end + +-- usage 数字化:内部 TokenUsage 的字段名与 Anthropic 的 input_tokens/output_tokens +-- 不同名,这里做最后一次翻译。 +local function to_anthropic_usage(u) + if type(u) ~= "table" then + return { input_tokens = 0, output_tokens = 0 } + end + return { + input_tokens = u.prompt or 0, + output_tokens = u.completion or 0, + } +end + +-- 非流式响应:统一结果 → Anthropic Messages 响应对象 +function adapter.serialize_response(payload) + local p = json.decode(payload) + local blocks = {} + -- thinking 放最前,与 Anthropic 自己的块顺序一致 + if p.reasoning ~= nil and p.reasoning ~= "" then + blocks[#blocks + 1] = { type = "thinking", thinking = p.reasoning } + end + blocks[#blocks + 1] = { type = "text", text = p.content or "" } + if type(p.tool_calls) == "table" and #p.tool_calls > 0 then + for _, tc in ipairs(p.tool_calls) do + blocks[#blocks + 1] = { + type = "tool_use", + id = tc.id or "", + name = tc.name or "", + input = tc.arguments or {}, + } + end + end + return json.encode({ + id = p.id or "", + type = "message", + role = "assistant", + model = p.model or "", + content = blocks, + stop_reason = to_stop_reason(p.finish_reason), + stop_sequence = nil, + usage = to_anthropic_usage(p.token_usage), + }) +end + +-- 角色前导:Anthropic 的流以 message_start 开头(不是 role delta) +function adapter.serialize_open_preamble(payload) + local p = json.decode(payload) + return json.encode({ + type = "message_start", + message = { + id = p.id or "", + type = "message", + role = "assistant", + model = p.model or "", + content = {}, + stop_reason = nil, + stop_sequence = nil, + usage = { input_tokens = 0, output_tokens = 0 }, + }, + }) +end + +-- 流式增量:统一 chunk → Anthropic content_block_delta +-- +-- 真实 Anthropic 流需要 content_block_start(开块)→ delta → stop(关块) +-- 的三段式。这里用与多数 Anthropic 客户端兼容的简化形式:首个文本增量 +-- 带 index=0 的 delta,后续同 index;关闭由 serialize_close 负责。 +function adapter.serialize_chunk(payload) + local p = json.decode(payload) + if p.done then + -- 终止由 serialize_close 统一发,这里不重复 + return "" + end + if p.reasoning ~= nil and p.reasoning ~= "" then + return json.encode({ + type = "content_block_delta", + index = 0, + delta = { type = "thinking_delta", thinking = p.reasoning }, + }) + end + return json.encode({ + type = "content_block_delta", + index = 0, + delta = { type = "text_delta", text = p.content or "" }, + }) +end + +-- 收尾:message_delta(stop_reason + usage)→ message_stop +function adapter.serialize_close(payload) + local p = json.decode(payload) + return json.encode({ + type = "message_delta", + delta = { + stop_reason = to_stop_reason(p.finish_reason), + stop_sequence = nil, + }, + usage = { output_tokens = 0 }, + }) .. "\n" .. json.encode({ type = "message_stop" }) +end + return adapter diff --git a/internal/lua/ser_check_test.go b/internal/lua/ser_check_test.go new file mode 100644 index 0000000..eae02bd --- /dev/null +++ b/internal/lua/ser_check_test.go @@ -0,0 +1,53 @@ +package lua + +import ( + "encoding/json" + "path/filepath" + "strings" + "testing" +) + +func TestAnthropicSerializerShape(t *testing.T) { + vm := NewVM(filepath.Join(t.TempDir(), "adapters")) + if err := vm.Start(); err != nil { + t.Fatal(err) + } + defer vm.Stop() + + if !vm.HasFn("anthropic", "serialize_response") { + t.Fatal("anthropic 未声明 serialize_response") + } + out, err := vm.Transform("anthropic", "serialize_response", + `{"model":"claude-x","id":"msg_1","content":"hello","finish_reason":"stop","token_usage":{"prompt":10,"completion":5,"total":15}}`) + if err != nil { + t.Fatal(err) + } + var m map[string]interface{} + if err := json.Unmarshal([]byte(out), &m); err != nil { + t.Fatalf("not JSON: %s", out) + } + if m["type"] != "message" || m["role"] != "assistant" { + t.Errorf("Anthropic 形状不对: %s", out) + } + if m["stop_reason"] != "end_turn" { + t.Errorf("stop_reason = %v, want end_turn", m["stop_reason"]) + } + blocks, _ := m["content"].([]interface{}) + if len(blocks) != 1 { + t.Fatalf("blocks = %v", m["content"]) + } + b0 := blocks[0].(map[string]interface{}) + if b0["type"] != "text" || b0["text"] != "hello" { + t.Errorf("首个块应为 text/hello: %v", b0) + } + u := m["usage"].(map[string]interface{}) + if u["input_tokens"] != float64(10) || u["output_tokens"] != float64(5) { + t.Errorf("usage 字段名/值不对: %v", u) + } + // 内核不得混入 OpenAI 字段 + for _, k := range []string{"object", "choices", "usage.prompt_tokens"} { + if strings.Contains(out, `"`+k+`"`) { + t.Errorf("OpenAI 字段 %q 混入 Anthropic 输出: %s", k, out) + } + } +}