package lua import ( "encoding/json" "os" "path/filepath" "strings" "sync" "testing" "time" ) // freshAdapterDir returns a path under a temp dir that does not exist yet, so // VM.Start() treats it as first-run and seeds the bundled adapters. func freshAdapterDir(t *testing.T) string { t.Helper() return filepath.Join(t.TempDir(), "adapters") } func TestLoadBundledAdapters(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() adapters := vm.ListAdapters() if len(adapters) == 0 { t.Fatal("no adapters loaded") } names := map[string]bool{} for _, a := range adapters { names[a.Name] = true } for _, want := range []string{"openai", "deepseek", "anthropic", "gemini", "ollama", "kimicode", "opencode"} { if !names[want] { t.Errorf("missing adapter %s (got %v)", want, names) } } } func TestFirstRunSeedsEveryBundledAdapter(t *testing.T) { dir := filepath.Join(t.TempDir(), "adapters") vm := NewVM(dir) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() // every embedded .lua must have been copied out (and no extra, non-.lua files) embedded, err := bundledAdapters.ReadDir("adapters") if err != nil { t.Fatal(err) } seeded := 0 for _, e := range embedded { if e.IsDir() || !strings.HasSuffix(e.Name(), ".lua") { continue } seeded++ if _, err := os.Stat(filepath.Join(dir, e.Name())); err != nil { t.Errorf("embedded adapter %s not seeded", e.Name()) } } if seeded == 0 { t.Fatal("no embedded adapters found") } written := 0 entries, _ := os.ReadDir(dir) for _, e := range entries { if !e.IsDir() && strings.HasSuffix(e.Name(), ".lua") { written++ } } if written != seeded { t.Fatalf("seeded %d files but %d embedded adapters exist", written, seeded) } // re-start with the existing dir: authoritative, never rewritten vm2 := NewVM(dir) if err := vm2.Start(); err != nil { t.Fatalf("second start: %v", err) } defer vm2.Stop() } func TestTransformRequest(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() out, err := vm.Transform("openai", "transform_request", `{"model":"x","disable_thinking":true,"messages":[]}`) if err != nil { t.Fatalf("transform: %v", err) } if strings.Contains(out, "disable_thinking") { t.Fatalf("disable_thinking not stripped: %s", out) } } func TestBuildHeadersFallbackStatic(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() hdrs, err := vm.BuildHeaders("anthropic", nil) if err != nil { t.Fatalf("build headers: %v", err) } if hdrs["anthropic-version"] != "2023-06-01" { t.Fatalf("static header missing: %v", hdrs) } } func TestOpenCodeAdapterFingerprint(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() hdrs, err := vm.BuildHeaders("opencode", map[string]interface{}{ "api_key": "public", "timestamp": 1700000000, "body": `{"model":"x"}`, "url": "https://opencode.ai/zen/v1/chat/completions", "method": "POST", }) if err != nil { t.Fatalf("build headers: %v", err) } // zen 按客户端指纹路由请求池:UA 必须是 opencode 真实格式, // 且必须携带 x-opencode-* 身份头,否则带 tools 的请求会被上游 // 判为匿名客户端并返回 network_error 空流(表现为空回复)。 if ua := hdrs["User-Agent"]; !strings.HasPrefix(ua, "opencode/") || !strings.Contains(ua, "ai-sdk/provider-utils/") { t.Fatalf("opencode adapter must send the real opencode client UA, got %q", ua) } for _, h := range []string{"x-opencode-client", "x-opencode-project", "x-opencode-session", "x-opencode-request"} { if hdrs[h] == "" { t.Fatalf("opencode adapter must send %q (zen client fingerprint), got %v", h, hdrs) } } hdrs2, _ := vm.BuildHeaders("opencode", map[string]interface{}{"api_key": "public", "timestamp": 1700000001, "body": `{"model":"y"}`, "method": "POST"}) if hdrs["x-opencode-request"] == hdrs2["x-opencode-request"] { t.Fatalf("x-opencode-request must vary per request, got %q twice", hdrs["x-opencode-request"]) } if hdrs["Authorization"] != "" { t.Fatalf("opencode adapter must not hardcode Authorization (config api_key supplies it), got %q", hdrs["Authorization"]) } // passthrough behaves like the openai adapter out, err := vm.Transform("opencode", "transform_request", `{"model":"x","disable_thinking":true,"extra_body":{},"messages":[{"role":"user","content":"hi"}]}`) if err != nil { t.Fatal(err) } if strings.Contains(out, "disable_thinking") || strings.Contains(out, "extra_body") { t.Fatalf("opencode transform_request must strip provider fields: %s", out) } } func TestOpenCodeNormalizesRoles(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() // zen 上游只接受 system/user/assistant/tool/latest_reminder; // OpenAI 系客户端发 `developer`(Claude Code 还会用 `function`)必须归一化成 system。 body := `{"model":"x","messages":[ {"role":"developer","content":"you are helpful"}, {"role":"system","content":"sys"}, {"role":"user","content":"hi"}, {"role":"assistant","content":"ok"}, {"role":"tool","tool_call_id":"t1","content":"{\"ok\":true}"} ]}` out, err := vm.Transform("opencode", "transform_request", body) if err != nil { t.Fatal(err) } if strings.Contains(out, `"developer"`) { t.Fatalf("developer role must be normalized: %s", out) } if !strings.Contains(out, `"system"`) || !strings.Contains(out, `"user"`) { t.Fatalf("allowed roles must survive: %s", out) } if !strings.Contains(out, `"assistant"`) { t.Fatalf("assistant role must survive: %s", out) } if !strings.Contains(out, `"tool"`) { t.Fatalf("tool role must survive: %s", out) } } func TestOpenCodeStripsMultiModalParts(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() body := `{"model":"x","messages":[ {"role":"user","content":[ {"type":"text","text":"describe"}, {"type":"image_url","image_url":{"url":"data:image/png;base64,QUJD"}}, {"type":"input_audio","input_audio":{"data":"QQ","format":"wav"}} ]}, {"role":"user","content":[{"type":"image_url","image_url":{"url":"https://ex.com/a.png"}}]}, {"role":"assistant","content":"plain"} ]}` out, err := vm.Transform("opencode", "transform_request", body) if err != nil { t.Fatal(err) } if strings.Contains(out, "image_url") || strings.Contains(out, "input_audio") { t.Fatalf("opencode must strip multimodal parts: %s", out) } if !strings.Contains(out, "describe") { t.Fatalf("text parts must survive: %s", out) } // image-only message dropped: only the text message and plain assistant message remain if got := strings.Count(out, `"role"`); got != 2 { t.Fatalf("image-only message must be dropped, got %d messages: %s", got, out) } } func TestBuildHeadersCustomHook(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() hdrs, err := vm.BuildHeaders("kimicode", map[string]interface{}{ "timestamp": int64(12345), "api_key": "k", "body": "{}", "method": "POST", "url": "http://x/chat", "source": map[string]interface{}{"meta": map[string]interface{}{ "app_id": "app-9", "app_secret": "s", "api_key": "k", }}, }) if err != nil { t.Fatalf("build headers: %v", err) } if hdrs["X-App-Id"] != "app-9" { t.Fatalf("x-app-id = %q", hdrs["X-App-Id"]) } if hdrs["X-App-Sign"] == "" { t.Fatal("expected signature header") } if hdrs["X-Timestamp"] != "12345" { t.Fatalf("timestamp = %q", hdrs["X-Timestamp"]) } } func TestDisableThinkingPassthrough(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() body := `{"model":"x","disable_thinking":true,"messages":[{"role":"user","content":"hi"}]}` out, err := vm.Transform("deepseek", "transform_request", body) if err != nil { t.Fatalf("deepseek transform: %v", err) } var req struct { ExtraBody map[string]interface{} `json:"extra_body"` } if err := json.Unmarshal([]byte(out), &req); err != nil { t.Fatalf("unmarshal: %v (%s)", err, out) } thinking, ok := req.ExtraBody["thinking"].(map[string]interface{}) if !ok { t.Fatalf("deepseek should emit extra_body.thinking on disable_thinking: %s", out) } if thinking["type"] != "disabled" { t.Fatalf("thinking.type = %v", thinking["type"]) } // anthropic: thinking is opt-in (Issue 3) — never emitted by default, and // disable_thinking is not a trigger either. out2, err := vm.Transform("anthropic", "transform_request", body) if err != nil { t.Fatalf("anthropic transform: %v", err) } if strings.Contains(out2, "thinking") { t.Fatalf("anthropic should drop thinking when disable_thinking: %s", out2) } out3, err := vm.Transform("anthropic", "transform_request", `{"model":"x","messages":[{"role":"user","content":"hi"}]}`) if err != nil { t.Fatalf("anthropic transform: %v", err) } if strings.Contains(out3, "thinking") { t.Fatalf("anthropic must not enable thinking by default: %s", out3) } // opt-in path: extra_body.thinking is forwarded verbatim out4, err := vm.Transform("anthropic", "transform_request", `{"model":"x","messages":[{"role":"user","content":"hi"}],"extra_body":{"thinking":{"type":"enabled","budget_tokens":2048}}}`) if err != nil { t.Fatalf("anthropic transform: %v", err) } if !strings.Contains(out4, `"type":"enabled"`) || !strings.Contains(out4, "2048") { t.Fatalf("anthropic should forward extra_body.thinking: %s", out4) } } func TestMultimodalTransform(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() body := `{"model":"x","messages":[{"role":"user","content":[ {"type":"text","text":"what is this?"}, {"type":"image_url","image_url":{"url":"data:image/png;base64,QUJD"}}, {"type":"image_url","image_url":{"url":"https://ex.com/a.png"}} ]}]}` // anthropic: image_url -> image block (base64/url), text preserved out, err := vm.Transform("anthropic", "transform_request", body) if err != nil { t.Fatalf("anthropic: %v", err) } for _, want := range []string{`"media_type":"image/png"`, `"data":"QUJD"`, `"type":"url","url":"https://ex.com/a.png"`, `"what is this?"`} { if !strings.Contains(out, want) { t.Fatalf("anthropic multimodal missing %s: %s", want, out) } } // gemini: image_url -> inline_data gout, err := vm.Transform("gemini", "transform_request", body) if err != nil { t.Fatalf("gemini: %v", err) } var g struct { Contents []struct { Parts []map[string]interface{} `json:"parts"` } `json:"contents"` } if err := json.Unmarshal([]byte(gout), &g); err != nil { t.Fatalf("gemini unmarshal: %v", err) } if len(g.Contents) == 0 { t.Fatalf("gemini no contents") } var found bool for _, p := range g.Contents[0].Parts { if v, ok := p["inline_data"].(map[string]interface{}); ok && v["data"] == "QUJD" && v["mime_type"] == "image/png" { found = true } } if !found { t.Fatalf("gemini missing inline_data image: %s", gout) } // ollama: image_url -> images base64 array out, err = vm.Transform("ollama", "transform_request", body) if err != nil { t.Fatalf("ollama: %v", err) } if !strings.Contains(out, `"images":["QUJD"]`) { t.Fatalf("ollama multimodal missing images: %s", out) } // openai passthrough keeps the content array intact po, _ := vm.Transform("openai", "transform_request", body) if !strings.Contains(po, `"image_url"`) || !strings.Contains(po, `,QUJD"`) { t.Fatalf("openai passthrough lost multimodal content: %s", po) } } func TestOpenCodeAdapterNormalizesDeveloperRole(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() raw := `{"model":"zen","messages":[` + `{"role":"developer","content":"be concise"},` + `{"role":"user","content":"hi"},` + `{"role":"assistant","content":"hello"},` + `{"role":"function","content":"{\"a\":1}"},` + `{"role":"latest_reminder","content":"remind"}]}` out, err := vm.Transform("opencode", "transform_request", raw) if err != nil { t.Fatalf("transform_request: %v", err) } var req struct { Messages []struct { Role string `json:"role"` } `json:"messages"` } if err := json.Unmarshal([]byte(out), &req); err != nil { t.Fatalf("output not JSON: %v\n%s", err, out) } want := []string{"system", "user", "assistant", "system", "latest_reminder"} if len(req.Messages) != len(want) { t.Fatalf("got %d messages, want %d: %s", len(req.Messages), len(want), out) } for i, w := range want { if req.Messages[i].Role != w { t.Errorf("message %d role = %q, want %q", i, req.Messages[i].Role, w) } } } func TestOpenCodeAdapterKeepsWhitelistedRolesAndDropsMultimodal(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() raw := `{"messages":[` + `{"role":"system","content":"sys"},` + `{"role":"user","content":[{"type":"text","text":"keep"},{"type":"image_url","url":"x"}]},` + `{"role":"user","content":[{"type":"image_url","url":"x"}]}]}` out, err := vm.Transform("opencode", "transform_request", raw) if err != nil { t.Fatalf("transform_request: %v", err) } var req struct { Messages []struct { Role string `json:"role"` Content any `json:"content"` } `json:"messages"` } if err := json.Unmarshal([]byte(out), &req); err != nil { t.Fatalf("output not JSON: %v\n%s", err, out) } if len(req.Messages) != 2 { t.Fatalf("got %d messages, want 2 (multimodal-only dropped): %s", len(req.Messages), out) } if req.Messages[0].Role != "system" || req.Messages[1].Role != "user" { t.Fatalf("unexpected roles: %s", out) } } func TestAdaptersPassFinishReason(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() // OpenAI-style chunk with a real finish reason must surface finish_reason // and done=true; an empty-string finish_reason (sensenova sends "" on // every chunk) must NOT terminate the stream. chunk := `{"choices":[{"index":0,"finish_reason":"tool_calls","delta":{"content":""}}]}` empty := `{"choices":[{"index":0,"finish_reason":"","delta":{"content":"x"}}]}` for _, name := range []string{"openai", "deepseek", "github", "groq", "kimicode", "mistral", "opencode"} { out, err := vm.Transform(name, "transform_stream_chunk", chunk) if err != nil { t.Fatalf("%s: %v", name, err) } if !strings.Contains(out, `"finish_reason":"tool_calls"`) { t.Fatalf("%s: finish_reason lost: %s", name, out) } if !strings.Contains(out, `"done":true`) { t.Fatalf("%s: done not set on real finish: %s", name, out) } out, err = vm.Transform(name, "transform_stream_chunk", empty) if err != nil { t.Fatalf("%s: %v", name, err) } if strings.Contains(out, `"done":true`) { t.Fatalf("%s: empty finish_reason must not end stream: %s", name, out) } } } // TestAnthropicToolRoundTrip verifies the OpenAI->Anthropic request mapping for // a full agent tool-call round (Issue 1): assistant tool_calls become // tool_use content blocks and role:"tool" results become user tool_result // blocks merged into ONE user message. This is the regression that made agent // clients (dsh/Claude Code/Cursor) repeatedly re-invoke the same tool. func TestAnthropicToolRoundTrip(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() round1 := `{"model":"x","tools":[{"type":"function","function":{"name":"calc","description":"multiply","parameters":{"type":"object","properties":{"a":{"type":"integer"},"b":{"type":"integer"}},"required":["a","b"]}}}],"messages":[{"role":"user","content":"what is 17*23?"}]}` out1, err := vm.Transform("anthropic", "transform_request", round1) if err != nil { t.Fatalf("round1 transform: %v", err) } var r1 struct { Tools []struct { Name string `json:"name"` InputSchema map[string]interface{} `json:"input_schema"` } `json:"tools"` Messages []map[string]interface{} `json:"messages"` } if err := json.Unmarshal([]byte(out1), &r1); err != nil { t.Fatalf("unmarshal r1: %v (%s)", err, out1) } if len(r1.Tools) != 1 || r1.Tools[0].Name != "calc" { t.Fatalf("tools not mapped: %s", out1) } // Round 2: assistant tool_calls + tool result round2 := `{"model":"x","messages":[ {"role":"user","content":"what is 17*23?"}, {"role":"assistant","content":"","tool_calls":[ {"id":"call_1","type":"function","function":{"name":"calc","arguments":"{\"a\":17,\"b\":23}"}} ]}, {"role":"tool","tool_call_id":"call_1","content":"391"} ]}` out2, err := vm.Transform("anthropic", "transform_request", round2) if err != nil { t.Fatalf("round2 transform: %v", err) } var r2 struct { Messages []struct { Role string `json:"role"` Content []struct { Type string `json:"type"` ID string `json:"id"` Name string `json:"name"` Input map[string]interface{} `json:"input"` ToolUseID string `json:"tool_use_id"` ContentText string `json:"content"` } `json:"content"` } `json:"messages"` } if err := json.Unmarshal([]byte(out2), &r2); err != nil { t.Fatalf("unmarshal r2: %v (%s)", err, out2) } // Message 2 (index 1) must be assistant with one tool_use block am := r2.Messages[1] if am.Role != "assistant" { t.Fatalf("msg[1].role = %q, want assistant", am.Role) } var toolUse struct { Type string `json:"type"` ID string `json:"id"` Name string `json:"name"` Input map[string]interface{} `json:"input"` } for _, b := range am.Content { if b.Type == "tool_use" { toolUse = struct { Type string `json:"type"` ID string `json:"id"` Name string `json:"name"` Input map[string]interface{} `json:"input"` }{b.Type, b.ID, b.Name, b.Input} } } if toolUse.ID != "call_1" || toolUse.Name != "calc" { t.Fatalf("tool_use not mapped: %+v", am.Content) } if toolUse.Input["a"] != float64(17) || toolUse.Input["b"] != float64(23) { t.Fatalf("tool_use input args not decoded from JSON string: %+v", toolUse.Input) } // Message 3 (index 2) must be user with one tool_result block um := r2.Messages[2] if um.Role != "user" { t.Fatalf("msg[2].role = %q, want user (tool_result)", um.Role) } if len(um.Content) != 1 || um.Content[0].Type != "tool_result" || um.Content[0].ToolUseID != "call_1" { t.Fatalf("tool_result not mapped: %+v", um.Content) } // Round 3: consecutive tool results must merge into ONE user message round3 := `{"model":"x","messages":[ {"role":"user","content":"do both"}, {"role":"assistant","content":"","tool_calls":[ {"id":"c1","type":"function","function":{"name":"calc","arguments":"{\"a\":1,\"b\":2}"}}, {"id":"c2","type":"function","function":{"name":"calc","arguments":"{\"a\":3,\"b\":4}"}} ]}, {"role":"tool","tool_call_id":"c1","content":"2"}, {"role":"tool","tool_call_id":"c2","content":"12"} ]}` out3, err := vm.Transform("anthropic", "transform_request", round3) if err != nil { t.Fatalf("round3 transform: %v", err) } var r3 struct { Messages []map[string]interface{} `json:"messages"` } if err := json.Unmarshal([]byte(out3), &r3); err != nil { t.Fatalf("unmarshal r3: %v (%s)", err, out3) } // messages: user, assistant(tool_use x2), user(tool_result x2 merged) if len(r3.Messages) != 3 { t.Fatalf("round3 len(messages) = %d, want 3 (merged tool results): %s", len(r3.Messages), out3) } last := r3.Messages[3-1] blocks := last["content"].([]interface{}) if len(blocks) != 2 { t.Fatalf("last user message content blocks = %d, want 2 merged tool_results: %s", len(blocks), out3) } } // TestOpenAITransformErrorEnvelopes covers Issue 6: the openai adapter must // condense every shape of upstream error body that real OpenAI-compatible // gateways emit, not just the {error:{message}} envelope. Unhandled shapes // used to fall through to Go's generic "unknown error" and dump raw HTML / // JSON into the server log. func TestOpenAITransformErrorEnvelopes(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() cases := []struct { name string status int body string want string }{ {"standard openai envelope", 429, `{"error":{"message":"Rate limit reached","type":"rate_limit"}}`, "Rate limit reached"}, {"error as bare string", 400, `{"error":"bad request"}`, "bad request"}, {"flat numeric code (qijiar/siliconflow style)", 400, `{"code":20012,"message":"Model does not exist. Please check it carefully.","data":null}`, "20012: Model does not exist. Please check it carefully."}, {"flat string code (remotezen style)", 401, `{"code":"INVALID_API_KEY","message":"Invalid API key"}`, "INVALID_API_KEY: Invalid API key"}, {"fastapi detail", 422, `{"detail":"validation failed"}`, "validation failed"}, {"nginx html error page", 413, ` 413 Request Entity Too Large

