mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-10-05 07:02:29 +00:00
被问"还有 auto 调度相关 stage 呢?"问出来的真实缺口。
## 问题
chainDrive 只返回 (resp, src, model, err),调用方只知道**最终哪个槽位赢了**。
遍历过程中算出来又丢掉的东西——哪些档被跳过、为什么跳过、哪些槽位硬失败、
哪档全忙——一律不可见。ChainErr 里其实有这些,但**只在全部失败时**才填,
而它是 error 返回值不是记录。于是:
"tier 1 冷却所以降级到 tier 3" == "tier 1 正常接单"
对插件而言 tier 只是个常量 -2("resolved by the chain"),信息量为零。而这
恰恰是优先级链存在的全部理由,也是"我那个贵模型为什么没被用"的答案。
## 做法(scheduler 侧零新依赖)
新增 TraceEvent / TraceSink,chainDrive 多一个可选 sink 参数:
- TraceEvent 是本包的普通 struct,sink 是 func 参数 ⇒ **不新增 import**,
scheduler 仍然可独立测试
- sink 为 nil 时每次 emit 只多一次 nil 判断;没有插件的网关在 AUTO 热路径上
零开销(gateway 的 chainTraceSink 直接返回 nil)
- 事件是纯观测:scheduler 不基于它做任何分支,gateway 也不把它喂回路由/
冷却/配额
四种 kind:tier_skip / slot_fail / tier_busy / selected,selected 每次成功
遍历恰好一次且是最后一步。顺序保证所有 step 在 routed 之前。
## 暴露给插件
新增 chain_step stage(逐个步骤),并在 request_end 载荷里加三个便于做报表的
字段:chain_walk(上限 12 步,防审计记录膨胀)、degraded、tier_served。
## ★ 计费口径(我按推荐的做,已写进文档,需要你确认)
**按实际服务的模型计费**:降级到 tier 3 仍按 tier 3 的价算,轨迹只作观测。
理由与 §7.5 的边界一致——插件只报表不执法,两套口径混在一起会引出"降级该不该
多收钱"这种无法从代码判断的争议。若要改成"按本该用的档计价",需要在 models
价目里允许按 tier 定价,这我没做,因为那是个产品决策。
## 计费插件同步消费
by_tier_served / skip_reasons / degraded_reqs 三个新维度。skip_reasons 的等待
时长做了归一(`no free slot within <wait>`),否则 busy-wait 文案一变就多一行。
降级次数在 request_end 里计而不是在 chain_step 里计:一次降级的请求要走多步,
按步计会重复计数。
## 判据(346 个测试全绿,新增 15 个)
scheduler 6 个:正常路径只发一个 selected / 跳档+降级可见 / 硬失败与跳档
严格区分(不可混为一谈,否则抖动上游看起来像空闲上游)/
nil sink 安全 / 全失败时轨迹与 ChainErr 并存且不互相破坏 /
空链不发事件
gateway 1 个端到端:tier 1 全 500 → 插件收到 slot_fail(tier 1) +
selected(tier 2),request_end 的 tier_served=2 且 degraded=true
lua 2 个:降级计数与按实际模型计价 / 跳过原因归一聚合
lua 1 个:chain_step 是真 stage 且顺序正确
3 个变异都红:去掉 slot_fail(3 个判据红)/ 去掉 tier_skip(1 个)/
去掉 degraded 字段(1 个)。
374 lines
10 KiB
Go
374 lines
10 KiB
Go
package lua
|
|
|
|
import (
|
|
"encoding/json"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"testing"
|
|
)
|
|
|
|
// newPluginVM builds a VM plus a plugin registry rooted at dir.
|
|
func newPluginVM(t *testing.T) (*VM, *Plugins, string) {
|
|
t.Helper()
|
|
dir := filepath.Join(t.TempDir(), "adapters")
|
|
vm := NewVM(dir)
|
|
if err := vm.Start(); err != nil {
|
|
t.Fatalf("vm start: %v", err)
|
|
}
|
|
t.Cleanup(vm.Stop)
|
|
pdir := filepath.Join(t.TempDir(), "plugins")
|
|
return vm, NewPlugins(vm, pdir), pdir
|
|
}
|
|
|
|
// loadPlugin writes one plugin to disk and loads it.
|
|
func loadPlugin(t *testing.T, ps *Plugins, name, code string) error {
|
|
t.Helper()
|
|
if err := os.MkdirAll(ps.dir, 0755); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := os.WriteFile(filepath.Join(ps.dir, name+".lua"), []byte(code), 0644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return ps.LoadSource(name, code)
|
|
}
|
|
|
|
// TestPluginManifestAndHooks: the two hook registration forms both work and the
|
|
// manifest is read.
|
|
func TestPluginManifestAndHooks(t *testing.T) {
|
|
_, ps, _ := newPluginVM(t)
|
|
code := `
|
|
local p = {}
|
|
p.name = "demo"
|
|
p.version = "1.2.3"
|
|
p.description = "a demo plugin"
|
|
p.author = "tester"
|
|
p.hooks = { request_end = "on_end" }
|
|
function p.on_end(payload)
|
|
payload.seen = true
|
|
payload.name_seen = "demo"
|
|
return payload
|
|
end
|
|
return p
|
|
`
|
|
if err := loadPlugin(t, ps, "demo", code); err != nil {
|
|
t.Fatalf("load: %v", err)
|
|
}
|
|
list := ps.List()
|
|
if len(list) != 1 {
|
|
t.Fatalf("List() = %d plugins, want 1", len(list))
|
|
}
|
|
if list[0]["name"] != "demo" || list[0]["version"] != "1.2.3" {
|
|
t.Errorf("manifest not read: %+v", list[0])
|
|
}
|
|
hooks := list[0]["hooks"].([]string)
|
|
if len(hooks) != 1 || hooks[0] != string(StageRequestEnd) {
|
|
t.Errorf("hooks = %v, want [request_end]", hooks)
|
|
}
|
|
out := ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m"})
|
|
if out["seen"] != true || out["name_seen"] != "demo" {
|
|
t.Errorf("hook did not mutate payload: %+v", out)
|
|
}
|
|
}
|
|
|
|
// TestPluginAnonymousHookForm: `request_end = function() end` directly on the
|
|
// table must register too, since a single-hook plugin should not need a name.
|
|
func TestPluginAnonymousHookForm(t *testing.T) {
|
|
_, ps, _ := newPluginVM(t)
|
|
code := `
|
|
local p = { name = "anon" }
|
|
p.request_end = function(payload)
|
|
payload.hit = 1
|
|
return payload
|
|
end
|
|
return p
|
|
`
|
|
if err := loadPlugin(t, ps, "anon", code); err != nil {
|
|
t.Fatalf("load: %v", err)
|
|
}
|
|
out := ps.Fire(StageRequestEnd, map[string]interface{}{})
|
|
if out["hit"] != float64(1) {
|
|
t.Errorf("anonymous hook did not fire: %+v", out)
|
|
}
|
|
}
|
|
|
|
// TestPluginHookStagesFireInOrder: each stage reaches only its own hooks.
|
|
func TestPluginHookStagesFireInOrder(t *testing.T) {
|
|
_, ps, _ := newPluginVM(t)
|
|
code := `
|
|
local p = { name = "stages" }
|
|
p.hooks = {
|
|
request_start = "s1",
|
|
routed = "s2",
|
|
request_end = "s3",
|
|
}
|
|
function p.s1(x) x.order = (x.order or "") .. "1" return x end
|
|
function p.s2(x) x.order = (x.order or "") .. "2" return x end
|
|
function p.s3(x) x.order = (x.order or "") .. "3" return x end
|
|
return p
|
|
`
|
|
if err := loadPlugin(t, ps, "stages", code); err != nil {
|
|
t.Fatalf("load: %v", err)
|
|
}
|
|
payload := map[string]interface{}{}
|
|
ps.Fire(StageRequestStart, payload)
|
|
ps.Fire(StageRouted, payload)
|
|
ps.Fire(StageRequestEnd, payload)
|
|
if payload["order"] != "123" {
|
|
t.Errorf("stage order = %v, want \"123\"", payload["order"])
|
|
}
|
|
}
|
|
|
|
// TestPluginErrorIsContained is the critical safety property: a throwing hook
|
|
// must not propagate. Forwarding depends on it.
|
|
func TestPluginErrorIsContained(t *testing.T) {
|
|
_, ps, _ := newPluginVM(t)
|
|
code := `
|
|
local p = { name = "boom" }
|
|
p.hooks = { request_end = "kaboom" }
|
|
function p.kaboom(payload)
|
|
error("intentional plugin failure")
|
|
end
|
|
return p
|
|
`
|
|
if err := loadPlugin(t, ps, "boom", code); err != nil {
|
|
t.Fatalf("load: %v", err)
|
|
}
|
|
// Must not panic and must return the payload unchanged.
|
|
out := ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m"})
|
|
if out["model"] != "m" {
|
|
t.Errorf("payload was altered by a failing plugin: %+v", out)
|
|
}
|
|
// And the failure must be visible, not silent.
|
|
errs := ps.HookErrors()
|
|
if errs["request_end"] == nil {
|
|
t.Error("a failing plugin left no error record; it would be silently missing")
|
|
}
|
|
}
|
|
|
|
// TestPluginFailingHookDoesNotBlockLaterPlugins: one bad plugin must not stop
|
|
// the next one from running.
|
|
func TestPluginFailingHookDoesNotBlockLaterPlugins(t *testing.T) {
|
|
_, ps, _ := newPluginVM(t)
|
|
bad := `
|
|
local p = { name = "bad" }
|
|
p.hooks = { request_end = "f" }
|
|
function p.f(x) error("boom") end
|
|
return p
|
|
`
|
|
good := `
|
|
local p = { name = "good" }
|
|
p.hooks = { request_end = "f" }
|
|
function p.f(x) x.good = true return x end
|
|
return p
|
|
`
|
|
_ = loadPlugin(t, ps, "bad", bad)
|
|
if err := loadPlugin(t, ps, "good", good); err != nil {
|
|
t.Fatalf("load good: %v", err)
|
|
}
|
|
out := ps.Fire(StageRequestEnd, map[string]interface{}{})
|
|
if out["good"] != true {
|
|
t.Errorf("a good plugin was blocked by a failing one: %+v", out)
|
|
}
|
|
}
|
|
|
|
// TestPluginSyntaxErrorIsIsolated: a plugin that will not compile is listed
|
|
// with its error and is never called — it must not prevent LoadDir from loading
|
|
// the rest.
|
|
func TestPluginSyntaxErrorIsIsolated(t *testing.T) {
|
|
_, ps, _ := newPluginVM(t)
|
|
broken := "this is not lua((("
|
|
good := `
|
|
local p = { name = "ok" }
|
|
p.hooks = { request_end = "f" }
|
|
function p.f(x) x.ok = true return x end
|
|
return p
|
|
`
|
|
_ = loadPlugin(t, ps, "broken", broken)
|
|
if err := loadPlugin(t, ps, "ok", good); err != nil {
|
|
t.Fatalf("load ok: %v", err)
|
|
}
|
|
if err := ps.LoadDir(); err != nil {
|
|
t.Fatalf("LoadDir: %v", err)
|
|
}
|
|
// The broken plugin must not be callable and must carry an error.
|
|
for _, row := range ps.List() {
|
|
if row["name"] == "broken" {
|
|
if row["loaded"] == true {
|
|
t.Error("a plugin with a syntax error reported itself as loaded")
|
|
}
|
|
if row["error"] == nil || row["error"] == "" {
|
|
t.Error("a broken plugin carries no error message")
|
|
}
|
|
}
|
|
}
|
|
// The good plugin still works.
|
|
out := ps.Fire(StageRequestEnd, map[string]interface{}{})
|
|
if out["ok"] != true {
|
|
t.Errorf("good plugin stopped working: %+v", out)
|
|
}
|
|
}
|
|
|
|
// TestPluginUIExtension: a plugin can contribute a page and elements.
|
|
func TestPluginUIExtension(t *testing.T) {
|
|
_, ps, _ := newPluginVM(t)
|
|
code := `
|
|
local p = { name = "ui" }
|
|
p.hooks = { request_end = "f" }
|
|
function p.f(x) return x end
|
|
p.ui = {
|
|
page = {
|
|
page_id = "billing",
|
|
title = "Billing",
|
|
icon = "💰",
|
|
order = 50,
|
|
mount = "<div id=billing>hi</div><script>console.log('m')</script>",
|
|
},
|
|
elements = {
|
|
{ target = "status", anchor = "top", mount = "<div>cost</div>" },
|
|
},
|
|
}
|
|
return p
|
|
`
|
|
if err := loadPlugin(t, ps, "ui", code); err != nil {
|
|
t.Fatalf("load: %v", err)
|
|
}
|
|
ui := ps.UI()
|
|
if ui.Page == nil {
|
|
t.Fatal("no page contributed")
|
|
}
|
|
if ui.Page.PageID != "billing" || ui.Page.Title != "Billing" {
|
|
t.Errorf("page = %+v", ui.Page)
|
|
}
|
|
if !strings.Contains(ui.Page.Mount, "console.log") {
|
|
t.Error("mount lost its script content")
|
|
}
|
|
if len(ui.Elements) != 1 || ui.Elements[0].Target != "status" {
|
|
t.Errorf("elements = %+v", ui.Elements)
|
|
}
|
|
}
|
|
|
|
// TestPluginHookReturnsNilIsNoOpinion: a hook returning nothing must leave the
|
|
// payload untouched (plugins should not be forced to echo it back).
|
|
func TestPluginHookReturnsNilIsNoOpinion(t *testing.T) {
|
|
_, ps, _ := newPluginVM(t)
|
|
code := `
|
|
local p = { name = "silent" }
|
|
p.hooks = { request_end = "f" }
|
|
function p.f(payload)
|
|
-- records nothing, returns nothing
|
|
return nil
|
|
end
|
|
return p
|
|
`
|
|
if err := loadPlugin(t, ps, "silent", code); err != nil {
|
|
t.Fatalf("load: %v", err)
|
|
}
|
|
out := ps.Fire(StageRequestEnd, map[string]interface{}{"model": "m", "ok": true})
|
|
if out["model"] != "m" || out["ok"] != true {
|
|
t.Errorf("a no-op hook disturbed the payload: %+v", out)
|
|
}
|
|
}
|
|
|
|
// TestPluginUIJSONShape is a wire-format guard: the kernel sends this to the
|
|
// browser, so the shape is a contract with the WebUI.
|
|
func TestPluginUIJSONShape(t *testing.T) {
|
|
_, ps, _ := newPluginVM(t)
|
|
code := `
|
|
local p = { name = "shape" }
|
|
p.hooks = { request_end = "f" }
|
|
function p.f(x) return x end
|
|
p.ui = { elements = { { target = "keys", mount = "<b>k</b>" } } }
|
|
return p
|
|
`
|
|
if err := loadPlugin(t, ps, "shape", code); err != nil {
|
|
t.Fatalf("load: %v", err)
|
|
}
|
|
b, err := json.Marshal(ps.UI())
|
|
if err != nil {
|
|
t.Fatalf("marshal UI: %v", err)
|
|
}
|
|
var view struct {
|
|
Elements []struct {
|
|
Target string `json:"target"`
|
|
Mount string `json:"mount"`
|
|
} `json:"elements"`
|
|
}
|
|
if err := json.Unmarshal(b, &view); err != nil {
|
|
t.Fatalf("unmarshal: %v", err)
|
|
}
|
|
if len(view.Elements) != 1 || view.Elements[0].Target != "keys" {
|
|
t.Errorf("UI JSON shape = %+v", view.Elements)
|
|
}
|
|
}
|
|
|
|
// TestPluginFireWithNoPluginsIsNoop: an empty registry must not allocate or fail.
|
|
func TestPluginFireWithNoPluginsIsNoop(t *testing.T) {
|
|
_, ps, _ := newPluginVM(t)
|
|
in := map[string]interface{}{"a": 1}
|
|
out := ps.Fire(StageRequestEnd, in)
|
|
if out["a"] != 1 || ps.Count() != 0 {
|
|
t.Errorf("empty registry misbehaved: %+v", out)
|
|
}
|
|
}
|
|
|
|
// TestChainStepIsARealStage: chain_step is documented as a distinct stage that
|
|
// fires once per step of an AUTO walk. If it were only a field on `routed`, a
|
|
// plugin author following the docs would silently get one event instead of the
|
|
// whole walk.
|
|
func TestChainStepIsARealStage(t *testing.T) {
|
|
found := false
|
|
for _, s := range AllStages {
|
|
if s == StageChainStep {
|
|
found = true
|
|
}
|
|
}
|
|
if !found {
|
|
t.Fatal("StageChainStep is not in AllStages, so the dispatcher never registers it")
|
|
}
|
|
// Ordering: it must sit between request_start and routed, which is what
|
|
// docs/plugins.md promises.
|
|
var iStart, iStep, iRouted = -1, -1, -1
|
|
for i, s := range AllStages {
|
|
switch s {
|
|
case StageRequestStart:
|
|
iStart = i
|
|
case StageChainStep:
|
|
iStep = i
|
|
case StageRouted:
|
|
iRouted = i
|
|
}
|
|
}
|
|
if !(iStart < iStep && iStep < iRouted) {
|
|
t.Errorf("stage order = %v, want request_start < chain_step < routed", AllStages)
|
|
}
|
|
_, ps, _ := newPluginVM(t)
|
|
code := `
|
|
local p = { name = "stepper" }
|
|
p.hooks = { chain_step = "s" }
|
|
function p.s(payload)
|
|
payload.kinds = (payload.kinds or "")
|
|
return nil
|
|
end
|
|
return p
|
|
`
|
|
if err := loadPlugin(t, ps, "stepper", code); err != nil {
|
|
t.Fatalf("load: %v", err)
|
|
}
|
|
// It must actually dispatch.
|
|
seen := false
|
|
for _, row := range ps.List() {
|
|
if row["name"] == "stepper" {
|
|
hooks := row["hooks"].([]string)
|
|
for _, h := range hooks {
|
|
if h == string(StageChainStep) {
|
|
seen = true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if !seen {
|
|
t.Error("a plugin registered for chain_step is not reported as such")
|
|
}
|
|
}
|