mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-09-20 08:57:57 +00:00
feat: opencode zen adapter + first-run config generation, fix stats/stream bugs
- adapters/opencode.lua: opencode.ai zen free pool adapter — sends the opencode client User-Agent (zen fingerprints clients by UA; non-official clients hit FreeUsageLimitError); pairs with api_key: public - config: no config file ships in the repo; first run generates a default config at the -config path with a random admin key, loopback listen and a keyless zen source (config.EnsureDefault); remove config.example.yaml - lua: seed bundled adapters from the embedded FS instead of a hardcoded name list - ui: widen model kind select (chat was clipped to 'cha') - phase 5 bugfixes: stats ms/s bucket mixing, cleanScopes nil, ctx.Err guards, direct-path ModelAvailable, empty stream body failure, bestImageModel rewrite, transform failure recording, Core.mu, timer, effective model for tool-calls
This commit is contained in:
@ -208,6 +208,19 @@ func (p *Provider) ModelByID(id string) *config.Model {
|
||||
return nil
|
||||
}
|
||||
|
||||
// ModelIDFold returns the configured model id matching id case-insensitively
|
||||
// ("" when no model matches). AUTO-chain slot models are normalized through
|
||||
// this so cooldown state, quota windows and the upstream model id all refer
|
||||
// to the exact configured spelling.
|
||||
func (p *Provider) ModelIDFold(id string) string {
|
||||
for _, m := range p.cfg.Models {
|
||||
if strings.EqualFold(m.ID, id) {
|
||||
return m.ID
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// ModelFor resolves the model name this provider should send upstream.
|
||||
// If the requested model is not owned by this provider (e.g. an AUTO chain
|
||||
// fallback), it returns this provider's highest-priority chat model instead.
|
||||
@ -239,6 +252,26 @@ func (p *Provider) bestChatModel() string {
|
||||
return bestID
|
||||
}
|
||||
|
||||
// bestImageModel returns the highest-priority image-kind model of this source
|
||||
// (fallback: first configured model). Used for AUTO image generation so a
|
||||
// mixed source never sends a chat model to /v1/images/generations.
|
||||
func (p *Provider) bestImageModel() string {
|
||||
bestID, bestPrio := "", -1
|
||||
for _, m := range p.cfg.Models {
|
||||
if m.Kind != "" && m.Kind != "image" {
|
||||
continue
|
||||
}
|
||||
if m.Priority > bestPrio {
|
||||
bestPrio = m.Priority
|
||||
bestID = m.ID
|
||||
}
|
||||
}
|
||||
if bestID == "" && len(p.cfg.Models) > 0 {
|
||||
bestID = p.cfg.Models[0].ID
|
||||
}
|
||||
return bestID
|
||||
}
|
||||
|
||||
// IsAutoID reports whether s is an AUTO routing placeholder.
|
||||
func isAutoID(s string) bool {
|
||||
s = strings.TrimSpace(s)
|
||||
@ -609,7 +642,11 @@ func (p *Provider) Chat(ctx context.Context, req *types.ChatRequest) (*types.Uni
|
||||
}
|
||||
raw, status, err := p.do(ctx, p.URL(), body, hdrs)
|
||||
if err != nil {
|
||||
p.RecordFailure(model, 0)
|
||||
// a client disconnect or cancelled context is neither a success nor
|
||||
// a failure for scheduling purposes — only upstream errors count
|
||||
if ctx.Err() == nil {
|
||||
p.RecordFailure(model, 0)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if status != 200 {
|
||||
@ -618,10 +655,19 @@ func (p *Provider) Chat(ctx context.Context, req *types.ChatRequest) (*types.Uni
|
||||
}
|
||||
unified, err := p.vm.Transform(p.adapter, "transform_response", raw)
|
||||
if err != nil {
|
||||
// adapter produced unusable output: a real failure the slot must
|
||||
// back off from, otherwise a broken adapter source is retried at
|
||||
// full latency forever
|
||||
if ctx.Err() == nil {
|
||||
p.RecordFailure(model, 0)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
var out types.UnifiedResponse
|
||||
if err := json.Unmarshal([]byte(unified), &out); err != nil {
|
||||
if ctx.Err() == nil {
|
||||
p.RecordFailure(model, 0)
|
||||
}
|
||||
return nil, fmt.Errorf("unmarshal unified response: %w (body: %s)", err, unified)
|
||||
}
|
||||
p.RecordSuccess(model)
|
||||
@ -664,7 +710,11 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
||||
ch := make(chan types.UnifiedChunk, 64)
|
||||
sel := <-rc
|
||||
if sel.err != nil {
|
||||
p.RecordFailure(model, 0)
|
||||
// client disconnect/cancel before the first byte: not a scheduling
|
||||
// failure (a healthy source must not be cooled by client cancellations)
|
||||
if ctx.Err() == nil {
|
||||
p.RecordFailure(model, 0)
|
||||
}
|
||||
p.Release()
|
||||
return nil, sel.err
|
||||
}
|
||||
@ -681,6 +731,8 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
||||
defer sel.resp.Body.Close()
|
||||
scanner := bufio.NewScanner(sel.resp.Body)
|
||||
scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
var chunks int
|
||||
var doneSeen bool
|
||||
for scanner.Scan() {
|
||||
line := strings.TrimSpace(scanner.Text())
|
||||
if line == "" || !strings.HasPrefix(line, "data:") {
|
||||
@ -691,6 +743,7 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
||||
continue
|
||||
}
|
||||
if data == "[DONE]" {
|
||||
doneSeen = true
|
||||
select {
|
||||
case ch <- types.UnifiedChunk{Done: true}:
|
||||
case <-ctx.Done():
|
||||
@ -711,6 +764,7 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
||||
if err := json.Unmarshal([]byte(unified), &ck); err != nil {
|
||||
continue
|
||||
}
|
||||
chunks++
|
||||
select {
|
||||
case ch <- ck:
|
||||
case <-ctx.Done():
|
||||
@ -720,9 +774,15 @@ func (p *Provider) ChatStream(ctx context.Context, req *types.ChatRequest) (<-ch
|
||||
// The stream ended cleanly ([DONE] seen or EOF without an upstream
|
||||
// read error): record success so a previously cooled model can be
|
||||
// retried. A client disconnect or mid-stream read error is neither
|
||||
// success nor failure for scheduling purposes.
|
||||
// success nor failure for scheduling purposes. A 200 that produced
|
||||
// zero chunks and no [DONE] is an empty stream, i.e. a failure
|
||||
// before the first chunk — record it so the slot can fall back.
|
||||
if ctx.Err() == nil && scanner.Err() == nil {
|
||||
p.RecordSuccess(model)
|
||||
if doneSeen || chunks > 0 {
|
||||
p.RecordSuccess(model)
|
||||
} else {
|
||||
p.RecordFailure(model, 0)
|
||||
}
|
||||
}
|
||||
}()
|
||||
return ch, nil
|
||||
@ -735,7 +795,15 @@ func (p *Provider) Image(ctx context.Context, req *types.ImageGenRequest) (*type
|
||||
return nil, err
|
||||
}
|
||||
defer p.Release()
|
||||
model := p.ModelFor(req.Model)
|
||||
// AUTO/unknown model: resolve to this source's image model, never to the
|
||||
// best chat model (a mixed source would otherwise send a chat id to
|
||||
// /v1/images/generations). An explicitly pinned model is honored as-is.
|
||||
if isAutoID(req.Model) || p.ModelByID(req.Model) == nil {
|
||||
r := *req
|
||||
r.Model = p.bestImageModel()
|
||||
req = &r
|
||||
}
|
||||
model := req.Model
|
||||
|
||||
b, _ := json.Marshal(req)
|
||||
transformed, err := p.vm.Transform(p.adapter+"_image", "transform_request", string(b))
|
||||
@ -749,7 +817,10 @@ func (p *Provider) Image(ctx context.Context, req *types.ImageGenRequest) (*type
|
||||
}
|
||||
raw, status, err := p.do(ctx, p.ImageURL(), transformed, hdrs)
|
||||
if err != nil {
|
||||
p.RecordFailure(model, 0)
|
||||
// client disconnect/cancel: not a scheduling failure
|
||||
if ctx.Err() == nil {
|
||||
p.RecordFailure(model, 0)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if status != 200 {
|
||||
@ -767,6 +838,9 @@ func (p *Provider) Image(ctx context.Context, req *types.ImageGenRequest) (*type
|
||||
}
|
||||
var img types.ImageGenResponse
|
||||
if err := json.Unmarshal([]byte(raw), &img); err != nil {
|
||||
if ctx.Err() == nil {
|
||||
p.RecordFailure(model, 0)
|
||||
}
|
||||
return nil, fmt.Errorf("unmarshal image response: %w", err)
|
||||
}
|
||||
out.ImageData = img.Data
|
||||
|
||||
@ -298,3 +298,106 @@ func TestChatBusyFailsFast(t *testing.T) {
|
||||
t.Fatalf("first chat: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func eventually(t *testing.T, timeout time.Duration, cond func() bool, msg string) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(timeout)
|
||||
for time.Now().Before(deadline) {
|
||||
if cond() {
|
||||
return
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("timed out: %s", msg)
|
||||
}
|
||||
|
||||
// TestChatClientCancelNotRecorded: a client disconnect before the response is
|
||||
// neither success nor failure — the (source, model) state must stay clean.
|
||||
func TestChatClientCancelNotRecorded(t *testing.T) {
|
||||
started := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
close(started)
|
||||
<-release // hold the upstream open; release at cleanup
|
||||
fmt.Fprint(w, `{"choices":[{"message":{"content":"late"}}]}`)
|
||||
}))
|
||||
defer up.Close()
|
||||
p := newTestProvider(t, src("mock", up.URL, "openai", "m"))
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := p.Chat(ctx, &types.ChatRequest{
|
||||
Model: "m",
|
||||
Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("x")}},
|
||||
})
|
||||
errCh <- err
|
||||
}()
|
||||
<-started
|
||||
cancel() // client disconnects mid-request
|
||||
if err := <-errCh; err == nil {
|
||||
t.Fatal("cancelled request must return an error")
|
||||
}
|
||||
close(release)
|
||||
eventually(t, 2*time.Second, func() bool {
|
||||
_, fc, _ := p.ModelHealthInfo("m")
|
||||
return fc == 0
|
||||
}, "client cancel must not record a scheduling failure")
|
||||
}
|
||||
|
||||
// TestChatStreamEmptyBodyNotSuccess: a 200 that yields zero chunks and no
|
||||
// [DONE] is a failure before the first chunk — the slot must back off.
|
||||
func TestChatStreamEmptyBodyNotSuccess(t *testing.T) {
|
||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
// 200 with an empty body: no chunks, no [DONE]
|
||||
}))
|
||||
defer up.Close()
|
||||
p := newTestProvider(t, src("mock", up.URL, "openai", "m"))
|
||||
ch, err := p.ChatStream(context.Background(), &types.ChatRequest{
|
||||
Model: "m",
|
||||
Messages: []types.ChatMessage{{Role: "user", Content: types.StringContent("x")}},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("stream: %v", err)
|
||||
}
|
||||
got := 0
|
||||
for range ch {
|
||||
got++
|
||||
}
|
||||
if got != 0 {
|
||||
t.Fatalf("want empty stream, got %d chunks", got)
|
||||
}
|
||||
eventually(t, 2*time.Second, func() bool {
|
||||
_, fc, _ := p.ModelHealthInfo("m")
|
||||
return fc >= 1
|
||||
}, "empty stream must record a failure")
|
||||
}
|
||||
|
||||
// TestImageAutoUsesImageModel: AUTO image generation on a mixed source must
|
||||
// send the image-kind model id, never the best chat model.
|
||||
func TestImageAutoUsesImageModel(t *testing.T) {
|
||||
var gotModel string
|
||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
var body map[string]interface{}
|
||||
_ = json.NewDecoder(r.Body).Decode(&body)
|
||||
gotModel, _ = body["model"].(string)
|
||||
fmt.Fprint(w, `{"created":1,"data":[{"url":"http://x/1.png"}]}`)
|
||||
}))
|
||||
defer up.Close()
|
||||
s := config.Source{Name: "mix", BaseURL: up.URL, Adapter: "openai", MaxConcurrent: 4}
|
||||
s.Models = []config.Model{
|
||||
{ID: "chat-m", Kind: "chat", Priority: 100},
|
||||
{ID: "img-m", Kind: "image", Priority: 50},
|
||||
}
|
||||
p := newTestProvider(t, s)
|
||||
resp, err := p.Image(context.Background(), &types.ImageGenRequest{Prompt: "cat", Model: "AUTO"})
|
||||
if err != nil {
|
||||
t.Fatalf("image: %v", err)
|
||||
}
|
||||
if len(resp.ImageData) == 0 {
|
||||
t.Fatal("no image data returned")
|
||||
}
|
||||
if gotModel != "img-m" {
|
||||
t.Fatalf("AUTO image must use the image-kind model, got %q", gotModel)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user