413 Request Entity Too Large


nginx/1.18.0 (Ubuntu)
`, "413 Request Entity Too Large"}, {"plain text body", 502, "upstream connect error", "upstream connect error"}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { got, ok, err := vm.TransformError("openai", tc.status, tc.body) if err != nil { t.Fatalf("TransformError: %v", err) } if !ok { t.Fatalf("hook returned no reason for %s body: %s", tc.name, tc.body) } if got != tc.want { t.Fatalf("reason = %q, want %q", got, tc.want) } }) } // A body carrying no usable message must fall through (ok=false) so the // Go-side generic condenser stays in charge instead of inventing text. t.Run("no usable message falls through", func(t *testing.T) { if _, ok, _ := vm.TransformError("openai", 500, `{"foo":"bar"}`); ok { t.Fatal("expected fallthrough for a body with no message field") } }) } // TestAdaptersReportZeroCacheHit covers the "distinguish missed from not // reported" contract on BOTH adapter paths: whenever an upstream reports a // cache field, the adapter must emit prompt_tokens_details even when the hit // count is 0, so the gateway can record cache_reported=true and the UI shows // 0% instead of "—". Adapters that dropped the 0 case made a reported miss // indistinguishable from an upstream that never reported cache info. func TestAdaptersReportZeroCacheHit(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatal(err) } defer vm.Stop() // OpenAI-shaped upstreams: usage.prompt_tokens_details.cached_tokens = 0 openaiLike := []string{"openai", "deepseek", "sensenova", "opencode", "agentrouter", "github", "groq", "kimicode", "mistral"} respBody := `{"choices":[{"message":{"content":"hi"},"finish_reason":"stop"}], "usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12, "prompt_tokens_details":{"cached_tokens":0}}}` streamBody := `{"choices":[],"usage":{"prompt_tokens":10,"completion_tokens":2, "total_tokens":12,"prompt_tokens_details":{"cached_tokens":0}}}` for _, name := range openaiLike { out, err := vm.Transform(name, "transform_response", respBody) if err != nil { t.Fatalf("%s transform_response: %v", name, err) } if !strings.Contains(out, "prompt_tokens_details") { t.Fatalf("%s: zero cached_tokens dropped in transform_response: %s", name, out) } out, err = vm.Transform(name, "transform_stream_chunk", streamBody) if err != nil { t.Fatalf("%s transform_stream_chunk: %v", name, err) } if !strings.Contains(out, "prompt_tokens_details") { t.Fatalf("%s: zero cached_tokens dropped in transform_stream_chunk: %s", name, out) } } // gemini: usageMetadata.cachedContentTokenCount = 0 gResp := `{"candidates":[{"content":{"parts":[{"text":"hi"}]},"finishReason":"STOP"}], "usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":2, "totalTokenCount":12,"cachedContentTokenCount":0}}` out, err := vm.Transform("gemini", "transform_response", gResp) if err != nil { t.Fatalf("gemini transform_response: %v", err) } if !strings.Contains(out, "prompt_tokens_details") { t.Fatalf("gemini: cachedContentTokenCount not normalized in transform_response: %s", out) } gStream := `{"usageMetadata":{"promptTokenCount":10,"candidatesTokenCount":2, "totalTokenCount":12,"cachedContentTokenCount":0}}` out, err = vm.Transform("gemini", "transform_stream_chunk", gStream) if err != nil { t.Fatalf("gemini transform_stream_chunk: %v", err) } if !strings.Contains(out, "prompt_tokens_details") { t.Fatalf("gemini: zero cachedContentTokenCount dropped in stream: %s", out) } // anthropic: usage.cache_read_input_tokens = 0 aResp := `{"content":[{"type":"text","text":"hi"}],"stop_reason":"end_turn", "usage":{"input_tokens":10,"output_tokens":2,"cache_read_input_tokens":0}}` out, err = vm.Transform("anthropic", "transform_response", aResp) if err != nil { t.Fatalf("anthropic transform_response: %v", err) } if !strings.Contains(out, "prompt_tokens_details") { t.Fatalf("anthropic: zero cache_read_input_tokens dropped in response: %s", out) } aStream := `{"type":"message_start","message":{"usage":{"input_tokens":10, "output_tokens":2,"cache_read_input_tokens":0}}}` out, err = vm.Transform("anthropic", "transform_stream_chunk", aStream) if err != nil { t.Fatalf("anthropic transform_stream_chunk: %v", err) } if !strings.Contains(out, "prompt_tokens_details") { t.Fatalf("anthropic: zero cache_read_input_tokens dropped in stream: %s", out) } // Negative control: an upstream that reports NO cache field at all must // not fabricate details (that would flip "not reported" into a fake 0%). noCache := `{"choices":[{"message":{"content":"hi"},"finish_reason":"stop"}], "usage":{"prompt_tokens":10,"completion_tokens":2,"total_tokens":12}}` out, err = vm.Transform("openai", "transform_response", noCache) if err != nil { t.Fatal(err) } if strings.Contains(out, "prompt_tokens_details") { t.Fatalf("openai fabricated cache details when upstream reported none: %s", out) } } // ---- elastic pool sizing (plan 阶段 2) ---- // poolOf reaches into the VM for one adapter's pool. Tests assert on internal // counters because the whole point of the change is that they no longer grow // monotonically. func poolOf(t *testing.T, v *VM, name string) *adapterPool { t.Helper() p := v.pool(name) if p == nil { t.Fatalf("adapter %q not loaded", name) } return p } func poolCounts(p *adapterPool) (created, idle, inUse int) { p.mu.Lock() defer p.mu.Unlock() return p.created, len(p.idle), p.inUse } // TestPoolStartsEmpty is the startup-memory claim: loading adapters must not // boot a single Lua state, no matter how large the configured ceiling is. func TestPoolStartsEmpty(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 64, "deepseek": 32}) for _, st := range vm.PoolStats() { if st.Created != 0 || st.Idle != 0 || st.InUse != 0 { t.Fatalf("adapter %q booted eagerly: created=%d idle=%d in_use=%d", st.Name, st.Created, st.Idle, st.InUse) } } } // TestGrowStepFollowsMaxConcurrent pins requirement 3: the growth step is // derived from the adapter's maximum concurrency. func TestGrowStepFollowsMaxConcurrent(t *testing.T) { p := newAdapterPool("t", "return {name='t'}", staticInfo{name: "t"}) for _, tc := range []struct{ max, want int }{ {1, 1}, {4, 1}, {8, 1}, {9, 2}, {16, 2}, {32, 4}, {64, 8}, {200, 8}, } { p.setMax(tc.max) p.mu.Lock() got := p.growStepLocked() p.mu.Unlock() if got != tc.want { t.Fatalf("max=%d: grow step = %d, want %d", tc.max, got, tc.want) } } } // TestShrinkStepFollowsConnections pins requirement 4: the shrink step is // driven by the live connection count — no connections means collapse, busy // means give up one state at a time. func TestShrinkStepFollowsConnections(t *testing.T) { p := newAdapterPool("t", "return {name='t'}", staticInfo{name: "t"}) p.setMax(32) p.used = true cases := []struct { created, idle, inUse, want int why string }{ {created: 9, idle: 9, inUse: 0, want: 8, why: "no connections: collapse to the resident floor in one round"}, {created: 9, idle: 8, inUse: 1, want: 4, why: "1 connection: halve the slack"}, {created: 9, idle: 6, inUse: 3, want: 2, why: "3 connections: quarter the slack"}, {created: 9, idle: 2, inUse: 7, want: 1, why: "busy: release one state at a time"}, {created: 4, idle: 0, inUse: 4, want: 0, why: "everything checked out: nothing to reclaim"}, {created: 1, idle: 1, inUse: 0, want: 0, why: "at the resident floor: keep the warm state"}, } for _, tc := range cases { p.mu.Lock() p.created, p.inUse = tc.created, tc.inUse p.idle = make([]*worker, tc.idle) got := p.shrinkStepLocked() p.mu.Unlock() if got != tc.want { t.Fatalf("%s: created=%d idle=%d inUse=%d -> step %d, want %d", tc.why, tc.created, tc.idle, tc.inUse, got, tc.want) } } } // TestPoolGrowsOnDemandAndReclaims is the end-to-end elasticity loop: boot on // demand, stay under the ceiling, then collapse back to the resident floor. func TestPoolGrowsOnDemandAndReclaims(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 16}) p := poolOf(t, vm, "openai") if created, _, _ := poolCounts(p); created != 0 { t.Fatalf("pool must start empty, created=%d", created) } // concurrent load: hold several workers at once so the pool must grow const hold = 6 var wg sync.WaitGroup release := make(chan struct{}) for i := 0; i < hold; i++ { wg.Add(1) go func() { defer wg.Done() w, err := p.acquire() if err != nil { t.Errorf("acquire: %v", err) return } <-release p.release(w) }() } // wait until all six are checked out deadline := time.Now().Add(5 * time.Second) for { _, _, inUse := poolCounts(p) if inUse >= hold { break } if time.Now().After(deadline) { t.Fatalf("workers never checked out, in_use=%d", inUse) } time.Sleep(5 * time.Millisecond) } created, _, inUse := poolCounts(p) if inUse != hold { t.Fatalf("in_use = %d, want %d", inUse, hold) } if created < hold { t.Fatalf("created = %d, must cover %d concurrent checkouts", created, hold) } if created > 16 { t.Fatalf("created = %d exceeds the ceiling 16", created) } close(release) wg.Wait() // all connections closed: the janitor collapses the pool to the floor. // shrinkGraceRounds rounds are required, and one extra round proves the // floor is stable rather than shrinking to zero. for i := 0; i < shrinkGraceRounds+2; i++ { vm.ReclaimIdleNow() } created, idle, inUse := poolCounts(p) if inUse != 0 { t.Fatalf("in_use = %d after release, want 0", inUse) } if created != residentWorkers || idle != residentWorkers { t.Fatalf("after reclaim: created=%d idle=%d, want %d/%d (resident floor)", created, idle, residentWorkers, residentWorkers) } } // TestReclaimNeedsGraceRounds: a single quiet round must not tear the pool // down — a gap between requests is not the end of a load period. func TestReclaimNeedsGraceRounds(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 8}) p := poolOf(t, vm, "openai") // build slack: 4 states created, all returned var ws []*worker for i := 0; i < 4; i++ { w, err := p.acquire() if err != nil { t.Fatalf("acquire: %v", err) } ws = append(ws, w) } for _, w := range ws { p.release(w) } created, _, _ := poolCounts(p) if created != 4 { t.Fatalf("created = %d, want 4", created) } if n := vm.ReclaimIdleNow(); n != 0 { t.Fatalf("first quiet round must not reclaim (grace), closed %d", n) } if c, _, _ := poolCounts(p); c != 4 { t.Fatalf("created = %d after the grace round, want 4", c) } if n := vm.ReclaimIdleNow(); n == 0 { t.Fatal("second quiet round must reclaim") } } // TestGrowthResetsGrace: growth (real contention) must cancel a pending shrink // decision, but ordinary sequential traffic must NOT — otherwise a busy gateway // that always has a request in flight would pin every state a past burst // created, which is exactly the leak this change removes. func TestGrowthResetsGrace(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 8}) p := poolOf(t, vm, "openai") var ws []*worker for i := 0; i < 4; i++ { w, _ := p.acquire() ws = append(ws, w) } for _, w := range ws { p.release(w) } if n := vm.ReclaimIdleNow(); n != 0 { t.Fatalf("grace round reclaimed %d", n) } // a lone sequential request reuses a warm state: no growth, so the pending // shrink decision stands and the second round reclaims w, _ := p.acquire() p.release(w) if n := vm.ReclaimIdleNow(); n == 0 { t.Fatal("sequential traffic must not keep resetting the grace counter") } // now rebuild slack and prove that GROWTH does reset it ws = ws[:0] for i := 0; i < 4; i++ { w, _ := p.acquire() // concurrent holds force created to grow ws = append(ws, w) } for _, w := range ws { p.release(w) } if n := vm.ReclaimIdleNow(); n != 0 { t.Fatalf("growth must reset the grace counter, reclaimed %d", n) } } // TestPoolCeilingIsRespectedUnderContention: acquires beyond the ceiling block // instead of booting unbounded states. func TestPoolCeilingIsRespectedUnderContention(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 2}) p := poolOf(t, vm, "openai") w1, err := p.acquire() if err != nil { t.Fatalf("acquire 1: %v", err) } w2, err := p.acquire() if err != nil { t.Fatalf("acquire 2: %v", err) } got := make(chan struct{}) go func() { w3, err := p.acquire() if err == nil { p.release(w3) } close(got) }() select { case <-got: t.Fatal("a third acquire must block at the ceiling of 2") case <-time.After(100 * time.Millisecond): } if created, _, _ := poolCounts(p); created != 2 { t.Fatalf("created = %d, must not exceed the ceiling 2", created) } p.release(w1) select { case <-got: case <-time.After(2 * time.Second): t.Fatal("releasing a state must unblock the waiter") } p.release(w2) } // TestLoweredCeilingReleasesStates: shrinking max_concurrent in the config must // actually give memory back rather than leaving orphaned states parked. func TestLoweredCeilingReleasesStates(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 8}) p := poolOf(t, vm, "openai") var ws []*worker for i := 0; i < 6; i++ { w, err := p.acquire() if err != nil { t.Fatalf("acquire: %v", err) } ws = append(ws, w) } if created, _, _ := poolCounts(p); created != 6 { t.Fatalf("created = %d, want 6", created) } // operator lowers the ceiling while the states are still checked out vm.ConfigureConcurrency(map[string]int{"openai": 2}) for _, w := range ws { p.release(w) } created, idle, _ := poolCounts(p) if created > 2 || idle > 2 { t.Fatalf("lowering the ceiling must drop the excess: created=%d idle=%d", created, idle) } } // TestPoolStatsSurfacesAlgorithm checks the observability payload the status API // exposes. func TestPoolStatsSurfacesAlgorithm(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 32}) p := poolOf(t, vm, "openai") w, err := p.acquire() if err != nil { t.Fatalf("acquire: %v", err) } var st PoolStats for _, s := range vm.PoolStats() { if s.Name == "openai" { st = s } } if st.Max != 32 { t.Fatalf("Max = %d, want 32", st.Max) } if st.InUse != 1 || st.Created != 1 { t.Fatalf("InUse=%d Created=%d, want 1/1", st.InUse, st.Created) } if st.GrowStep != 4 { t.Fatalf("GrowStep = %d, want ceil(32/8)=4", st.GrowStep) } if st.PeakInUse != 1 { t.Fatalf("PeakInUse = %d, want 1", st.PeakInUse) } p.release(w) // stats must be sorted by name for a stable UI names := []string{} for _, s := range vm.PoolStats() { names = append(names, s.Name) } for i := 1; i < len(names); i++ { if names[i-1] > names[i] { t.Fatalf("PoolStats not sorted: %v", names) } } } // TestHotReloadKeepsCeiling: replacing an adapter's code must not reset its // elastic ceiling back to 1. func TestHotReloadKeepsCeiling(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 24}) if err := vm.LoadAdapterSource("openai", "return {name='openai', version='9'}"); err != nil { t.Fatalf("reload: %v", err) } p := poolOf(t, vm, "openai") p.mu.Lock() maxW := p.maxW p.mu.Unlock() if maxW != 24 { t.Fatalf("ceiling after hot reload = %d, want 24", maxW) } } // TestConcurrentTransformStress exercises the grow/shrink paths together under // real hook calls, and asserts the accounting stays consistent (no leaked // states, no negative counters). func TestConcurrentTransformStress(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 16}) var wg sync.WaitGroup for i := 0; i < 40; i++ { wg.Add(1) go func() { defer wg.Done() raw := `{"model":"m","messages":[{"role":"user","content":"hi"}]}` if _, err := vm.Transform("openai", "transform_request", raw); err != nil { t.Errorf("transform: %v", err) } }() } wg.Wait() p := poolOf(t, vm, "openai") created, idle, inUse := poolCounts(p) if inUse != 0 { t.Fatalf("in_use = %d after the stress run, want 0 (leaked checkout)", inUse) } if created != idle { t.Fatalf("created=%d idle=%d must match when nothing is checked out", created, idle) } if created > 16 { t.Fatalf("created = %d exceeds the ceiling", created) } for i := 0; i < shrinkGraceRounds+2; i++ { vm.ReclaimIdleNow() } created, idle, _ = poolCounts(p) if created != residentWorkers || idle != residentWorkers { t.Fatalf("post-stress reclaim: created=%d idle=%d, want %d/%d", created, idle, residentWorkers, residentWorkers) } } // TestStopIsIdempotent guards the janitor's stop channel against a double close. func TestStopIsIdempotent(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } vm.Stop() vm.Stop() // must not panic } // TestSequentialTrafficDoesNotBatch: a single caller looping must keep reusing // one warm state — batch prewarm is for genuine concurrency only. func TestSequentialTrafficDoesNotBatch(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 64}) // grow step would be 8 p := poolOf(t, vm, "openai") for i := 0; i < 20; i++ { w, err := p.acquire() if err != nil { t.Fatalf("acquire %d: %v", i, err) } p.release(w) } created, _, _ := poolCounts(p) if created != 1 { t.Fatalf("sequential traffic must reuse one state, created=%d", created) } } // TestContentionBatchPrewarms: when callers are actually QUEUED behind a full // pool, the pool warms extra states for them instead of booting one per request. // The batch is bounded by the queue depth, not by the ceiling, so it can never // warm states nobody is waiting for. func TestContentionBatchPrewarms(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() // ceiling 32 -> step cap 4; queue depth will be the real bound vm.ConfigureConcurrency(map[string]int{"openai": 32}) p := poolOf(t, vm, "openai") // Fill the pool to its ceiling-independent state: hold 1, then queue 3 more // so waiting == 3 when the next miss happens. p.mu.Lock() p.maxW = 1 // force the next acquires to queue p.mu.Unlock() w1, err := p.acquire() if err != nil { t.Fatalf("acquire 1: %v", err) } done := make(chan *worker, 3) for i := 0; i < 3; i++ { go func() { w, err := p.acquire() if err != nil { done <- nil return } done <- w }() } // wait until all three are queued deadline := time.Now().Add(5 * time.Second) for { p.mu.Lock() w := p.waiting p.mu.Unlock() if w >= 3 { break } if time.Now().After(deadline) { t.Fatalf("waiters never queued, waiting=%d", w) } time.Sleep(5 * time.Millisecond) } // raise the ceiling: the next miss sees waiting=3 and prewarms a batch p.setMax(32) got := make([]*worker, 0, 3) for i := 0; i < 3; i++ { select { case w := <-done: if w == nil { t.Fatal("a queued acquire failed") } got = append(got, w) case <-time.After(5 * time.Second): t.Fatal("queued acquires never completed after the ceiling was raised") } } created, _, _ := poolCounts(p) if created < 4 { t.Fatalf("queued demand must be satisfied, created=%d want >=4", created) } if created > 32 { t.Fatalf("prewarm exceeded the ceiling, created=%d", created) } p.release(w1) for _, w := range got { p.release(w) } } // TestNoPrewarmWithoutWaiters is the production regression this bound fixes: two // concurrent requests against a wide adapter (ceiling 76, step 8) must NOT warm // eight states. Warming for nobody meant the janitor discarded the surplus a // minute later, so RSS oscillated between ~46 MB and ~56 MB forever. func TestNoPrewarmWithoutWaiters(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 76}) p := poolOf(t, vm, "openai") // two overlapping requests: the second misses while the first holds its // state, but nobody is blocked, so no batch may be warmed w1, err := p.acquire() if err != nil { t.Fatalf("acquire 1: %v", err) } w2, err := p.acquire() if err != nil { t.Fatalf("acquire 2: %v", err) } time.Sleep(100 * time.Millisecond) // give any background prewarm a chance created, _, _ := poolCounts(p) if created != 2 { t.Fatalf("two overlapping requests must create exactly 2 states, got %d", created) } p.release(w1) p.release(w2) } // TestBurstLeftoverIsReclaimedUnderSteadyTraffic reproduces what production // showed right after the first deploy: a burst grew the openai pool to 9 states, // then steady low-concurrency traffic kept arriving and the pool never shrank, // because any checkout reset the shrink grace counter. Slack must be reclaimed // even while the gateway keeps serving requests one at a time. func TestBurstLeftoverIsReclaimedUnderSteadyTraffic(t *testing.T) { vm := NewVM(freshAdapterDir(t)) if err := vm.Start(); err != nil { t.Fatalf("start: %v", err) } defer vm.Stop() vm.ConfigureConcurrency(map[string]int{"openai": 32}) p := poolOf(t, vm, "openai") // burst: 9 concurrent holds var ws []*worker for i := 0; i < 9; i++ { w, err := p.acquire() if err != nil { t.Fatalf("acquire: %v", err) } ws = append(ws, w) } for _, w := range ws { p.release(w) } if created, _, _ := poolCounts(p); created < 9 { t.Fatalf("burst should have grown the pool, created=%d", created) } // steady state: one request between every janitor round, forever for round := 0; round < 8; round++ { w, err := p.acquire() if err != nil { t.Fatalf("round %d acquire: %v", round, err) } p.release(w) vm.ReclaimIdleNow() } created, idle, inUse := poolCounts(p) if inUse != 0 { t.Fatalf("in_use = %d, want 0", inUse) } if created > residentWorkers+idleHeadroom { t.Fatalf("burst leftovers were never reclaimed under steady traffic: created=%d idle=%d", created, idle) } }