feat(adapters): 入向端点移交适配器 —— adapter.ingress + deserialize_request，内核主动让出

plan.md 的 R2 目标里有三行钩子契约，出向两条（serialize_response/serialize_chunk） Phase 2 已落地，第三条 parse_request(protocol_body) -> unified 一直空着： /v1/messages 的入向解析硬编码在内核 Go 里（anthropicRequest.toUnified）， Anthropic 改协议就得改内核。本次补齐这条对偶。 ## 机制（与出向序列化对称） - 适配器声明 adapter.ingress = "/v1/messages" 认领入站路径，并实现 deserialize_request(raw_body) -> unified_json。 - 内核 handleMessages 经新增的 parseViaIngress 判定： - 有适配器认领该路径且实现了钩子 → 让出，解析由适配器完成； - 无适配器认领 → 内核内置解析运行（R5 兜底，未配置部署仍可服务）。 - 认领了但钩子执行失败 → 硬错误 400，不静默换用另一种协议解析（客户端说的正是该适配器声称支持的协议，悄悄换解析器只会错误调度）。 - 这是内核里唯一的新判定点，与 serializerFor 对称：出向选序列化器，入向选解析器。 ## 内核改动 - internal/lua/vm.go：staticInfo 增 ingress 字段；inspectScript 读取；新增 VM.AdapterForIngress(path)（确定性选择：同名多认领取字典序最小）。 - internal/gateway/messages.go：handleMessages 改为先读 body → parseViaIngress （让出）→ 未认领时回落内置 anthropicRequest.toUnified。内置解析与 anthropicRequest 类型保留——它们是兜底，不是死代码。 ## 适配器改动 - internal/lua/adapters/anthropic.lua：声明 adapter.ingress = "/v1/messages"，新增 deserialize_request，把内核原 Go 解析逻辑逐字段移植（system 两种拼写、 content 原样透传、缺省 thinking={enabled:false}、缺 messages 报错）。内置 anthropic 默认随包播种，因此真实部署里内核总是让出给它， Go 侧 anthropicRequest.toUnified 退化为纯兜底。 ## 测试 - gateway：TestMessagesIngressYieldsToBundledAnthropic（让出后仍 200）、 TestAdapterForIngressClaim（声明可被找到、未声明路径为空、钩子存在）、 TestMessagesKernelFallbackWhenNoAdapterClaims（移除认领适配器后内核兜底仍 200）。 - lua：TestAnthropicDeserializeRequestShape 逐字段核对 Lua 与 Go 产出一致（system 拼接、content、max_tokens/stream/temperature/top_p/top_k/stop_sequences、缺省 thinking），TestAnthropicDeserializeRejectsMissingMessages。 - 全量回归：gateway/lua/core/config/types/provider/scheduler/e2e 全绿。 - 文档：docs/lua-adapters.md 补「入向让出钩子」章节 + plan.md 台账更新。
2026-11-12 20:27:23 +00:00 · 2026-10-12 16:32:09 +08:00
parent 5948ad19ca
commit 3c59c33c58
6 changed files with 420 additions and 6 deletions
--- a/internal/gateway/messages.go
+++ b/internal/gateway/messages.go
@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"net/http"
 	"strings"
 	"time"
@ -40,16 +41,34 @@ func (g *Gateway) handleMessages(w http.ResponseWriter, r *http.Request) {
 		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())
+	body, err := io.ReadAll(r.Body)
+	if err != nil {
+		writeError(w, http.StatusBadRequest, "invalid_request", "read body: "+err.Error())
 		return
 	}
-	req, err := raw.toUnified(g.core.DefaultModel())
-	if err != nil {
-		writeError(w, http.StatusBadRequest, "invalid_request", err.Error())
+	// 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()
@ -292,6 +311,44 @@ func systemToString(raw json.RawMessage) string {
 	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"
--- a/internal/gateway/messages_test.go
+++ b/internal/gateway/messages_test.go
@ -301,3 +301,104 @@ func TestMessagesRequiresAuth(t *testing.T) {
 	}
 	_ = 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)
+	}
+}