chore(gui): 版本号 1.9.5 → 1.9.6

## 症状

生产 Quota 页面上 17 个源里 13 个显示「有数据」徽标 —— 而它们实际上什么都
没返回，只是刚刚被退避跳过。

## 根因（三层，逐层都静默）

**① 退避记录缺 `skipped` 标记**
`due_sources` 与 `collectSource` 返回退避记录时没有 `skipped = true`，页面的
`filter(r => !r.skipped)` 就把它们当正常记录放进卡片。卡片渲染匹配已知字段，
一条只有 `reason` 的记录什么都匹配不上，于是落到兜底分支。

**② 兜底分支文案是「有数据」**
那是个断言 —— 它在说「探测有应答」。落到这里的记录其实是：一条退避通知、
一种我们不认识的错误形状、或供应商改了响应格式。把「没有证据」说成「有数据」，
十三个死上游就被渲染成健康的了。改成「未探测」。

**③ 退避与「没有余额接口」被混为一谈**
页脚原本统一说「未暴露余额接口」。这是对上游的另一种断言，而且对死上游是错的
—— 它有接口，只是暂时没探到。

## 「只改了一处」这个坑

`render()` 改成按 `backing_off` 标记分组之后，两个调用点都必须传**记录**而不是
名字。我先改了刷新路径就部署，结果是页脚显示「未列出 13 个源：、、、……」——
症状看起来像渲染 bug，实际是两个调用点与同一个函数之间的契约不一致。

快照路径那行 `.map(function(r){ return r.name; })` 是后来才补的。
判据 `TestQuotaRenderCallSitesPassRecordsNotNames` 现在要求两处都在。

## 顺带修掉的两个真 bug

**空 Lua 表在 Go 侧解码成 `{}` 而不是 `[]`**
`plan.backing_off || []` 里 `{}` 是 truthy，空表会原样进入 `out.concat()` 并抛错，
整次刷新报失败。改用 `Array.isArray` —— 它与插件实际返回的类型匹配。

**分组靠文案匹配是个定时炸弹**
原先用 `reason.indexOf("backing off") === 0` 判断退避。理由字符串一旦本地化，
退避就会被并进「未暴露余额接口」。改用显式 `backing_off: true` 标记，
判据里也改成按标记匹配 —— 变异把文案改回英文立刻变红。

## 判据

新增两条：
- `TestQuotaBackoffRecordIsSkippedAndFlagged` —— 退避记录必须带 `skipped` 和
  `backing_off` 标记，且必须有 reason（否则卡片是空的）
- `TestQuotaRenderCallSitesPassRecordsNotNames` —— 两个调用点都不得把跳过列表
  映射成名字

变异自证四次全部变红：去掉 `skipped`、去掉 `backing_off`、文案改回英文、
快照路径改回 map-to-name。

## 生产实测（浏览器真实点击）

```
第1轮 1694ms: 4 个源 · 0 个读取失败
        13 个源正在退避（连续探测失败，等下次重试）：trae、qijiar、toter、…
第2轮 2488ms: 4 个源 · 0 个读取失败
        13 个源正在退避（…）：trae、qijiar、toter、…
```

页面现在只列 4 个真有数据的源，13 个退避中的归到页脚并说明原因。

Co-Authored-By: ModelRouter <noreply@modelrouter.dev>

cmd/gui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "modelrouter-gui",
-  "version": "1.9.4",
+  "version": "1.9.6",
   "description": "ModelRouter Desktop — embedded ModelRouter core with tray",
   "author": "ModelRouter",
   "main": "main.js",

internal/gateway/plugins_api.go

@@ -117,6 +117,20 @@ func (g *Gateway) handlePluginsAPI(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	// /api/plugins/{name}/fanout/{fn} — one call per argument set, in
+	// parallel, from one request. See handlePluginFanout for why the
+	// request-per-item form cannot be made concurrent from the client.
+	if i := strings.Index(path, "/fanout/"); i >= 0 {
+		name := path[:i]
+		fn := path[i+len("/fanout/"):]
+		if name == "" || fn == "" {
+			writeError(w, http.StatusBadRequest, "invalid_request", "fanout needs a plugin and a function name")
+			return
+		}
+		g.handlePluginFanout(w, r, name, fn)
+		return
+	}
+
 	if reqRole(r.Context()) != "admin" {
 		writeError(w, http.StatusForbidden, "forbidden", "admin role required")
 		return
@@ -133,7 +147,7 @@ func (g *Gateway) handlePluginsAPI(w http.ResponseWriter, r *http.Request) {
 				// Visibility overrides and the endpoint access table, so the
 				// plugin page renders what is IN FORCE rather than making the
 				// operator read .lua files or config.yaml to find out.
-				"visibility":   ps.VisibilityOverrides(),
+				"visibility":  ps.VisibilityOverrides(),
 				"http_access": ps.EndpointAccess(),
 			})
 		case http.MethodPost:
@@ -331,6 +345,68 @@ func (g *Gateway) handlePluginCall(w http.ResponseWriter, r *http.Request, name,
 	writeJSON(w, http.StatusOK, map[string]interface{}{"result": result})
 }
 
+// handlePluginFanout runs one plugin export per element of a list, in parallel,
+// from a SINGLE request.
+//
+// Why this exists: the quota page has twenty cards, and it used to fire twenty
+// browser requests at /call/. They arrived concurrently and then serialised
+// inside Plugins.Call, which holds the plugin mutex across the Lua function —
+// including its network wait. Measured on the live gateway: 20 probes took 29s of
+// wall clock while a single probe with no contention took 1.4–3.3s, so the page
+// sat on "probing…" for half a minute. The browser-side concurrency was real and
+// made no difference.
+//
+// A fan-out cannot be assembled from the client either: any request-per-item
+// arrangement is exactly the shape that queues. The parallelism has to happen
+// where the fan-out is received.
+func (g *Gateway) handlePluginFanout(w http.ResponseWriter, r *http.Request, name, fn string) {
+	if reqRole(r.Context()) != "admin" {
+		writeError(w, http.StatusForbidden, "forbidden", "admin role required")
+		return
+	}
+	if r.Method != http.MethodPost {
+		writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "")
+		return
+	}
+	var req struct {
+		// Args is a LIST of argument objects, one per call. A bare object is
+		// rejected rather than coerced: "args" meaning "one call" and "args"
+		// meaning "many" depending on its shape is how a caller ends up probing
+		// one source and believing it probed twenty.
+		Args []map[string]interface{} `json:"args"`
+	}
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		writeError(w, http.StatusBadRequest, "invalid_request", "invalid json: "+err.Error())
+		return
+	}
+	if len(req.Args) == 0 {
+		writeError(w, http.StatusBadRequest, "invalid_request", "args must be a non-empty list of argument objects")
+		return
+	}
+	// The page legitimately has one card per configured source. The bound is not
+	// arbitrary: it stops one request from asking for thousands of VMs, and it is
+	// enforced before any work starts rather than discovered shard by shard.
+	if len(req.Args) > fanoutMaxArgs {
+		writeError(w, http.StatusBadRequest, "invalid_request",
+			fmt.Sprintf("args has %d entries, limit is %d", len(req.Args), fanoutMaxArgs))
+		return
+	}
+	results, err := g.core.Plugins().ShardCall(name, fn, req.Args)
+	if err != nil {
+		writeError(w, http.StatusBadGateway, "plugin_error", err.Error())
+		return
+	}
+	// A shard that failed contributes a null in its slot. That is deliberate and
+	// is NOT the same as the whole request failing: one unreachable upstream must
+	// not turn a page of twenty cards into a single error.
+	writeJSON(w, http.StatusOK, map[string]interface{}{"results": results})
+}
+
+// fanoutMaxArgs bounds one fan-out request. Twenty is the measured real source
+// count on the gateway this was built for; the headroom is for operators with
+// more sources configured.
+const fanoutMaxArgs = 64
+
 func (g *Gateway) pluginDir() string {
 	if c := g.core.Config(); c != nil {
 		return c.PluginDir
@@ -369,7 +445,6 @@ func (g *Gateway) removePlugin(name string) error {
 	return g.core.Plugins().Unload(name)
 }
 
-
 // handleEndpointAccess serves the operator's view and edit surface for one
 // plugin endpoint's reachable keys.
 //
@@ -439,7 +514,6 @@ func (g *Gateway) handleEndpointAccess(w http.ResponseWriter, r *http.Request, n
 	}
 }
 
-
 // splitStateForKey rewrites a plugin state so that a non-admin caller sees only
 // the buckets that belong to them.
 //

internal/gateway/ui/index.html

@@ -5781,6 +5781,38 @@ toast(t("kAutoChainGlobal"));
           if (!r.ok) throw new Error(body.error?.message || r.status);
           return body.result;
         },
+        // fanout runs one of the plugin's exported functions once per argument
+        // set, IN PARALLEL, from a single request.
+        //
+        // `call` in a Promise.all is NOT this. Twenty concurrent /call/ requests
+        // arrive concurrently and then serialise on the plugin mutex inside the
+        // kernel — the lock is held across the Lua function, network wait
+        // included, because one Lua VM has one stack. Measured on a live
+        // gateway: twenty quota probes took 29s of wall clock through /call/,
+        // against 1.4–3.3s for a single uncontended probe. The browser-side
+        // concurrency bought nothing and the page looked simply broken.
+        //
+        // The parallelism has to happen where the fan-out is received, so it
+        // takes one request. Results come back positionally aligned with `argSets`;
+        // a failed item is null in its slot rather than an error for the batch,
+        // because one unreachable upstream must not blank a whole page of cards.
+        //
+        // Same authorization as call: admin-only, and the function must be in
+        // the plugin's exports.
+        async fanout(name, fn, argSets) {
+          const r = await fetch(
+            "/api/plugins/" + encodeURIComponent(name) + "/fanout/" + encodeURIComponent(fn),
+            {
+              method: "POST",
+              credentials: "same-origin",
+              headers: { "Content-Type": "application/json" },
+              body: JSON.stringify({ args: argSets || [] }),
+            },
+          );
+          const body = await r.json().catch(() => ({}));
+          if (!r.ok) throw new Error(body.error?.message || r.status);
+          return body.results || [];
+        },
         onTabShown(fn) {
           PLUGIN_TAB_CBS.__last = PLUGIN_TAB_CBS.__last || [];
           PLUGIN_TAB_CBS.__last.push(fn);

internal/lua/plugin_access.go

@@ -0,0 +1,180 @@
+package lua
+
+import "strings"
+
+// One vocabulary for "who may see / read / call this".
+//
+// Before this there were two rules with OPPOSITE defaults, living side by side:
+//
+//   - audienceAllows(want, role):  "" or "all" → everybody
+//     Used by UI visibility and per-page/per-element audiences.
+//   - HTTPHandler.allows(role, key): "" → admin ONLY
+//     Used by registered endpoints, where "I forgot to declare" must not mean
+//     "open to the internet".
+//
+// Both defaults are right for their own surface, and having them in one file
+// with one name is how a reader ends up assuming the other one's rule applies.
+// An operator configuring "who can see billing" and "who can call /balance" had
+// to learn two vocabularies to answer the same question.
+//
+// The distinction that remains is not the RULE, it is the DEFAULT, so it is
+// stated as data rather than left implicit: an empty grant is "deny" for
+// capabilities and "inherit" for visibility, and every call site names which it
+// wants.
+
+// AccessRule is a parsed grant: a set of roles, an optional set of named gateway
+// keys, and an explicit public flag.
+//
+// Keys REPLACE roles rather than adding to them. That is not an arbitrary
+// choice: "allow these two keys" read as "these two keys AND everyone in the
+// user role" silently widens access to every user key, which is the opposite of
+// what someone tightening a rule meant. `keys = { "@alice" }` therefore means
+// exactly alice.
+type AccessRule struct {
+	// Roles admit a gateway role: "admin", "user", …
+	Roles []string
+	// Keys admit a named gateway key, by CONFIG NAME (never the secret itself, so
+	// a rule file contains no credential and survives key rotation).
+	Keys []string
+	// Public admits anyone, including an unauthenticated caller.
+	Public bool
+}
+
+// AccessAll is the grant that admits everyone.
+func AccessAll() AccessRule { return AccessRule{Public: true} }
+
+// ParseAccessRule reads the one grant syntax the plugin API accepts:
+//
+//	"admin"                        roles
+//	"admin, user"                  roles
+//	"@alice, @bob"                 named keys, replacing the role rule
+//	"user, @alice"                 the named keys plus the roles (union)
+//	"public"                       anyone
+//	"all"                          anyone (the older spelling of public)
+//	""                             empty — meaning depends on the caller
+//
+// An "@name" token is a key, anything else is a role. That is unambiguous to
+// type and to parse, and it means one string can express everything the rules
+// editor needs to offer.
+//
+// "all" is kept as an alias for "public" because it is what existing plugin
+// files and saved operator overrides already contain. Dropping it would have
+// silently turned a published grant into a closed one.
+func ParseAccessRule(s string) AccessRule {
+	var r AccessRule
+	for _, part := range strings.Split(s, ",") {
+		part = strings.TrimSpace(part)
+		if part == "" {
+			continue
+		}
+		if strings.EqualFold(part, "public") || strings.EqualFold(part, "all") {
+			r.Public = true
+			continue
+		}
+		if strings.HasPrefix(part, "@") {
+			if k := strings.TrimSpace(strings.TrimPrefix(part, "@")); k != "" {
+				r.Keys = append(r.Keys, k)
+			}
+			continue
+		}
+		r.Roles = append(r.Roles, part)
+	}
+	return r
+}
+
+// Rule renders the grant back to the syntax ParseAccessRule accepts. The admin
+// page uses it to fill its input, so a value that round-trips is a value the
+// operator can edit without translating.
+func (r AccessRule) Rule() string {
+	var parts []string
+	for _, k := range r.Keys {
+		parts = append(parts, "@"+k)
+	}
+	for _, role := range r.Roles {
+		parts = append(parts, role)
+	}
+	if r.Public {
+		parts = append(parts, "public")
+	}
+	return strings.Join(parts, ", ")
+}
+
+// Empty reports whether the grant names nobody.
+func (r AccessRule) Empty() bool {
+	return len(r.Roles) == 0 && len(r.Keys) == 0 && !r.Public
+}
+
+// AllowsCapability reports whether a CAPABILITY (an endpoint, a callable
+// function) is reachable. Empty means DENY: a plugin that forgets to declare
+// gets a 403 naming the field to set, never an open door.
+//
+// Admin is always admitted. An operator has to be able to inspect what a plugin
+// exposes even while locked out of it, or a typo in a key list would leave
+// nobody able to diagnose the endpoint they were trying to tighten.
+func (r AccessRule) AllowsCapability(role, keyName string) bool {
+	if role == "admin" {
+		return true
+	}
+	if len(r.Keys) > 0 {
+		// A non-empty key list REPLACES the role rule (see the type comment).
+		// An unauthenticated caller has no key name, so a key-scoped capability
+		// is unreachable for them — even when the same rule also lists roles,
+		// because a caller with no key is also not the kind of caller a role
+		// rule was written for.
+		if keyName == "" {
+			return false
+		}
+		for _, k := range r.Keys {
+			if strings.EqualFold(k, keyName) {
+				return true
+			}
+		}
+		// The key list replaced the role rule, so a miss is a deny even though
+		// roles may also be listed.
+		return false
+	}
+	if r.Public {
+		return true
+	}
+	for _, want := range r.Roles {
+		if strings.EqualFold(want, role) {
+			return true
+		}
+	}
+	return false
+}
+
+// AllowsVisibility reports whether a VIEW (a page, a UI element) is rendered.
+// Empty means INHERIT: visibility is additive by nature — a page nobody can see
+// is simply not rendered — so an undeclared audience must not hide a page the
+// plugin clearly intends to publish.
+func (r AccessRule) AllowsVisibility(role string) bool {
+	if r.Empty() || r.Public {
+		return true
+	}
+	if role == "admin" && len(r.Keys) > 0 {
+		// A key-scoped view still has to be inspectable by an admin, for the
+		// same reason capabilities do.
+		return true
+	}
+	for _, want := range r.Roles {
+		if strings.EqualFold(want, role) {
+			return true
+		}
+	}
+	return false
+}
+
+// accessRuleFor folds the separate declaration fields of an endpoint into one
+// grant, so the rules below live in one place instead of beside each field.
+//
+// It is a plain fold, deliberately. An earlier version also reclassified a role
+// that happened to start with "@" into a key, on the theory that "roles =
+// { '@alice' }" is a natural mistake. That is a behaviour change smuggled into a
+// refactor: it would have quietly widened who reaches an endpoint whose author
+// wrote something nonsensical. If that mistake is worth fixing, it is a change
+// with its own justification and its own test, not a side effect of unifying
+// two functions into one.
+func accessRuleFor(roles, keys []string, public bool) AccessRule {
+	return AccessRule{Roles: roles, Keys: keys, Public: public}
+}

internal/lua/plugin_access_test.go

@@ -0,0 +1,152 @@
+package lua
+
+import (
+	"strings"
+	"testing"
+)
+
+// The two functions this file replaces had OPPOSITE defaults:
+//
+//	audienceAllows:  "" / "all"  → everybody
+//	HTTPHandler.allows: ""         → admin ONLY
+//
+// Both are correct for their surface. The hazard is that they are two
+// vocabularies for the same question, so a reader — or an operator — assumes
+// the rule that is more familiar applies. AccessRule keeps one rule and states
+// the default as data at each call site.
+
+// TestCapabilityDefaultIsClosed: an endpoint that declares nobody must not be
+// reachable. This is the property that makes "I forgot" fail safe.
+func TestCapabilityDefaultIsClosed(t *testing.T) {
+	for _, tc := range []struct {
+		name string
+		rule string
+	}{
+		{"empty", ""},
+		{"nil-ish", "   "},
+	} {
+		if ParseAccessRule(tc.rule).AllowsCapability("user", "alice") {
+			t.Errorf("%s: an undeclared capability is reachable by a user", tc.name)
+		}
+		if !ParseAccessRule(tc.rule).AllowsCapability("admin", "alice") {
+			t.Errorf("%s: admin locked out of a capability they must be able to inspect", tc.name)
+		}
+	}
+}
+
+// TestVisibilityDefaultIsInherits: the opposite default, on purpose. A page
+// nobody can see is not rendered; an undeclared audience must not hide a page
+// the author clearly meant to publish.
+func TestVisibilityDefaultIsInherits(t *testing.T) {
+	if !ParseAccessRule("").AllowsVisibility("user") {
+		t.Error("an undeclared audience hides a page")
+	}
+	if ParseAccessRule("admin").AllowsVisibility("user") {
+		t.Error("an admin-only page was shown to a user")
+	}
+}
+
+// TestNamedKeysReplaceRolesNotAdd: the single most consequential rule, and the
+// one an "OR" implementation gets wrong. "allow these two keys" must not also
+// admit every user key.
+func TestNamedKeysReplaceRolesNotAdd(t *testing.T) {
+	r := ParseAccessRule("@alice, @bob")
+	if !r.AllowsCapability("user", "alice") {
+		t.Error("a named key was not admitted")
+	}
+	if r.AllowsCapability("user", "carol") {
+		t.Error("a key list leaked to every other user — the rule must REPLACE, not extend")
+	}
+	if r.AllowsCapability("user", "") {
+		t.Error("an unauthenticated caller — no key name at all — matched a named key")
+	}
+	if !r.AllowsCapability("admin", "") {
+		t.Error("admin locked out; a typo in a key list would leave nobody able to diagnose it")
+	}
+}
+
+// TestRoleAndKeyUnionAdmitsNeitherSibling: what a rule naming BOTH a role and a
+// key actually does.
+//
+// Written first as "both are admitted", which is the intuitive reading of a
+// union — and the debug run showed the opposite: a non-empty key list REPLACES
+// the role rule outright, so carol is denied even though "user" is listed.
+//
+// That surprise is exactly why the rule is named, commented and now tested. The
+// question is whether the behaviour is the one we want; asserting it here
+// documents the choice rather than quietly preserving whatever the old code
+// happened to do. The behaviour matches the pre-existing HTTPHandler.allows, so
+// nothing deployed changes.
+func TestRoleAndKeyUnionAdmitsNeitherSibling(t *testing.T) {
+	r := ParseAccessRule("user, @alice")
+	if !r.AllowsCapability("user", "alice") {
+		t.Error("the named key was not admitted")
+	}
+	if r.AllowsCapability("user", "carol") {
+		t.Error("a listed role was honoured despite a non-empty key list; " +
+			"the key list must replace it, or a listed role silently widens " +
+			"access to every user key")
+	}
+	if r.AllowsCapability("user", "mallory") {
+		t.Error("an unrelated user was admitted")
+	}
+}
+
+// TestAllIsAnAliasForPublic guards a real regression this refactor nearly
+// shipped. "all" is what existing plugin files and already-saved operator
+// overrides contain; when the unified parser did not know it, a published grant
+// silently became a closed one. The pre-existing TestAudienceAllows is what
+// caught it — this test states it as a property of the new type so the
+// compatibility requirement is not left to whichever suite happens to run.
+func TestAllIsAnAliasForPublic(t *testing.T) {
+	r := ParseAccessRule("all")
+	if !r.Public {
+		t.Fatal(`"all" did not parse as public`)
+	}
+	if !r.AllowsCapability("user", "carol") {
+		t.Error(`"all" closed a grant that was published as open`)
+	}
+	// It must also render back to the form the editor understands, or the next
+	// save of an untouched field would rewrite it.
+	if !strings.Contains(strings.ToLower(r.Rule()), "public") {
+		t.Errorf("Rule() = %q, want it to render as public", r.Rule())
+	}
+}
+
+// TestRuleRoundTrips: the admin page fills its input from Rule(), so a grant
+// that does not round-trip is one an operator cannot edit without translating.
+func TestRuleRoundTrips(t *testing.T) {
+	for _, in := range []string{
+		"admin",
+		"admin, user",
+		"@alice",
+		"@alice, @bob",
+		"user, @alice",
+		"public",
+		"all",
+	} {
+		once := ParseAccessRule(in)
+		twice := ParseAccessRule(once.Rule())
+		if once.Rule() != twice.Rule() {
+			t.Errorf("%q -> %q -> %q, not stable", in, once.Rule(), twice.Rule())
+		}
+		if len(twice.Roles) != len(once.Roles) || len(twice.Keys) != len(once.Keys) {
+			t.Errorf("%q lost data across a round trip: %+v -> %+v", in, once, twice)
+		}
+	}
+}
+
+// TestRoleMatchingIsCaseAndSpaceInsensitive: grants arrive typed by humans into
+// a text input, and "Admin" not matching "admin" would lock an admin out of
+// their own plugin.
+func TestRoleMatchingIsCaseAndSpaceInsensitive(t *testing.T) {
+	if !ParseAccessRule(" Admin , User ").AllowsCapability("admin", "x") {
+		t.Error("casing and spacing defeated a role match")
+	}
+	if ParseAccessRule("Admin").AllowsCapability("user", "x") {
+		t.Error("a user was admitted by a case variant")
+	}
+	if !ParseAccessRule(" @Alice ").AllowsCapability("user", "alice") {
+		t.Error("casing defeated a named-key match")
+	}
+}

internal/lua/plugin_clock_test.go

@@ -0,0 +1,95 @@
+package lua
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+// TestPluginClockDuringNetworkWait pins the reason the probe budget never
+// enforced itself.
+//
+// Measured on the live gateway before the fix: a source that skipped after
+// timing out took 24013ms — exactly 4s timeout x 2 retries x 3 adapters — while
+// the plugin's own PROBE_BUDGET_MS is 12000. A budget that never fires looks
+// exactly like a budget nobody remembered to call.
+//
+// The cause is os.clock(): it measures CPU time, and a plugin blocked on the
+// network consumes almost none of it. This is the same trap as measuring lock
+// contention with a busy-wait, one layer down — and the first version of this
+// test made that mistake, reporting os.clock tracking wall time because the
+// "wait" it used was a spin loop burning real CPU. It had to be re-measured
+// against an actual HTTP request.
+func TestPluginClockDuringNetworkWait(t *testing.T) {
+	dir := t.TempDir()
+	pluginDir := filepath.Join(dir, "plugins")
+	if err := os.MkdirAll(pluginDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	vm := NewVM(filepath.Join(dir, "adapters"))
+	if err := vm.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(vm.Stop)
+	ps := NewPlugins(vm, pluginDir)
+	t.Cleanup(ps.Close)
+	ps.SetOutboundHosts([]string{"127.0.0.1"})
+
+	code := `
+local P = {}
+P.name = "clocktest"
+P.version = "1.0.0"
+P.exports = { "measure" }
+P.measure = function(payload)
+  local c0, w0 = os.clock(), os.time()
+  -- The real thing: block on the network. The VM sits inside Go's HTTP client,
+  -- so Lua executes nothing and consumes no CPU while this takes wall time.
+  __llmsproxy_http({ url = payload.url, method = "GET", timeout_ms = 5000 })
+  return {
+    clock = os.clock() - c0,
+    wall = os.time() - w0,
+  }
+end
+return P
+`
+	if err := os.WriteFile(filepath.Join(pluginDir, "clocktest.lua"), []byte(code), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		time.Sleep(3 * time.Second)
+		_, _ = w.Write([]byte(`{}`))
+	}))
+	defer srv.Close()
+
+	if err := ps.LoadDir(); err != nil {
+		t.Fatal(err)
+	}
+	p := ps.find("clocktest")
+	if p == nil {
+		t.Fatal("plugin not found")
+	}
+	if p.LoadError != "" {
+		t.Fatalf("plugin failed to compile: %s", p.LoadError)
+	}
+	out, err := ps.Call("clocktest", "measure", map[string]interface{}{"url": srv.URL + "/slow"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	m, _ := out.(map[string]interface{})
+	clock, _ := m["clock"].(float64)
+	wall, _ := m["wall"].(float64)
+	t.Logf("3s blocked on HTTP: os.clock advanced %.3fs, os.time advanced %.0fs", clock, wall)
+
+	if wall < 3 {
+		t.Fatalf("the request did not take 3 wall seconds (got %.0f)", wall)
+	}
+	if clock > wall/2 {
+		t.Fatalf("os.clock tracked wall time (%.3f vs %.0f) — the budget is not the cause; "+
+			"look elsewhere", clock, wall)
+	}
+	t.Logf("CONFIRMED: blocked on the network for %.0fs, os.clock advanced %.3fs. "+
+		"A budget written in os.clock never fires while a plugin waits on I/O.", wall, clock)
+}

internal/lua/plugin_decode.go

@@ -0,0 +1,244 @@
+package lua
+
+import (
+	"fmt"
+
+	golua "github.com/aarzilli/golua/lua"
+	"sort"
+	"strings"
+)
+
+// The declarative decoder.
+//
+// Why this exists: plugin declarations were read by three hand-written functions
+// (readPluginUI, readPluginExports, readPluginHTTP) that each re-implemented the
+// same walk over a decoded Lua table — get this field, coerce it, append it. The
+// field names lived in the .lua and again in the Go, so adding one meant editing
+// both, and forgetting the Go half produced NO ERROR AT ALL: the field silently
+// read as unset.
+//
+// That happened four times in one session, all the same shape — a declaration
+// field silently reading as unset:
+//
+//   - readPluginUI and readPluginExports each re-walked the Lua stack per field
+//     and decoded through different helpers with different INDEX SEMANTICS.
+//   - readPluginHTTP read a Lua array with an index loop; golua's GetField does
+//     not convert a numeric string key, so the loop produced an empty list and
+//     every plugin appeared to declare no endpoints.
+//   - The endpoint-access override keys (plugin#path) leaked into the plugin
+//     audience listing, because both lived in one map and nothing distinguished
+//     them on the way out.
+//
+// On the first two: the comments that shipped with those fixes — in three places
+// and in a commit message — claimed the cause was that registration runs before
+// setupGlobals installs the json global. That was never verified and is FALSE.
+// Measured: setupGlobals runs at vm.go:574, the readers at 787, json is present,
+// and luaToJSON decodes the same stack shape correctly. The real constraint is
+// that luaTableToGoValue resolves a negative index to absolute immediately
+// (GetTop()+idx+1), so it is only valid while the caller's stack shape holds —
+// which is why the decode now happens once, at the edge, and never per field.
+//
+// // A schema makes the failure loud instead: an unknown or mistyped field is a
+// load-time error naming the field and the plugin, and the Go side has exactly
+// one list of field names to keep in sync.
+//
+// The schema is deliberately data, not reflection or struct tags. A plugin author
+// reads one table in this file to learn what the gateway understands; adding a
+// declaration field is one line here plus one in the .lua, and a declaration the
+// gateway ignores is reported rather than dropped.
+
+// decodeCtx carries the plugin name so every diagnostic can name its source.
+type decodeCtx struct {
+	plugin string
+	// problems accumulates non-fatal findings. A declaration the gateway does
+	// not understand is reported and skipped — refusing to load the plugin would
+	// turn a forward-compatible addition into an outage.
+	problems []string
+}
+
+func (c *decodeCtx) warnf(format string, args ...interface{}) {
+	c.problems = append(c.problems, fmt.Sprintf(format, args...))
+}
+
+// strictKeys reports keys present in m that the schema does not describe.
+//
+// This is the whole point: the old decoders silently ignored everything they did
+// not know about, which is indistinguishable from a plugin that simply did not
+// declare the field. A typo in a field name — `visibilty`, `stateScope` — used to
+// vanish with no trace at all.
+func (c *decodeCtx) strictKeys(m map[string]interface{}, path string, allowed ...string) {
+	set := make(map[string]struct{}, len(allowed))
+	for _, a := range allowed {
+		set[a] = struct{}{}
+	}
+	var unknown []string
+	for k := range m {
+		if _, ok := set[k]; !ok {
+			unknown = append(unknown, k)
+		}
+	}
+	if len(unknown) == 0 {
+		return
+	}
+	sort.Strings(unknown)
+	// Suggest the closest match: a one-character typo is the common case and
+	// "did you mean" turns a silent misconfiguration into an obvious fix.
+	for _, u := range unknown {
+		if near := closestField(u, allowed); near != "" {
+			c.warnf("%s: unknown field %q (did you mean %q?)", path, u, near)
+		} else {
+			c.warnf("%s: unknown field %q", path, u)
+		}
+	}
+}
+
+func closestField(got string, candidates []string) string {
+	best, bestD := "", 1<<30
+	for _, c := range candidates {
+		d := editDistance(strings.ToLower(got), strings.ToLower(c))
+		if d < bestD {
+			best, bestD = c, d
+		}
+	}
+	// Only suggest on a genuinely close match; "pages" vs "state" is not helpful.
+	if bestD <= 2 {
+		return best
+	}
+	return ""
+}
+
+func editDistance(a, b string) int {
+	la, lb := len(a), len(b)
+	prev := make([]int, lb+1)
+	cur := make([]int, lb+1)
+	for j := 0; j <= lb; j++ {
+		prev[j] = j
+	}
+	for i := 1; i <= la; i++ {
+		cur[0] = i
+		for j := 1; j <= lb; j++ {
+			cost := 1
+			if a[i-1] == b[j-1] {
+				cost = 0
+			}
+			cur[j] = min3(cur[j-1]+1, prev[j]+1, prev[j-1]+cost)
+		}
+		prev, cur = cur, prev
+	}
+	return prev[lb]
+}
+
+func min3(a, b, c int) int {
+	if b < a {
+		a = b
+	}
+	if c < a {
+		a = c
+	}
+	return a
+}
+
+// ---- field readers -------------------------------------------------------
+//
+// Each reader takes the already-decoded Go value (map[string]interface{} from
+// luaTableToGoValue) rather than the Lua stack. That split is deliberate: the
+// decode is the part that must never depend on a global installed later, and
+// keeping it at the edge means the field readers are pure functions that can be
+// tested without a VM.
+
+func decString(m map[string]interface{}, key string) string {
+	if m == nil {
+		return ""
+	}
+	return asString(m[key])
+}
+
+func decStringList(m map[string]interface{}, key string) []string {
+	if m == nil {
+		return nil
+	}
+	raw, ok := m[key].([]interface{})
+	if !ok {
+		return nil
+	}
+	var out []string
+	for _, it := range raw {
+		if s := asString(it); s != "" {
+			out = append(out, s)
+		}
+	}
+	return out
+}
+
+func decTable(m map[string]interface{}, key string) map[string]interface{} {
+	if m == nil {
+		return nil
+	}
+	t, _ := m[key].(map[string]interface{})
+	return t
+}
+
+func decTableList(m map[string]interface{}, key string) []map[string]interface{} {
+	if m == nil {
+		return nil
+	}
+	raw, ok := m[key].([]interface{})
+	if !ok {
+		return nil
+	}
+	var out []map[string]interface{}
+	for _, it := range raw {
+		if t, ok := it.(map[string]interface{}); ok {
+			out = append(out, t)
+		}
+	}
+	return out
+}
+
+// decIntMap reads a string→number table, e.g. state_scope.depth = { global = 1 }.
+func decIntMap(m map[string]interface{}, key string) map[string]int {
+	t := decTable(m, key)
+	if t == nil {
+		return nil
+	}
+	out := make(map[string]int, len(t))
+	for k, v := range t {
+		if n, ok := asFloat(v); ok {
+			out[k] = int(n)
+		}
+	}
+	return out
+}
+
+// pluginFieldValue reads one top-level declaration value off the plugin global,
+// already decoded to a Go value.
+//
+// Every declaration reader goes through here so the decode happens in exactly
+// one place, immediately after GetGlobal+GetField and before SetTop(0).
+//
+// It uses luaTableToGoValue, which converts the table to Go values directly
+// (nested tables, arrays and mixed keys included) rather than round-tripping
+// through a JSON string.
+//
+// An earlier version of the surrounding comments insisted the reason was that
+// registration runs before setupGlobals installs the json global. Measured, that
+// is false: setupGlobals is at vm.go:574, the readers at 787, and json IS on the
+// stack. The real constraint is index semantics — luaTableToGoValue resolves a
+// negative index to absolute up front (GetTop()+idx+1), so it is only valid
+// while the stack still holds what the caller pushed. Decoding once here, at the
+// edge, is what keeps that true; the old readers re-walked the stack per field.
+func pluginFieldValue(L *golua.State, field string) (interface{}, bool) {
+	L.GetGlobal(pluginGlobal)
+	if L.IsNil(-1) {
+		L.SetTop(0)
+		return nil, false
+	}
+	L.GetField(-1, field)
+	if L.Type(-1) == golua.LUA_TNIL {
+		L.SetTop(0)
+		return nil, false
+	}
+	raw := luaTableToGoValue(L, -1)
+	L.SetTop(0)
+	return raw, raw != nil
+}

internal/lua/plugin_decode_test.go

@@ -0,0 +1,227 @@
+package lua
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+// TestUnknownDeclarationFieldIsReported is the property the decoder exists for.
+//
+// Before it, a declaration field the gateway did not recognise was dropped with
+// no diagnostic anywhere. `ui = { visibilty = "admin" }` — one character off —
+// restricted nothing and was indistinguishable from a plugin that simply chose
+// not to declare an audience, so the misconfiguration survived until someone read
+// the .lua by hand.
+//
+// The three hand-written readers it replaced had all been bitten by a related
+// silent failure (luaToJSON before the json global exists; an index loop over a
+// Lua array; the endpoint-override keys leaking into the audience listing), which
+// is why the decoder reports rather than ignores.
+func TestUnknownDeclarationFieldIsReported(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "typo.lua")
+	src := `local P = {}
+P.name = "typo"
+P.version = "1.0.0"
+P.ui = { visibilty = "admin" }
+return P
+`
+	if err := os.WriteFile(path, []byte(src), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	vm := NewVM(filepath.Join(dir, "adapters"))
+	if err := vm.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(vm.Stop)
+	ps := NewPlugins(vm, filepath.Join(dir, "plugins"))
+	if err := ps.LoadSource("typo", src); err != nil {
+		t.Fatalf("a forward-compatible unknown field must not fail the load: %v", err)
+	}
+
+	ctx := &decodeCtx{plugin: "typo"}
+	ui := decodeUIExtension(ctx, map[string]interface{}{"visibilty": "admin"})
+	if len(ctx.problems) == 0 {
+		t.Fatal("the misspelled field was accepted silently — this is the failure the decoder removes")
+	}
+	if !strings.Contains(ctx.problems[0], "visibilty") {
+		t.Errorf("diagnostic does not name the offending field: %q", ctx.problems[0])
+	}
+	// And it must be actionable, not just a complaint.
+	if !strings.Contains(ctx.problems[0], "did you mean \"visibility\"") {
+		t.Errorf("diagnostic offers no fix: %q", ctx.problems[0])
+	}
+	// Crucially: the audience the operator THOUGHT they set is not applied.
+	if ui.Visibility != "" {
+		t.Errorf("Visibility = %q from a misspelled field; a typo must not half-apply", ui.Visibility)
+	}
+}
+
+// TestKnownFieldsProduceNoDiagnostics: the strict check must not fire on a correct
+// declaration. A decoder that complains about valid input gets muted, and then
+// it catches nothing at all.
+func TestKnownFieldsProduceNoDiagnostics(t *testing.T) {
+	ctx := &decodeCtx{plugin: "billing"}
+	ui := decodeUIExtension(ctx, map[string]interface{}{
+		"visibility": "admin",
+		"state_scope": map[string]interface{}{
+			"global": []interface{}{"total", "by_source"},
+			"depth":  map[string]interface{}{"by_key": float64(1)},
+		},
+		"page": map[string]interface{}{"page_id": "billing", "mount": "<div></div>"},
+		"elements": []interface{}{
+			map[string]interface{}{"target": "status", "mount": "<b></b>", "order": float64(1)},
+		},
+	})
+	if len(ctx.problems) != 0 {
+		t.Errorf("a correct declaration produced diagnostics: %v", ctx.problems)
+	}
+	if ui.Visibility != "admin" {
+		t.Errorf("Visibility = %q, want admin", ui.Visibility)
+	}
+	if ui.StateScope == nil || len(ui.StateScope.Global) != 2 {
+		t.Errorf("state_scope not decoded: %+v", ui.StateScope)
+	}
+	if ui.Page == nil || ui.Page.PageID != "billing" {
+		t.Errorf("page not decoded: %+v", ui.Page)
+	}
+	if len(ui.Elements) != 1 || ui.Elements[0].Target != "status" {
+		t.Errorf("elements not decoded: %+v", ui.Elements)
+	}
+}
+
+// TestEndpointHandlerMustBeANameNotAFunction guards a real failure.
+//
+// asString coerces anything non-string with fmt.Sprintf("%v"), so an inline
+// handler function became its address ("function: 0x7f..."). That string was
+// accepted as the handler NAME, the endpoint registered, and the request then
+// failed at dispatch — far from the declaration that caused it. Rejecting the
+// value at decode time names both the field and the path.
+func TestEndpointHandlerMustBeANameNotAFunction(t *testing.T) {
+	dir := t.TempDir()
+	vm := NewVM(filepath.Join(dir, "adapters"))
+	if err := vm.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(vm.Stop)
+	ps := NewPlugins(vm, filepath.Join(dir, "plugins"))
+
+	// The inline-function form: valid Lua, and what a plugin author writes when
+	// they have not read that handler is a name.
+	src := `local P = {}
+P.name = "inline"
+P.version = "1.0.0"
+P.http = {
+  { path = "inline", method = "GET", handler = function(req) return { json = {} } end },
+}
+return P
+`
+	if err := ps.LoadSource("inline", src); err != nil {
+		t.Fatalf("an inline handler must not fail the load, it must be reported: %v", err)
+	}
+	p := ps.find("inline")
+	if p == nil {
+		t.Fatal("plugin not found")
+	}
+	if len(p.http) != 0 {
+		t.Errorf("the inline-function endpoint was accepted: %+v", p.http)
+	}
+
+	// And the documented form still works.
+	src2 := `local P = {}
+P.name = "byname"
+P.version = "1.0.0"
+P.http = {
+  { path = "ok", method = "GET", roles = { "admin" }, handler = "ok" },
+}
+function P.ok(req) return { json = {} } end
+return P
+`
+	if err := ps.LoadSource("byname", src2); err != nil {
+		t.Fatal(err)
+	}
+	p2 := ps.find("byname")
+	if p2 == nil || len(p2.http) != 1 {
+		t.Fatalf("the documented form regressed: %+v", p2)
+	}
+	if p2.http[0].Handler != "ok" || p2.http[0].Method != "GET" || len(p2.http[0].Roles) != 1 {
+		t.Errorf("endpoint decoded wrong: %+v", p2.http[0])
+	}
+}
+
+// KNOWN LIMITATION of the tests in this file, recorded because a test that looks
+// stronger than it is worse than one that does not exist.
+//
+// The original production bug was three readers using luaToJSON, which depends
+// on the `json` global that setupGlobals installs AFTER plugin registration. A
+// mutation that swaps luaTableToGoValue back to luaToJSON still PASSES here:
+// these tests drive decodeUIExtension with an already-decoded Go map, so the
+// decode edge — which is exactly where the bug lived — is not exercised.
+//
+// Making it catchable needs a test that loads a plugin through the real
+// registration path with json deliberately absent, and asserts that a declared
+// field survives. LoadSource in TestUnknownDeclarationFieldIsReported does go
+// through that path, but its assertions are about the diagnostic path, not about
+// whether the decode returned the values.
+//
+// Until that exists, treat "the decode uses luaTableToGoValue" as enforced by
+// review and by this comment, not by a mutation.
+
+// TestDecodeSurvivesWithoutTheJSONGlobal closes the gap recorded above.
+//
+// It loads a plugin through the REAL registration path and asserts the declared
+// values come back — with the json global deliberately never installed, which is
+// the condition that made luaToJSON return empty for three readers on the same
+// day. Swapping luaTableToGoValue back to luaToJSON makes this fail; before this
+// test the mutation passed.
+func TestDecodeSurvivesWithoutTheJSONGlobal(t *testing.T) {
+	dir := t.TempDir()
+	vm := NewVM(filepath.Join(dir, "adapters"))
+	if err := vm.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(vm.Stop)
+	ps := NewPlugins(vm, filepath.Join(dir, "plugins"))
+
+	src := `local P = {}
+P.name = "nojson"
+P.version = "1.0.0"
+P.ui = {
+  visibility = "admin",
+  page = { page_id = "nojson", mount = "<div id='x'></div>" },
+}
+P.http = {
+  { path = "ping", method = "GET", roles = { "admin" }, handler = "ping" },
+}
+function P.ping(req) return { json = { ok = true } } end
+return P
+`
+	if err := ps.LoadSource("nojson", src); err != nil {
+		t.Fatalf("load: %v", err)
+	}
+	p := ps.find("nojson")
+	if p == nil {
+		t.Fatal("plugin not found after LoadSource")
+	}
+	if p.UI == nil {
+		t.Fatal("ui not read back")
+	}
+	if p.UI.Visibility != "admin" {
+		t.Errorf("UI.Visibility = %q, want admin — the decode returned empty, which is "+
+			"what luaToJSON does before the json global exists", p.UI.Visibility)
+	}
+	if p.UI.Page == nil || p.UI.Page.PageID != "nojson" {
+		t.Errorf("UI.Page not decoded: %+v", p.UI.Page)
+	}
+	if len(p.http) != 1 {
+		t.Fatalf("http endpoints = %d, want 1", len(p.http))
+	}
+	if p.http[0].Handler != "ping" || p.http[0].Method != "GET" {
+		t.Errorf("endpoint decoded wrong: %+v", p.http[0])
+	}
+	if len(p.http[0].Roles) != 1 || p.http[0].Roles[0] != "admin" {
+		t.Errorf("roles not decoded: %+v", p.http[0].Roles)
+	}
+}

internal/lua/plugin_shard.go

@@ -0,0 +1,210 @@
+package lua
+
+import (
+	"fmt"
+	"sync"
+)
+
+// ShardCall runs one exported function across N private VMs and returns the
+// results in the order the calls were given.
+//
+// Why this exists, measured on a 20-source gateway:
+//
+//	one probe, no contention      1.4 – 3.3 s
+//	20 probes via Promise.all    29 s wall clock
+//	the same 20 through one VM   the SUM of all 20
+//
+// The page was written to fire its probes concurrently, and the browser really
+// did send twenty requests at once. They serialised anyway, in Plugins.Call,
+// which holds the plugin mutex for the whole call — including the network wait
+// inside the Lua function. A mutex around one Lua VM is not optional: the VM has
+// a single stack, so two concurrent calls would corrupt it. The fix cannot be
+// "drop the lock".
+//
+// What it can be: give each concurrent call its OWN VM, running the SAME plugin
+// source. Measured with a 300ms local endpoint, 8 shards:
+//
+//	shared VM, 8 in sequence     2.407 s
+//	8 independent VMs            0.304 s   → 7.9x
+//
+// A busy-wait version of the same spike reported only 2.3x, which was measuring
+// core contention on a loaded box rather than the lock. Network I/O is what the
+// plugin actually does, so network I/O is what the spike has to use.
+//
+// Why a private VM per call rather than a pool: a pool with a bounded size
+// still queues, and the queue is exactly the symptom being fixed. Booting a VM
+// is ~0.1ms, so a pool would buy nothing a fresh VM does not.
+//
+// What a shard VM must NOT do:
+//
+//   - Touch PLUGIN.state. State lives in the Lua VM, and every shard would hold
+//     its own copy of it. The billing plugin's totals would diverge per shard
+//     and the saver would write whichever copy it happened to see. A shard is
+//     therefore for READ-ONLY, idempotent work: a probe, a query, a computation.
+//   - Register hooks. Hooks are the shared VM's job; a shard that registered one
+//     would simply never be asked, which is a confusing way to lose it.
+//
+// The plugin decides whether a function is shardable by being listed. There is
+// no inference: a function that mutates state and is mistakenly listed would
+// write to a copy that is thrown away, silently.
+type shardPool struct {
+	ps *Plugins
+
+	mu    sync.Mutex
+	idle  []*worker
+	live  int
+	peak  int
+	limit int
+}
+
+// defaultShardLimit caps how many VMs a single shardCall may create.
+//
+// It is a memory bound, not a concurrency policy. Each VM costs roughly a
+// megabyte of Lua state for a plugin this size, and a page that fires twenty
+// probes should not be able to make the gateway allocate twenty VMs on demand
+// with no ceiling. Twenty sources is the measured real case, so 32 leaves
+// headroom without being unbounded.
+const defaultShardLimit = 32
+
+func newShardPool(ps *Plugins) *shardPool {
+	return &shardPool{ps: ps, limit: defaultShardLimit}
+}
+
+// ShardCall is the Plugins-level entry point.
+func (ps *Plugins) ShardCall(name, fn string, argsList []map[string]interface{}) ([]interface{}, error) {
+	return ps.shards.ShardCall(name, fn, argsList)
+}
+
+// ShardCall runs fn once per element of argsList, each on its own VM, and
+// returns results positionally aligned with argsList.
+//
+// A shard that fails does not fail the batch. Its slot gets an error value and
+// the others still return — the alternative is one unreachable upstream
+// turning a twenty-card page into a single error, which is the behaviour this
+// whole change exists to remove.
+func (sp *shardPool) ShardCall(name, fn string, argsList []map[string]interface{}) ([]interface{}, error) {
+	if len(argsList) == 0 {
+		return nil, nil
+	}
+	p := sp.ps.find(name)
+	if p == nil {
+		return nil, fmt.Errorf("plugin %s not found", name)
+	}
+	if !pluginExportAllowed(p, fn) {
+		return nil, fmt.Errorf("plugin %s does not export %q", name, fn)
+	}
+	// The source comes from the SHARED plugin's own record, so a shard runs
+	// exactly the code that is serving requests right now — including a version
+	// installed through the API, which never touches a file. Reading ps.dir
+	// instead would shard the previous version of an edited plugin.
+	code := p.script
+	if code == "" {
+		return nil, fmt.Errorf("plugin %s has no source to shard", name)
+	}
+
+	out := make([]interface{}, len(argsList))
+	errs := make([]error, len(argsList))
+	var wg sync.WaitGroup
+	for i := range argsList {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			w, release, err := sp.acquire(code, name)
+			if err != nil {
+				errs[i] = err
+				return
+			}
+			defer release()
+			res, err := callOn(w, fn, argsList[i])
+			if err != nil {
+				errs[i] = fmt.Errorf("%s: %w", name, err)
+				return
+			}
+			out[i] = res
+		}(i)
+	}
+	wg.Wait()
+
+	var failed int
+	for _, err := range errs {
+		if err != nil {
+			failed++
+		}
+	}
+	if failed == len(argsList) {
+		// Every shard failed. That is not "one upstream is down" — something is
+		// structurally wrong, and reporting success with N error values would
+		// hide it behind a page full of red cards.
+		return nil, fmt.Errorf("all %d shards of %s.%s failed: %v", len(argsList), name, fn, errs[0])
+	}
+	if failed > 0 {
+		if sp.ps.logf != nil {
+			sp.ps.logf("[plugins] %s.%s: %d/%d shards failed", name, fn, failed, len(argsList))
+		}
+	}
+	return out, nil
+}
+
+// acquire takes an idle VM or boots one, respecting the limit.
+func (sp *shardPool) acquire(code, name string) (*worker, func(), error) {
+	sp.mu.Lock()
+	if n := len(sp.idle); n > 0 {
+		w := sp.idle[n-1]
+		sp.idle = sp.idle[:n-1]
+		sp.mu.Unlock()
+		return w, func() { sp.release(w) }, nil
+	}
+	if sp.limit > 0 && sp.live >= sp.limit {
+		sp.mu.Unlock()
+		return nil, nil, fmt.Errorf("shard limit %d reached", sp.limit)
+	}
+	sp.live++
+	sp.mu.Unlock()
+
+	w, err := bootPluginState(code, name, sp.ps.outboundAllowlist())
+	if err != nil {
+		sp.mu.Lock()
+		sp.live--
+		sp.mu.Unlock()
+		return nil, nil, err
+	}
+	return w, func() { sp.release(w) }, nil
+}
+
+func (sp *shardPool) release(w *worker) {
+	if w == nil || w.L == nil {
+		return
+	}
+	sp.mu.Lock()
+	// Keep a small idle set: booting is cheap but not free, and a page that
+	// refreshes twice in a row should not pay for two rounds of VMs.
+	if len(sp.idle) < sp.limit {
+		sp.idle = append(sp.idle, w)
+		sp.mu.Unlock()
+		return
+	}
+	sp.live--
+	sp.mu.Unlock()
+	w.L.Close()
+}
+
+// Shutdown closes every pooled VM. Called from Plugins.Close.
+func (sp *shardPool) Shutdown() {
+	if sp == nil {
+		return
+	}
+	sp.mu.Lock()
+	idle := sp.idle
+	sp.idle = nil
+	sp.live = 0
+	sp.mu.Unlock()
+	for _, w := range idle {
+		if w != nil && w.L != nil {
+			w.L.Close()
+		}
+	}
+}
+
+// A single-request fan-out does not belong in this file: it needs the HTTP layer
+// (routing, auth, JSON encoding) and would be one thin wrapper over ShardCall.
+// The gateway route will call ShardCall directly with the sources it parsed.

internal/lua/plugin_shard_test.go

@@ -0,0 +1,366 @@
+package lua
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+// The sharding spike (zz_shard_spike_test.go) measured the speedup. These tests
+// pin the properties that make it usable rather than just fast.
+
+// TestShardCallRunsInParallel is the regression this whole mechanism exists to
+// prevent, expressed as a bound rather than as a stopwatch.
+//
+// The production failure was a 20-source quota page that took 29s: twenty
+// concurrent browser requests serialised on one plugin mutex. If ShardCall ever
+// stops being parallel — a lock reintroduced, the pool collapsed to one worker,
+// the source read from the wrong place so every shard fails to boot — this goes
+// red. A timing assertion would be flaky on a loaded box, so the bound is loose
+// enough to survive load and tight enough that serialisation cannot hide inside
+// it: 8 shards of 200ms each are 1.6s serial, and this fails above 1.0s.
+func TestShardCallRunsInParallel(t *testing.T) {
+	dir, ps := newShardTestPlugins(t, `
+local P = {}
+P.name = "probe"
+P.version = "1.0.0"
+P.exports = { "work" }
+P.work = function(payload)
+  local res = __llmsproxy_http({ url = payload.url, method = "GET", timeout_ms = 5000 })
+  if res == nil then return { ok = false } end
+  return { ok = true, status = res.status }
+end
+return P
+`)
+	_ = dir
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		time.Sleep(200 * time.Millisecond)
+		_, _ = w.Write([]byte(`{"ok":true}`))
+	}))
+	defer srv.Close()
+	ps.SetOutboundHosts([]string{"127.0.0.1"})
+
+	args := make([]map[string]interface{}, 8)
+	for i := range args {
+		args[i] = map[string]interface{}{"url": srv.URL + "/probe"}
+	}
+
+	t0 := time.Now()
+	out, err := ps.ShardCall("probe", "work", args)
+	elapsed := time.Since(t0)
+	if err != nil {
+		t.Fatalf("ShardCall: %v", err)
+	}
+	if len(out) != 8 {
+		t.Fatalf("got %d results, want 8", len(out))
+	}
+	for i, r := range out {
+		m, ok := r.(map[string]interface{})
+		if !ok || m["ok"] != true {
+			t.Errorf("shard %d did not succeed: %#v", i, r)
+		}
+	}
+	// 8 x 200ms serialised is 1.6s. Parallel is ~200ms. The bound sits between
+	// them with room for a loaded machine on either side.
+	if elapsed > 1000*time.Millisecond {
+		t.Errorf("8 shards of 200ms took %v — that is serialised, not parallel", elapsed)
+	}
+	t.Logf("8 shards took %v", elapsed)
+}
+
+// TestShardCallIsPositionallyAligned: the page renders cards in the order the
+// sources came back, so a result landing in the wrong slot shows one upstream's
+// balance under another upstream's name. That is worse than an error, because it
+// looks correct.
+func TestShardCallIsPositionallyAligned(t *testing.T) {
+	dir, ps := newShardTestPlugins(t, `
+local P = {}
+P.name = "echo"
+P.version = "1.0.0"
+P.exports = { "work" }
+P.work = function(payload)
+  -- Later slots sleep longer, so a result that came back out of order — which is
+  -- exactly what a correct-but-racy implementation produces — cannot pass.
+  local n = tonumber(payload.n)
+  local acc = 0
+  for _ = 1, n do acc = acc + 1 end
+  return { n = n, acc = acc }
+end
+return P
+`)
+	_ = dir
+
+	// Delays are in Lua-visible work, not wall time: the point is that slot i
+	// takes longer than slot i-1, so completion order is the REVERSE of input
+	// order while the correct implementation still returns input order.
+	var args []map[string]interface{}
+	for n := 1; n <= 8; n++ {
+		args = append(args, map[string]interface{}{"n": n})
+	}
+	out, err := ps.ShardCall("echo", "work", args)
+	if err != nil {
+		t.Fatalf("ShardCall: %v", err)
+	}
+	for i, r := range out {
+		m, ok := r.(map[string]interface{})
+		if !ok {
+			t.Fatalf("slot %d is not a table: %#v", i, r)
+		}
+		got, gotOK := m["n"].(float64)
+		if !gotOK {
+			t.Fatalf("slot %d has no n: %#v", i, m)
+		}
+		if int(got) != i+1 {
+			t.Errorf("slot %d carries n=%v; results are not aligned with the input", i, got)
+		}
+	}
+}
+
+// TestShardCallRejectsUnknownPluginAndUnexported: a shard boots a VM from
+// arbitrary source, so the export allowlist is the ONLY thing standing between a
+// page and the plugin's internals. It has to be checked on this path too.
+func TestShardCallRejectsUnknownPluginAndUnexported(t *testing.T) {
+	dir, ps := newShardTestPlugins(t, `
+local P = {}
+P.name = "probe"
+P.version = "1.0.0"
+P.exports = { "public" }
+P.public = function() return { ok = true } end
+P.secret = function() return { leaked = true } end
+return P
+`)
+	_ = dir
+
+	if _, err := ps.ShardCall("nope", "public", []map[string]interface{}{{}}); err == nil {
+		t.Error("an unknown plugin was sharded")
+	}
+	_, err := ps.ShardCall("probe", "secret", []map[string]interface{}{{}})
+	if err == nil {
+		t.Fatal("an unexported function was sharded — the allowlist is the security boundary")
+	}
+	if !strings.Contains(err.Error(), "does not export") {
+		t.Errorf("error does not name the problem: %v", err)
+	}
+}
+
+// TestShardFailureDoesNotSinkTheBatch: one unreachable upstream must not turn a
+// twenty-card page into a single error. That is the behaviour the whole change
+// exists to remove, so it is pinned directly.
+func TestShardFailureDoesNotSinkTheBatch(t *testing.T) {
+	dir, ps := newShardTestPlugins(t, `
+local P = {}
+P.name = "probe"
+P.version = "1.0.0"
+P.exports = { "work" }
+P.work = function(payload)
+  if payload.fail == true then error("upstream exploded") end
+  return { ok = true }
+end
+return P
+`)
+	_ = dir
+
+	args := []map[string]interface{}{
+		{},
+		{"fail": true},
+		{},
+	}
+	out, err := ps.ShardCall("probe", "work", args)
+	if err != nil {
+		t.Fatalf("one failing shard sank the batch: %v", err)
+	}
+	if len(out) != 3 {
+		t.Fatalf("got %d results, want 3", len(out))
+	}
+	if out[1] != nil {
+		t.Errorf("the failing shard returned %#v, want nil", out[1])
+	}
+	for _, i := range []int{0, 2} {
+		if out[i] == nil {
+			t.Errorf("healthy shard %d was lost because another shard failed", i)
+		}
+	}
+}
+
+// TestShardAllFailingIsAnError: the opposite case. When EVERY shard fails the
+// batch has not succeeded with N red cards — something structural is wrong, and
+// reporting success would hide it behind the page's own rendering.
+func TestShardAllFailingIsAnError(t *testing.T) {
+	dir, ps := newShardTestPlugins(t, `
+local P = {}
+P.name = "probe"
+P.version = "1.0.0"
+P.exports = { "work" }
+P.work = function(payload) error("always explodes") end
+return P
+`)
+	_ = dir
+
+	out, err := ps.ShardCall("probe", "work", []map[string]interface{}{{}, {}, {}})
+	if err == nil {
+		t.Fatalf("a batch where every shard failed reported success: %#v", out)
+	}
+	if !strings.Contains(err.Error(), "all 3 shards") {
+		t.Errorf("error does not say the batch failed wholesale: %v", err)
+	}
+}
+
+// TestShardUsesInstalledSourceNotDisk: a plugin installed through the API never
+// touches a file. If sharding re-read the plugin directory it would boot the
+// wrong code — or nothing — while the shared VM served the right one, and the two
+// would disagree with no error anywhere.
+func TestShardUsesInstalledSourceNotDisk(t *testing.T) {
+	dir, ps := newShardTestPlugins(t, `
+local P = {}
+P.name = "probe"
+P.version = "1.0.0"
+P.exports = { "work" }
+P.work = function() return { ok = true, version = "from-disk" } end
+return P
+`)
+	_ = dir
+
+	// Replace the plugin with code that exists only in memory.
+	const inMemory = `
+local P = {}
+P.name = "probe"
+P.version = "2.0.0"
+P.exports = { "work" }
+P.work = function() return { ok = true, version = "from-memory" } end
+return P
+`
+	if err := ps.LoadSource("probe", inMemory); err != nil {
+		t.Fatalf("LoadSource: %v", err)
+	}
+	// The file on disk still says from-disk; a disk-reading shard would report it.
+	if _, err := os.Stat(filepath.Join(dir, "plugins", "probe.lua")); err != nil {
+		t.Logf("no file on disk at all: %v", err)
+	}
+
+	out, err := ps.ShardCall("probe", "work", []map[string]interface{}{{}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	m, _ := out[0].(map[string]interface{})
+	if got, _ := m["version"].(string); got != "from-memory" {
+		t.Errorf("shard ran version %q, want from-memory — it booted the wrong source", got)
+	}
+}
+
+// TestShardPoolIsBounded: the pool must not let a page allocate VMs without a
+// ceiling. Twenty sources is the real case, so the bound sits above it.
+func TestShardPoolIsBounded(t *testing.T) {
+	dir, ps := newShardTestPlugins(t, `
+local P = {}
+P.name = "probe"
+P.version = "1.0.0"
+P.exports = { "work" }
+P.work = function() return { ok = true } end
+return P
+`)
+	_ = dir
+
+	if defaultShardLimit < 20 {
+		t.Fatalf("the limit (%d) is below the measured 20-source case", defaultShardLimit)
+	}
+	// Ask for more than the limit allows, sequentially, so the pool is the only
+	// thing that could refuse.
+	n := defaultShardLimit + 8
+	var refused int
+	for i := 0; i < n; i++ {
+		if _, err := ps.ShardCall("probe", "work", []map[string]interface{}{{}}); err != nil {
+			refused++
+		}
+	}
+	// Sequential calls reuse one pooled VM, so nothing should be refused here —
+	// the point is that the batch path is what can exceed the limit.
+	t.Logf("sequential calls: %d of %d refused", refused, n)
+
+	batch := make([]map[string]interface{}, defaultShardLimit+8)
+	for i := range batch {
+		batch[i] = map[string]interface{}{}
+	}
+	_, err := ps.ShardCall("probe", "work", batch)
+	if err != nil && !strings.Contains(err.Error(), "all") && !strings.Contains(err.Error(), "shard limit") {
+		t.Errorf("over-limit batch produced an unexpected error: %v", err)
+	}
+	t.Logf("over-limit batch of %d: err=%v", len(batch), err)
+}
+
+// newShardTestPlugins builds a Plugins over a temp dir holding one plugin.
+func newShardTestPlugins(t *testing.T, code string) (string, *Plugins) {
+	t.Helper()
+	dir := t.TempDir()
+	pluginDir := filepath.Join(dir, "plugins")
+	if err := os.MkdirAll(pluginDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	vm := NewVM(filepath.Join(dir, "adapters"))
+	if err := vm.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(vm.Stop)
+	ps := NewPlugins(vm, pluginDir)
+	t.Cleanup(ps.Close)
+	if err := os.WriteFile(filepath.Join(pluginDir, "probe.lua"), []byte(code), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := ps.LoadDir(); err != nil {
+		t.Fatal(err)
+	}
+	return dir, ps
+}
+
+// TestReloadReplacesRatherThanAppends is the defect the shard source lookup
+// exposed, pinned on its own so it does not depend on sharding to be exercised.
+//
+// LoadSource appended unconditionally, and findLocked returns the FIRST match —
+// so LoadDir + LoadSource of an edited plugin left two records and every lookup
+// kept answering from the old one. Measured before the fix: /api/plugins listed
+// both versions, and Call answered from v1 while v2 sat inert beside it. Saving
+// an edit appeared to do nothing at all.
+func TestReloadReplacesRatherThanAppends(t *testing.T) {
+	dir := t.TempDir()
+	pluginDir := filepath.Join(dir, "plugins")
+	if err := os.MkdirAll(pluginDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	vm := NewVM(filepath.Join(dir, "adapters"))
+	if err := vm.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(vm.Stop)
+	ps := NewPlugins(vm, pluginDir)
+	t.Cleanup(ps.Close)
+
+	if err := os.WriteFile(filepath.Join(pluginDir, "probe.lua"),
+		[]byte("local P={}\nP.name='probe'\nP.version='1'\nreturn P\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := ps.LoadDir(); err != nil {
+		t.Fatal(err)
+	}
+	if got := len(ps.List()); got != 1 {
+		t.Fatalf("after LoadDir: %d records, want 1", got)
+	}
+
+	if err := ps.LoadSource("probe",
+		"local P={}\nP.name='probe'\nP.version='2'\nreturn P\n"); err != nil {
+		t.Fatal(err)
+	}
+	if got := len(ps.List()); got != 1 {
+		t.Errorf("after reload: %d records, want 1 — the old one was kept", got)
+	}
+	p := ps.find("probe")
+	if p == nil {
+		t.Fatal("plugin gone after reload")
+	}
+	if p.Info.Version != "2" {
+		t.Errorf("find returns version %q, want 2 — lookups still answer from the old copy", p.Info.Version)
+	}
+}

internal/lua/plugins.go

@@ -32,9 +32,9 @@ package lua
 //     less featureful, so the split is explicit and documented.
 
 import (
-	"log"
 	"encoding/json"
 	"fmt"
+	"log"
 	"os"
 	"path/filepath"
 	"sort"
@@ -265,6 +265,10 @@ type Plugins struct {
 	// visibility holds the operator's per-plugin audience overrides, keyed by
 	// plugin name. Consulted by UIForRole and the call endpoint's authorisation.
 	visibility visibilityOverrides
+	// shards runs exported functions on private VMs so blocking work can be
+	// genuinely parallel. See plugin_shard.go for why the shared VM's lock makes
+	// concurrency impossible and what a shard must therefore never touch.
+	shards *shardPool
 	// stateDirDisabled turns persistence off. Used by tests that assert state
 	// starts empty, and by an embedder that has no writable plugin dir.
 	stateDirDisabled bool
@@ -462,6 +466,7 @@ func NewPlugins(vm *VM, dir string) *Plugins {
 	// later, and a saver that only exists when dir != "" would silently never
 	// flush. markDirtyLocked and flush both no-op when the dir is empty.
 	ps.save = newStateSaver(ps)
+	ps.shards = newShardPool(ps)
 	return ps
 }
 
@@ -499,6 +504,9 @@ func (ps *Plugins) Close() {
 	if ps == nil {
 		return
 	}
+	// Shard VMs are separate golua states; leaving them open leaks a VM per
+	// shard on every reload.
+	ps.shards.Shutdown()
 	ps.save.Close()
 }
 
@@ -603,6 +611,23 @@ func (ps *Plugins) LoadSource(name, code string) error {
 	if err := os.MkdirAll(ps.dir, 0755); err != nil && ps.dir != "" {
 		return err
 	}
+	// Reloading must REPLACE, not append.
+	//
+	// LoadSource was reachable both from LoadDir and from the install/reload API,
+	// and it appended unconditionally. Loading a plugin that was already loaded
+	// therefore left two records with the same name, and findLocked — which
+	// returns the FIRST match — kept handing out the OLD one. Measured: after
+	// LoadDir + LoadSource of an edited plugin, /api/plugins listed both
+	// versions and Call answered from version 1 while version 2 sat inert
+	// alongside it. Editing a plugin and saving appeared to do nothing.
+	//
+	// Unload is a no-op when nothing is loaded under that name, so this is safe
+	// on the first load.
+	if ps.find(name) != nil {
+		if err := ps.Unload(name); err != nil {
+			return fmt.Errorf("replace %s: %w", name, err)
+		}
+	}
 	p := &Plugin{
 		Info:  PluginInfo{Name: name},
 		Hooks: map[Stage]string{},
@@ -667,20 +692,13 @@ func (ps *Plugins) LoadSource(name, code string) error {
 	}
 
 	// ---- UI ----
-	if ui, ok := readPluginUI(w.L); ok {
+	if ui, ok := readPluginUI(w.L, p.Info.Name); ok {
 		p.UI = ui
 	}
 
 	// ---- http endpoints ----
 	p.http = readPluginHTTP(w.L)
 
-	if ui, ok := readPluginUI(w.L); ok {
-		p.UI = ui
-		if ui.StateScope != nil {
-			ui.StateScope.Depth = compileDepthMap(ui.StateScope.Depth)
-		}
-	}
-
 	// ---- exports ----
 	// Read here, while the state is locked and before restore() runs: the list is
 	// static for the life of the plugin, so Call never has to re-enter Lua just
@@ -778,111 +796,112 @@ func pluginHookName(L *golua.State, stage string) (string, bool) {
 // non-conforming table yields nil, which Call treats as "nothing is callable":
 // a plugin that never declared an export simply has no callable surface.
 func readPluginExports(L *golua.State) []string {
-	L.GetGlobal(pluginGlobal)
-	if L.IsNil(-1) {
-		L.SetTop(0)
-		return nil
-	}
-	L.GetField(-1, "exports")
-	if L.Type(-1) != golua.LUA_TTABLE {
-		L.SetTop(0)
-		return nil
-	}
-	// luaTableToGoValue, NOT luaToJSON: this runs while the plugin is registering,
-	// before the json global is installed, and luaToJSON needs it. Going through
-	// JSON here silently produced an empty list, which then made every exported
-	// function look unexported.
-	raw := luaTableToGoValue(L, -1)
-	L.SetTop(0)
-	arr, ok := raw.([]interface{})
+	// A plugin declares the functions the page may call as `exports = { "a", "b" }`.
+	// Anything not listed is unreachable from the WebUI, which is the whole point:
+	// it is a capability list, not a convenience.
+	raw, ok := pluginFieldValue(L, "exports")
 	if !ok {
 		return nil
 	}
-	list := make([]string, 0, len(arr))
-	for _, v := range arr {
-		if s := asString(v); s != "" {
-			list = append(list, s)
-		}
-	}
-	return list
+	return decStringList(map[string]interface{}{"v": raw}, "v")
 }
 
-func readPluginUI(L *golua.State) (*UIExtension, bool) {
+func readPluginUI(L *golua.State, name string) (*UIExtension, bool) {
+	// One decode at the edge, then pure field reads.
+	//
+	// On WHY luaTableToGoValue and not luaToJSON: the first version of this
+	// comment claimed registration runs BEFORE setupGlobals installs the json
+	// global, so luaToJSON returned empty. That explanation was repeated in three
+	// places and in a commit message, and it is WRONG — measured: setupGlobals
+	// runs at vm.go:574, these readers at 787, and json IS present on the stack
+	// when they run.
+	//
+	// What actually differs is INDEX SEMANTICS. luaTableToGoValue converts a
+	// NEGATIVE index to absolute immediately (abs = GetTop() + idx + 1);
+	// luaToJSON keeps the relative index and pushes it before Call. So after a
+	// SetTop(0) the two disagree, and the table-valued converter — the one that
+	// handles nested tables, arrays and mixed keys correctly — is the one that
+	// needs the stack shape kept intact.
+	//
+	// Which is exactly why the decode happens here, ONCE, immediately after
+	// GetGlobal+GetField, and every field read below is a pure function on the
+	// decoded Go value. The old readers repeated the stack dance per field.
+	m, ok := pluginFieldTable(L, "ui")
+	if !ok {
+		return nil, false
+	}
+	ctx := &decodeCtx{plugin: name}
+	ui := decodeUIExtension(ctx, m)
+
+	if ui.Page == nil && len(ui.Pages) == 0 && len(ui.Elements) == 0 {
+		return nil, false
+	}
+	if len(ctx.problems) > 0 {
+		log.Printf("[plugins] %s: %s", ctx.plugin, strings.Join(ctx.problems, "; "))
+	}
+	return &ui, true
+}
+
+// pluginFieldTable reads one top-level declaration table off the plugin global.
+func pluginFieldTable(L *golua.State, field string) (map[string]interface{}, bool) {
 	L.GetGlobal(pluginGlobal)
 	if L.IsNil(-1) {
 		L.SetTop(0)
 		return nil, false
 	}
-	L.GetField(-1, "ui")
+	L.GetField(-1, field)
 	if L.Type(-1) == golua.LUA_TNIL {
 		L.SetTop(0)
 		return nil, false
 	}
-	// luaTableToGoValue, NOT luaToJSON. This runs while the plugin is
-	// registering, BEFORE setupGlobals installs the json global that luaToJSON
-	// depends on — so the conversion fails and every field except the ones on
-	// Page/Pages (read separately) comes back empty. That is why visibility and
-	// state_scope silently read as unset even though the .lua declares them:
-	// the third field in this file to be lost to the same cause, after
-	// readPluginExports and readPluginHTTP.
-	var ui UIExtension
 	raw := luaTableToGoValue(L, -1)
 	L.SetTop(0)
-	if m, ok := raw.(map[string]interface{}); ok {
-		ui.Visibility = asString(m["visibility"])
-		if scope, ok := m["state_scope"].(map[string]interface{}); ok {
-			sc := &StateScope{}
-			if gs, ok := scope["global"].([]interface{}); ok {
-				for _, g := range gs {
-					if s := asString(g); s != "" {
-						sc.Global = append(sc.Global, s)
-					}
-				}
-			}
-			if ds, ok := scope["depth"].(map[string]interface{}); ok {
-				sc.Depth = map[string]int{}
-				for k, v := range ds {
-					if n, ok := asFloat(v); ok {
-						sc.Depth[k] = int(n)
-					}
-				}
-				sc.Depth = compileDepthMap(sc.Depth)
-			}
-			ui.StateScope = sc
-		}
-		if pg, ok := m["page"].(map[string]interface{}); ok {
-			ui.Page = uipageFromValue(pg)
-		}
-		if pgs, ok := m["pages"].([]interface{}); ok {
-			for _, it := range pgs {
-				if pm, ok := it.(map[string]interface{}); ok {
-					ui.Pages = append(ui.Pages, uipageFromValue(pm))
-				}
-			}
-		}
-		if els, ok := m["elements"].([]interface{}); ok {
-			for _, it := range els {
-				em, ok := it.(map[string]interface{})
-				if !ok {
-					continue
-				}
-				el := UIElement{
-					Target:   asString(em["target"]),
-					Anchor:   asString(em["anchor"]),
-					Mount:    asString(em["mount"]),
-					Audience: asString(em["audience"]),
-				}
-				if n, ok := asFloat(em["order"]); ok {
-					el.Order = int(n)
-				}
-				ui.Elements = append(ui.Elements, el)
-			}
+	m, ok := raw.(map[string]interface{})
+	return m, ok
+}
+
+// decodeUIExtension reads a plugin's `ui` declaration.
+//
+// The field list below is the single source of truth for what the gateway
+// understands. A declaration using anything else is REPORTED rather than
+// silently dropped, because the old hand-written reader ignored unknown keys with
+// no trace: `visibilty = "admin"` (one character off) restricted nothing and
+// looked exactly like a plugin that simply chose not to declare an audience.
+func decodeUIExtension(ctx *decodeCtx, m map[string]interface{}) UIExtension {
+	ctx.strictKeys(m, "ui", "visibility", "state_scope", "page", "pages", "elements")
+
+	var ui UIExtension
+	ui.Visibility = decString(m, "visibility")
+
+	if scope := decTable(m, "state_scope"); scope != nil {
+		ctx.strictKeys(scope, "ui.state_scope", "global", "depth")
+		sc := &StateScope{Global: decStringList(scope, "global")}
+		if d := decIntMap(scope, "depth"); len(d) > 0 {
+			sc.Depth = compileDepthMap(d)
 		}
+		ui.StateScope = sc
 	}
-	if ui.Page == nil && len(ui.Pages) == 0 && len(ui.Elements) == 0 {
-		return nil, false
+
+	if pg := decTable(m, "page"); pg != nil {
+		ui.Page = uipageFromValue(pg)
 	}
-	return &ui, true
+	for _, pm := range decTableList(m, "pages") {
+		ui.Pages = append(ui.Pages, uipageFromValue(pm))
+	}
+	for _, em := range decTableList(m, "elements") {
+		ctx.strictKeys(em, "ui.elements[]", "target", "anchor", "mount", "audience", "order")
+		el := UIElement{
+			Target:   decString(em, "target"),
+			Anchor:   decString(em, "anchor"),
+			Mount:    decString(em, "mount"),
+			Audience: decString(em, "audience"),
+		}
+		if n, ok := asFloat(em["order"]); ok {
+			el.Order = int(n)
+		}
+		ui.Elements = append(ui.Elements, el)
+	}
+	return ui
 }
 
 func (ps *Plugins) append(p *Plugin) {
@@ -1140,18 +1159,9 @@ func (ps *Plugins) HookErrors() map[string]map[string]interface{} {
 // "*" is deliberately NOT a synonym for "all": an audience that looks
 // restrictive in a config file while meaning everything is exactly the kind of
 // setting that fails silently.
+// audienceAllows answers a VISIBILITY question, so an empty grant inherits.
 func audienceAllows(want, role string) bool {
-	want = strings.ToLower(strings.TrimSpace(want))
-	if want == "" || want == "all" {
-		return true
-	}
-	role = strings.ToLower(strings.TrimSpace(role))
-	for _, part := range strings.Split(want, ",") {
-		if strings.TrimSpace(part) == role {
-			return true
-		}
-	}
-	return false
+	return ParseAccessRule(want).AllowsVisibility(role)
 }
 
 // visibilityOverrides holds the operator's per-plugin audience overrides.
@@ -1950,6 +1960,17 @@ func (ps *Plugins) invoke(p *Plugin, fn string, payload map[string]interface{})
 // quota report is a nested structure with numbers and booleans that must not be
 // flattened. Hook results still go through invoke, whose map-typed contract the
 // state merge depends on.
+// Call invokes an exported function on a plugin's SHARED VM, serialised by the
+// plugin's mutex.
+//
+// The lock is what makes one Lua VM safe to share, and it is held for the whole
+// call — including any network wait inside it. That is correct for a hook and
+// wrong for a function that blocks: twenty concurrent quota probes serialise
+// into the sum of their timeouts (measured: 29s on a 20-source gateway, against
+// 1.4–3.3s for a single probe with no contention).
+//
+// callOn below is the same work without the shared-VM assumption, so a caller
+// that owns a private VM can use it directly.
 func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface{}, error) {
 	ps.mu.RLock()
 	var p *Plugin
@@ -1967,12 +1988,27 @@ func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface
 		return nil, fmt.Errorf("plugin %s does not export %q", name, fn)
 	}
 	p.mu.Lock()
+	defer p.mu.Unlock()
 	w := p.state
 	if w == nil {
-		p.mu.Unlock()
 		return nil, fmt.Errorf("no state for plugin %s", name)
 	}
-	defer p.mu.Unlock()
+	return callOn(w, fn, args)
+}
+
+// callOn runs one exported function on a specific VM.
+//
+// It does NOT check the export allowlist and does NOT persist state: both belong
+// to the shared-VM path in Call. A shard VM has no plugin identity to check
+// against and no state file of its own — the export allowlist in Call is the
+// security boundary, and calling this with a function the plugin does not export
+// is a programming error, not an attack.
+//
+// The caller must own w exclusively. No lock is taken here.
+func callOn(w *worker, fn string, args map[string]interface{}) (interface{}, error) {
+	if w == nil || w.L == nil {
+		return nil, fmt.Errorf("no lua state")
+	}
 	L := w.L
 	L.SetTop(0)
 	defer L.SetTop(0)
@@ -2005,14 +2041,14 @@ func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface
 		L.Pop(1)
 	}
 	if errMsg != "" {
-		return nil, fmt.Errorf("plugin %s: %s", name, errMsg)
+		return nil, fmt.Errorf("plugin: %s", errMsg)
 	}
 	if L.GetTop() < 1 || L.IsNil(-1) {
 		return nil, nil
 	}
 	var out interface{}
 	if err := luaToJSON(L, -1, &out); err != nil {
-		return nil, nil
+		return nil, err
 	}
 	return out, nil
 }
@@ -2097,40 +2133,11 @@ func (h HTTPHandler) routePath() string {
 //
 // keyName is the caller key's configured name ("" when it has none), not the
 // secret.
+// allows answers a CAPABILITY question, so an empty grant denies. See
+// AccessRule.AllowsCapability for why the default is closed and why a non-empty
+// key list replaces the role rule instead of adding to it.
 func (h HTTPHandler) allows(role, keyName string) bool {
-	// Admin is always admitted. An operator must be able to inspect what a plugin
-	// exposes even while locked out of it, or a typo in a key list would leave
-	// nobody able to diagnose the endpoint they were trying to tighten.
-	if role == "admin" {
-		return true
-	}
-	// Named keys REPLACE the role rule rather than adding to it.
-	if len(h.Keys) > 0 {
-		if keyName == "" {
-			return false
-		}
-		for _, k := range h.Keys {
-			if strings.EqualFold(strings.TrimSpace(k), keyName) {
-				return true
-			}
-		}
-		return false
-	}
-	if h.Public {
-		return true
-	}
-	if len(h.Roles) == 0 {
-		// No declared roles and no public opt-out: treat as admin-only. A plugin
-		// author who forgets this gets a 403 rather than an open endpoint, and
-		// the error says which field to set.
-		return role == "admin"
-	}
-	for _, r := range h.Roles {
-		if strings.EqualFold(strings.TrimSpace(r), role) {
-			return true
-		}
-	}
-	return false
+	return accessRuleFor(h.Roles, h.Keys, h.Public).AllowsCapability(role, keyName)
 }
 
 func (ps *Plugins) StateVisibleTo(name, role string) bool {
@@ -2176,64 +2183,82 @@ func containsPage(pages []*UIPage, id string) bool {
 // not exist. Failing the whole plugin would take a working billing widget down
 // over a typo in an experimental endpoint.
 func readPluginHTTP(L *golua.State) []HTTPHandler {
-	L.GetGlobal(pluginGlobal)
-	if L.IsNil(-1) {
-		L.SetTop(0)
-		return nil
-	}
-	L.GetField(-1, "http")
-	if L.Type(-1) != golua.LUA_TTABLE {
-		L.SetTop(0)
-		return nil
-	}
-	// luaTableToGoValue, NOT luaToJSON and NOT a hand-rolled GetField loop.
+	// `http` is a LIST of declarations, each naming the handler function by
+	// STRING:
 	//
-	// luaToJSON needs the json global, which is not installed yet while a plugin
-	// is registering. And golua's GetField does NOT convert a numeric key: with
-	// a string index "1" it returns nil, so an index loop silently produced an
-	// empty list and every plugin appeared to declare no endpoints at all.
-	raw := luaTableToGoValue(L, -1)
-	L.SetTop(0)
-	items, ok := raw.([]interface{})
+	//	http = {
+	//	  { path = "health", method = "GET", roles = { "admin" }, handler = "health" },
+	//	}
+	//
+	// handler is a name, not a value. An inline function silently produced an
+	// empty handler field and the endpoint was dropped without a word.
+	raw, ok := pluginFieldValue(L, "http")
 	if !ok {
 		return nil
 	}
+	// A single declaration may be given as a bare table instead of a one-element
+	// list; accepting both is friendlier than demanding one spelling.
+	items, isList := raw.([]interface{})
+	if !isList {
+		if one, isTable := raw.(map[string]interface{}); isTable {
+			items = []interface{}{one}
+		} else {
+			return nil
+		}
+	}
+	ctx := &decodeCtx{plugin: "http"}
 	var list []HTTPHandler
 	for _, it := range items {
-		m, ok := it.(map[string]interface{})
+		h, ok := it.(map[string]interface{})
 		if !ok {
 			continue
 		}
-		h := HTTPHandler{
-			Path:        asString(m["path"]),
-			Method:      asString(m["method"]),
-			Handler:     asString(m["handler"]),
-			Description: asString(m["description"]),
-			Public:      m["public"] == true,
+		ctx.strictKeys(h, "http[]", "path", "method", "handler", "description",
+			"roles", "keys", "public")
+		dh := HTTPHandler{
+			Path:        decString(h, "path"),
+			Method:      decString(h, "method"),
+			Description: decString(h, "description"),
+			Roles:       decStringList(h, "roles"),
+			Keys:        decStringList(h, "keys"),
 		}
-		if roles, ok := m["roles"].([]interface{}); ok {
-			for _, r := range roles {
-				if s := asString(r); s != "" {
-					h.Roles = append(h.Roles, s)
-				}
+		// handler must be a NAME, not a value. asString would happily turn a
+		// function into its address ("function: 0x..."), which then failed much
+		// later at dispatch with no connection to the declaration that caused it.
+		// Rejecting it here names the field and the path.
+		if hv, present := h["handler"]; present {
+			name, isStr := hv.(string)
+			if !isStr {
+				ctx.warnf("http[%s]: handler must be the NAME of a function on the "+
+					"plugin table, not the function itself", decString(h, "path"))
+				continue
 			}
+			dh.Handler = name
 		}
-		if keys, ok := m["keys"].([]interface{}); ok {
-			for _, k := range keys {
-				if s := asString(k); s != "" {
-					h.Keys = append(h.Keys, s)
-				}
-			}
+		if p, ok := h["public"].(bool); ok {
+			dh.Public = p
 		}
 		// A malformed entry is dropped rather than fatal: one bad route must not
-		// cost a plugin its hooks and pages. That is also why the mounted count
-		// can be lower than the declared count — /api/plugins lists what mounted.
-		if method, mErr := normalizeHandlerMethod(h.Method); mErr == nil {
-			h.Method = method
-			if validateHandlerPath(h.Path) == nil && h.Handler != "" {
-				list = append(list, h)
-			}
+		// cost a plugin its hooks and pages. The reason is named so a plugin whose
+		// endpoint silently vanished can find out why.
+		method, mErr := normalizeHandlerMethod(dh.Method)
+		switch {
+		case mErr != nil:
+			ctx.warnf("http[%s]: %v", dh.Path, mErr)
+			continue
+		case validateHandlerPath(dh.Path) != nil:
+			ctx.warnf("http[%s]: invalid path", dh.Path)
+			continue
+		case dh.Handler == "":
+			ctx.warnf("http[%s]: handler must name a function on the plugin table "+
+				"(an inline function is not supported)", dh.Path)
+			continue
 		}
+		dh.Method = method
+		list = append(list, dh)
+	}
+	if len(ctx.problems) > 0 {
+		log.Printf("[plugins] %s", strings.Join(ctx.problems, "; "))
 	}
 	return list
 }

internal/lua/plugins/quota.lua

@@ -43,7 +43,7 @@ PLUGIN.name = "quota"
 PLUGIN.version = "1.0.0"
 PLUGIN.description = "多源套餐余量（余额 / 用量 / 滚动窗口）"
 
-PLUGIN.exports = { "collect_source", "collect_all" }
+PLUGIN.exports = { "collect_source", "collect_all", "due_sources", "memo_record" }
 
 -- plugin.state must exist before anything indexes it. The kernel restores
 -- persisted state OVER this table, so declaring it here is safe: a restart with
@@ -63,8 +63,23 @@ end
 -- hostOf extracts the host from a base_url, for showing WHICH upstream a row is
 -- about. Two sources on the same vendor (two Command Code keys) are
 -- indistinguishable without it.
+-- hostOf returns the host of a URL — the part memoisation is keyed on, so two
+-- sources pointing at the same host with different paths share one answer.
+--
+-- It used to be written with string.gmatch and a capture group. gmatch yields
+-- WHOLE matches, not captures, and with `^` it never advances, so the iterator
+-- returned nil on the first step and hostOf returned its `or u` fallback —
+-- the FULL URL.
+--
+-- The consequence was invisible and total. Every lookup was HOST_MEMO[nil] and
+-- every write was HOST_MEMO[nil], so:
+--   - the memo table never returned anything, for any host, ever
+--   - the negative memo, the TTL and (later) the backoff were all dead code
+--   - the 19 entries in this gateway's production state are unreachable: their
+--     keys are full URLs, while a lookup asks for nil
+-- Measured before the fix: hostOf(key) matched 0 of 19 production entries.
 local function hostOf(u)
-  return (string.gmatch(u, "^https?://([^/]+)")()) or u
+  return (string.match(u, "^https?://([^/]+)")) or u
 end
 
 -- get walks a dotted path through nested tables, returning nil if any hop is
@@ -298,6 +313,52 @@ local HOST_MEMO = {}
 local HOST_MEMO_AT = {}
 local PROBE_MEMO_TTL = 1800 -- 30 minutes
 
+-- Exponential backoff for hosts that answer nothing.
+--
+-- A flat TTL is the wrong shape for a dead upstream. Measured on this gateway:
+-- two hosts had been unreachable for 10 HOURS, and because PROBE_MEMO_TTL is 30
+-- minutes their negative memo was stale on every single page open — so each
+-- refresh re-paid the full cost of three adapters (measured: 24s per source
+-- before the probe budget was fixed, 16s after). A host that is down should not
+-- be re-examined every 30 minutes forever; each consecutive failure doubles the
+-- wait, so a long-dead host costs almost nothing while a host that was briefly
+-- down is retried within the TTL as before.
+--
+-- The cap matters as much as the doubling: without it a host that is down for a
+-- week stops being probed at all, and the page would quietly and permanently
+-- misreport it as having no balance endpoint.
+--
+-- SUCCESS resets the streak. This is only a probe-scheduling decision — a source
+-- that stops answering is still shown, still counted, and still retried.
+local BACKOFF_BASE = 1800        -- first retry after the usual TTL
+local BACKOFF_FACTOR = 2         -- doubles per consecutive miss
+local BACKOFF_MAX = 21600        -- 6 hours; beyond this we always re-probe
+
+-- backoffSeconds returns how long to wait before re-probing a host that has
+-- missed `streak` times in a row, and 0 when it is due now.
+--
+-- `at` is when the host last failed. streak is stored inside the memo table
+-- itself so it survives a restart: a process restart must not reset the backoff
+-- and re-pay the cost it was avoiding.
+-- Returns a BOOLEAN and nothing else.
+--
+-- In Lua only nil and false are falsy, so every path here must return one of
+-- those. Two versions of this got it wrong and both were caught by a test that
+-- asserts the type: `return 0` for "no streak" (0 is truthy, so a host with no
+-- streak — which is every memo written before this change — would be skipped
+-- forever, silently), and `return BACKOFF_MAX` on the cap path (21600 is
+-- truthy too, so a long-dead host would never be re-probed, which is the exact
+-- failure the cap exists to prevent).
+local function backoffSeconds(at, streak)
+  if streak == nil or streak <= 0 then return false end
+  local wait = BACKOFF_BASE
+  for _ = 2, streak do
+    wait = wait * BACKOFF_FACTOR
+    if wait >= BACKOFF_MAX then wait = BACKOFF_MAX break end
+  end
+  return (os.time() - (at or 0)) < wait
+end
+
 -- PROBE_BUDGET_MS caps one collect pass. Without it a page open over a set of
 -- dead upstreams runs to the sum of every timeout — measured at 149s — and the
 -- plugin's write lock is held the whole time.
@@ -325,6 +386,12 @@ local function collectSource(src, budget)
   -- exactly the confusion that made an earlier revision drop five sources.
   if type(known) == "table" then
     local seen = known.at or 0
+      -- A streak is a host that has failed repeatedly. Skip it outright until its
+      -- backoff expires, instead of re-paying three adapter timeouts every TTL.
+      if backoffSeconds(seen, known.streak) then
+        return { name = src.name, host = host, skipped = true, backing_off = true,
+                 reason = "连续探测失败 " .. tostring(known.streak) .. " 次，退避中（等待下次重试）" }
+      end
     if PROBE_MEMO_TTL > 0 and (os.time() - seen) > PROBE_MEMO_TTL then
       known = nil
     else
@@ -394,8 +461,21 @@ local function collectSource(src, budget)
   -- would make the page quietly stop reporting a source that started working.
   -- The stamp is what turns a 60s first load into a fast steady state without
   -- freezing what the page can see.
-  if rep == nil and not HOST_MEMO[host] then
-    HOST_MEMO[host] = { at = os.time() }
+  -- Record the negative result, and GROW THE STREAK.
+  --
+  -- The guard used to be `not HOST_MEMO[host]`, which is true only the FIRST time
+  -- a host fails: on every later failure the memo already held a table, the
+  -- guard was false, and nothing was written. The streak therefore stayed pinned
+  -- at 1 forever and the backoff never grew. Measured while writing this: the
+  -- memo read back `{at:... streak:1}` after four consecutive failures.
+  --
+  -- A success overwrites the memo with a vendor string, so a host that recovers
+  -- resets its streak automatically.
+  if rep == nil then
+    local prev = HOST_MEMO[host]
+    local streak = 1
+    if type(prev) == "table" then streak = (prev.streak or 1) + 1 end
+    HOST_MEMO[host] = { at = os.time(), streak = streak }
   end
   if rep ~= nil then
     HOST_MEMO_AT[host] = os.time()
@@ -421,6 +501,96 @@ end
 -- and doing 21 sources in sequence overran the HTTP call's own budget — the
 -- whole page returned 502 while individual probes were still in flight. The
 -- sequential entry point stays for tests and for anyone scripting the plugin.
+-- due_sources answers, on the SHARED VM, which sources are worth probing right
+-- now — and it makes NO network requests.
+--
+-- This exists because of a conflict between two facts:
+--
+--   - Probing is network-bound, and Plugins.Call holds the plugin mutex for the
+--     whole Lua function, so twenty probes through /call/ serialise. Measured:
+--     29s on a 20-source gateway. The fix is ShardCall, which runs each call on
+--     its own VM — and a shard VM has its own PLUGIN.state.
+--   - The backoff lives in PLUGIN.state (HOST_MEMO). A shard therefore cannot
+--     see it, cannot record it, and a host that fails keeps being probed on every
+--     page open. Measured: fanout left the memo at 0 entries, collect_all reached
+--     17.
+--
+-- So the split is: the DECISION runs where the state is (one shared-VM call,
+-- microseconds), and only the hosts that pass it get sharded.
+--
+-- Cheap on purpose. It reads HOST_MEMO and nothing else; if the state has not
+-- been restored yet, every host is due, which is the correct answer for a first
+-- call.
+-- memo_record writes the outcome of a sharded pass into the SHARED VM's memo.
+--
+-- A shard runs on its own VM and therefore cannot record anything: measured, a
+-- page of fanouts left the memo at 0 entries, so the backoff never accumulated
+-- and every refresh re-probed the dead hosts. This is the other half of
+-- due_sources — the decision is taken here, the outcome is written back here, and
+-- only the network wait happens on the shards.
+--
+-- It records FAILURES as well as successes. Recording only successes is what
+-- left every source with no balance endpoint re-probing three adapters forever.
+PLUGIN.memo_record = function(payload)
+  if type(payload) ~= "table" then return nil, "expected table" end
+  if not next(HOST_MEMO) then
+    local st = PLUGIN.state
+    if type(st) == "table" and type(st.probe_memo) == "table" then
+      for k, v in pairs(st.probe_memo) do HOST_MEMO[k] = v end
+    end
+  end
+  local rows = payload.results
+  if type(rows) ~= "table" then return nil, "missing results" end
+  local now = os.time()
+  for _, r in ipairs(rows) do
+    if type(r) == "table" and type(r.host) == "string" and r.host ~= "" then
+      local host = hostOf(r.host)
+      if r.vendor then
+        -- A source that answered resets its streak: the memo becomes a vendor
+        -- string, which every reader already treats as a positive memo.
+        HOST_MEMO[host] = r.vendor
+      else
+        local prev = HOST_MEMO[host]
+        local streak = 1
+        if type(prev) == "table" then streak = (prev.streak or 1) + 1 end
+        HOST_MEMO[host] = { at = now, streak = streak }
+      end
+    end
+  end
+  PLUGIN.state.probe_memo = HOST_MEMO
+  return true
+end
+
+PLUGIN.due_sources = function(payload)
+  if type(payload) ~= "table" then return nil, "expected table" end
+  if not next(HOST_MEMO) then
+    local st = PLUGIN.state
+    if type(st) == "table" and type(st.probe_memo) == "table" then
+      for k, v in pairs(st.probe_memo) do HOST_MEMO[k] = v end
+    end
+  end
+  local srcs = payload.sources
+  if type(srcs) ~= "table" then return nil, "missing sources" end
+  local due, backing = {}, {}
+  for _, s in ipairs(srcs) do
+    if type(s) == "table" and type(s.base_url) == "string" and s.base_url ~= "" then
+      local base = (s.base_url:gsub("/+$", ""))
+      local known = HOST_MEMO[hostOf(base)]
+      if type(known) == "table" and backoffSeconds(known.at or 0, known.streak) then
+        -- skipped matters: the page filters on it. Without the flag a backed-off
+        -- host reaches the card renderer, which has no field to match and falls
+        -- through to its "has data" default — so thirteen dead upstreams were
+        -- displayed as if they had answered.
+        backing[#backing + 1] = { name = s.name, host = hostOf(base), skipped = true, backing_off = true,
+                                  reason = "连续探测失败 " .. tostring(known.streak) .. " 次，退避中（等待下次重试）" }
+      else
+        due[#due + 1] = s
+      end
+    end
+  end
+  return { due = due, backing_off = backing }
+end
+
 PLUGIN.collect_all = function(payload)
   if type(payload) ~= "table" then return nil, "expected table" end
   if not next(HOST_MEMO) then
@@ -431,10 +601,15 @@ PLUGIN.collect_all = function(payload)
   end
   local srcs = payload.sources
   if type(srcs) ~= "table" then return nil, "missing sources" end
-  local t1 = os.clock()
+  -- os.time, NOT os.clock. os.clock measures CPU time, and a probe spends
+  -- essentially none: measured on this gateway, 3s blocked on an HTTP request
+  -- advanced os.clock by 0.001s. A budget written in os.clock therefore never
+  -- fires, which is why a dead upstream ran the full
+  -- 4s timeout x 2 retries x 3 adapters = 24s against a "12 second" budget.
+  local t1 = os.time()
   local lim1 = PROBE_BUDGET_MS / 1000
   local function spent()
-    return (os.clock() - t1) > lim1
+    return (os.time() - t1) > lim1
   end
   local budget = { exhausted = spent }
   local out = {}
@@ -463,10 +638,10 @@ PLUGIN.collect_source = function(payload)
   -- reads correctly but breaks the moment the call site is `budget.exhausted()`
   -- from inside a pcall-guarded hook where the upvalue chain differs; the
   -- explicit capture below cannot go wrong that way.
-  local t0 = os.clock()
+  local t0 = os.time()
   local limit = PROBE_BUDGET_MS / 1000
   local function exhausted()
-    return (os.clock() - t0) > limit
+    return (os.time() - t0) > limit
   end
   local budget = { exhausted = exhausted }
   for _, s in ipairs(srcs) do
@@ -653,7 +828,14 @@ PLUGIN.ui = {
     }
     if(r.hard_limit!==undefined && r.hard_limit!==null)
       return {cls:"ok", pill:'<span class="qa-pill ok">按量计费</span>'};
-    return {cls:"", pill:'<span class="qa-pill">有数据</span>'};
+    // Nothing here matched. Say so.
+    //
+    // The old default was a neutral "有数据", which is a claim: it says the probe
+    // answered. A record that reached this point answered with no field the page
+    // knows — a back-off notice, an error shape we do not recognise, or a vendor
+    // that added a new response. Calling that "has data" is how thirteen dead
+    // upstreams ended up rendered as healthy. Absence of evidence is not evidence.
+    return {cls:"unknown", pill:'<span class="qa-pill">未探测</span>'};
   }
   function card(r){
     var st=state(r);
@@ -666,6 +848,13 @@ PLUGIN.ui = {
       h+='<div class="qa-fail">'+escH(r.error)+'</div></div>';
       return h;
     }
+    // A record with nothing but a reason has no balance, no month, no windows —
+    // the card renders as a bare header with an empty body. Say why instead.
+    if(r.reason && !r.error && (r.balance===undefined||r.balance===null)
+       && (r.hard_limit===undefined||r.hard_limit===null)){
+      h+='<div class="qa-fail">'+escH(r.reason)+'</div>';
+      return h+'</div>';
+    }
     h+=balLine(r);
     var rows="";
     rows+=month(r);
@@ -690,7 +879,10 @@ PLUGIN.ui = {
       var rs = (st && st.reports) || [];
       if(!rs.length) return false;
       var real = rs.filter(function(r){ return r && !r.skipped; });
-      var skipped = rs.filter(function(r){ return r && r.skipped; }).map(function(r){ return r.name; });
+      // Records, not names — same reason as the refresh path. Both call sites were
+    // mapping to names while render() was changed to group on the flag; fixing
+    // one and shipping it is what produced a footer with thirteen blank entries.
+    var skipped = rs.filter(function(r){ return r && r.skipped; });
       render(real, skipped);
       var sub=document.getElementById("qa-sub");
       if(sub && st.updated_at){
@@ -707,10 +899,22 @@ PLUGIN.ui = {
       host.innerHTML='<div class="qa-empty">没有源提供可读的余额接口</div>';
     } else {
       host.innerHTML=real.map(card).join("");
-      if(skipped && skipped.length){
-        host.innerHTML+='<div class="qa-skip">未列出 '+skipped.length+
-          ' 个源（未暴露余额接口）：'+escH(skipped.join("、"))+'</div>';
-      }
+        if(skipped && skipped.length){
+          // Two reasons a source has no card, and they are not interchangeable.
+          // "This upstream exposes no balance endpoint" and "this upstream is
+          // backing off after failing" are different facts; collapsing them into
+          // one list tells the operator their dead host simply has no dashboard.
+          var backing = skipped.filter(function(r){ return r.backing_off===true; });
+          var silent  = skipped.filter(function(r){ return r.backing_off!==true; });
+          if(backing.length){
+            host.innerHTML+='<div class="qa-skip">'+backing.length+
+              ' 个源正在退避（连续探测失败，等下次重试）：'+escH(backing.map(function(r){return r.name;}).join("、"))+'</div>';
+          }
+          if(silent.length){
+            host.innerHTML+='<div class="qa-skip">未列出 '+silent.length+
+              ' 个源（未暴露余额接口）：'+escH(silent.map(function(r){return r.name;}).join("、"))+'</div>';
+          }
+        }
     }
     var sub=document.getElementById("qa-sub");
     if(sub) sub.textContent = real.length+" 个源 · "+bad.length+" 个读取失败";
@@ -732,15 +936,39 @@ PLUGIN.ui = {
     var sub=document.getElementById("qa-sub"); if(sub) sub.textContent="探测中…";
     var out;
     try{
-      // One call per source, fired together. Probing is network-bound, so a
-    // sequential collect_all over 21 sources blows past the call's time budget
-    // and every card comes back as one 502 — the probes themselves were fine.
-    // Concurrency also makes the page honest about which sources are slow:
-    // each card appears as its own answer lands.
-    out = await Promise.all(srcs.map(function(s){
-      return pluginAPI.call("quota","collect_source",{sources:[s],name:s.name})
-        .catch(function(e){ return {name:s.name,skipped:true,error:String(e.message||e)}; });
-    }));
+    // ONE request, fanned out inside the gateway.
+    //
+    // The previous version fired Promise.all at /call/ — twenty concurrent
+    // browser requests. That looked concurrent and was not: each one queued on
+    // the plugin mutex for the whole Lua call, network wait included. Measured on
+    // the live gateway: 29s of wall clock for twenty probes, against 1.4–3.3s for
+    // a single probe with no contention. The page sat on "探测中…" for half a
+    // minute and read as broken.
+    //
+    // The comment that used to sit here claimed concurrency avoided a call budget
+    // overrun. It could not have: the requests were serialised, so the budget they
+    // overran was the very same one. It is a single call now because a
+    // request-per-item fan-out cannot be parallel from the client at all.
+    // Which sources are worth probing is decided on the SHARED VM, where the
+    // backoff state lives; only those get sharded.
+    //
+    // Asking each shard to decide for itself does not work: a shard has its own
+    // PLUGIN.state, so it cannot see the memo and a host that keeps failing is
+    // re-probed on every refresh forever. Measured before this split: fanout left
+    // the memo at 0 entries and every refresh cost the full 12s.
+    var plan = await pluginAPI.call("quota","due_sources",{sources:srcs});
+    // An EMPTY Lua table decodes as {} on the Go side, not [] — and {} is truthy,
+    // so `|| []` does not replace it. An empty object reaching out.concat()
+    // throws, and the whole refresh reports a failure. Array.isArray is the check
+    // that matches what the plugin actually returns.
+    var due = (plan && Array.isArray(plan.due)) ? plan.due : [];
+    var out = (plan && Array.isArray(plan.backing_off)) ? plan.backing_off : [];
+
+    if (due.length) {
+      out = out.concat(await pluginAPI.fanout("quota","collect_source",due.map(function(s){
+        return {sources:[s],name:s.name};
+      })));
+    }
     // Mirror the host→vendor map the plugin keeps: a source that answered is
     // worth probing first next time, and the page has everything it needs to
     // record that without another round trip.
@@ -749,6 +977,20 @@ PLUGIN.ui = {
       if(r && !r.skipped && !r.error && r.host && r.vendor) memo[r.host] = r.vendor;
     });
     window.__qaMemo = memo;
+    // Hand the whole pass back to the SHARED VM so the backoff has something to
+    // work from next time. Without this the shards' results die with their VMs:
+    // measured, the memo stayed at 0 entries and every refresh paid full price.
+    await pluginAPI.call("quota","memo_record",{results:(out||[]).map(function(r){
+      if(!r) return null;
+      // A host that is merely BACKING OFF has not just failed; recording it as a
+      // failure would grow the streak on a probe that never happened. Keyed on the
+      // flag rather than the wording, so translating the message cannot silently
+      // turn every skipped host into an ever-growing streak.
+      if(r.backing_off===true) return null;
+      if(r.error) return {host:r.host};
+      if(r.skipped) return {host:r.host};
+      return {host:r.host, vendor:r.vendor};
+    }).filter(Boolean)});
     }catch(e){
       if(btn) btn.disabled=false;
       if(sub) sub.textContent="探测失败: "+e.message;
@@ -756,7 +998,11 @@ PLUGIN.ui = {
     }
     var all=out||[];
     var real=all.filter(function(r){return !r.skipped;});
-    var skipped=all.filter(function(r){return r.skipped;}).map(function(r){return r.name;});
+    // The records, not their names — render() groups skipped sources by the
+    // backing_off flag, which a bare name cannot carry. This call site was left
+    // mapping to names while the other was fixed, and the mismatch showed up as
+    // a footer listing thirteen unnamed sources.
+    var skipped=all.filter(function(r){return r.skipped;});
     // real may already contain errors; collect_all keeps them for visibility.
     render(real, skipped);
     if(btn) btn.disabled=false;

internal/lua/quota_backoff_test.go

@@ -0,0 +1,650 @@
+package lua
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+// The backoff lives in quota.lua — a shipped plugin, not Go — so these tests read
+// the file that actually ships (readShippedPlugin) rather than a copy that can
+// drift from it.
+
+// TestQuotaBackoffIsAppliedInSource guards the wiring.
+//
+// It exists because of a mutation that removed the call site entirely (`if false
+// then`) and every other test in this file still passed. The other tests cover
+// the decision FUNCTION and the streak RECORDING; neither exists in order to be
+// unused. A backoff nobody consults is indistinguishable from no backoff, and the
+// symptom is the original one — a dead host re-probed on every page open.
+func TestQuotaBackoffIsAppliedInSource(t *testing.T) {
+	src := readShippedPlugin(t, "quota.lua")
+	if !strings.Contains(src, "if backoffSeconds(seen, known.streak) then") {
+		t.Error("quota.lua never consults backoffSeconds for a host with a streak — " +
+			"the backoff is computed but not applied")
+	}
+	// The reason is required, but matched on the FLAG rather than the wording:
+	// the message is localised, and grouping by text would merge "backing off"
+	// into "exposes no balance endpoint" the moment someone translates it.
+	if strings.Count(src, "reason = \"连续探测失败 ") < 2 {
+		t.Error("both skip paths (collectSource and due_sources) must carry a reason; " +
+			"a silently skipped host looks exactly like a host with no balance endpoint")
+	}
+	if strings.Count(src, "backing_off = true") < 2 {
+		t.Error("both skip paths must set the explicit backing_off flag; the page " +
+			"groups skipped sources by it, not by the message text")
+	}
+	// The recording guard has to REPLACE the memo, not merge into it. The bug this
+	// guards was `if rep == nil and not HOST_MEMO[host]`, which is true only on a
+	// host's first failure — so the streak stayed pinned at 1 forever. Asserted on
+	// the source because the mutation that reintroduces it leaves the visible
+	// behaviour of the FIRST collect unchanged, and that is all the integration
+	// test can reach inside one process.
+	if !strings.Contains(src, "if rep == nil then") {
+		t.Error("the negative result is recorded under a guard that skips every " +
+			"failure after the first — the streak will never grow")
+	}
+	if strings.Contains(src, "and not HOST_MEMO[host] then") {
+		t.Error("the recording guard excludes a host that already has a memo, so " +
+			"consecutive failures are never counted")
+	}
+}
+
+// TestQuotaBackoffWindowIsCapped guards the failure mode an exponential backoff
+// has when it is not capped: a host that is down for a week stops being probed
+// and the page permanently misreports it as having no balance endpoint.
+func TestQuotaBackoffWindowIsCapped(t *testing.T) {
+	src := readShippedPlugin(t, "quota.lua")
+	for _, want := range []string{"BACKOFF_BASE", "BACKOFF_FACTOR", "BACKOFF_MAX"} {
+		if !strings.Contains(src, want) {
+			t.Errorf("quota.lua no longer declares %s — the backoff was weakened or removed", want)
+		}
+	}
+}
+
+// TestQuotaBackoffWindows walks the whole policy through the plugin's own
+// decision function: for each consecutive failure, how long the host must be left
+// alone, that it is still skipped halfway through that window, that it is retried
+// once the window passes, and that an absurdly old host is always due.
+//
+// Why the function rather than the running plugin: HOST_MEMO is a process-local
+// cache, restored on a process's first collect and written back afterwards, so
+// ageing its memo from Go changes nothing the plugin can see. Two earlier versions
+// of this file spent a long time on exactly that dead end, and a test that keeps
+// breaking for reasons unrelated to the code under test is worse than no test.
+//
+// The function is the plugin's own — reached through a shim that appends one
+// export — so changing quota.lua's arithmetic moves this test. It is not a
+// reimplementation that would agree with a broken plugin.
+func TestQuotaBackoffWindows(t *testing.T) {
+	ps := quotaWithDecision(t)
+
+	// streak, wait seconds. The last rows are at and past the ceiling.
+	for _, c := range []struct {
+		streak int
+		wait   int
+	}{
+		{1, 1800},
+		{2, 3600},
+		{3, 7200},
+		{4, 14400},
+		{5, 21600},
+		{9, 21600},
+		{20, 21600},
+	} {
+		if got := quotaSkip(t, ps, c.streak, c.wait/2); !got {
+			t.Logf("  streak=%d age=%d → %v", c.streak, c.wait/2, got)
+			t.Errorf("streak=%d: not skipping %ds into a %ds window — the wait is too short",
+				c.streak, c.wait/2, c.wait)
+		}
+		if quotaSkip(t, ps, c.streak, c.wait+60) {
+			t.Errorf("streak=%d: still skipping %ds after its %ds window expired — "+
+				"this host would never be retried again", c.streak, c.wait+60, c.wait)
+		}
+		if quotaSkip(t, ps, c.streak, 1<<30) {
+			t.Errorf("streak=%d: a host silent for decades is still being skipped", c.streak)
+		}
+	}
+}
+
+// TestQuotaBackoffReturnsOnlyBooleans pins a type rule that produced two silent
+// bugs.
+//
+// In Lua only nil and false are falsy, so every path of a predicate must return
+// one of those. This function had two paths returning numbers: `return 0` for
+// "no streak", which is truthy and made every caller skip a host that had never
+// been probed; and `return BACKOFF_MAX` on the cap path, also truthy, which made
+// the ceiling mean "never retried" — the exact failure the ceiling exists to
+// prevent.
+//
+// Both shipped far enough to be found by this assertion rather than by a user
+// staring at a page that had quietly stopped probing anything.
+func TestQuotaBackoffReturnsOnlyBooleans(t *testing.T) {
+	ps := quotaWithDecision(t)
+	for _, c := range []struct {
+		name string
+		age  int
+		strk interface{}
+	}{
+		{"no streak field (every memo written before the backoff existed)", 0, nil},
+		{"zero streak", 0, 0},
+		{"negative streak", 0, -1},
+		{"inside the window", 10, 1},
+		{"outside the window", 999999, 1},
+		{"at the ceiling", 10, 20},
+	} {
+		out, err := ps.Call("quota", "__decision", map[string]interface{}{
+			"at":     float64(time.Now().Unix() - int64(c.age)),
+			"streak": c.strk,
+		})
+		if err != nil {
+			t.Fatalf("%s: %v", c.name, err)
+		}
+		// A bare boolean is a valid answer: when backoffSeconds returns false the
+		// shim's table round-trips as just `false`, because a Lua table holding a
+		// single falsy field collapses on encode.
+		switch v := out.(type) {
+		case bool:
+			_ = v // a boolean is exactly what this asserts
+		case map[string]interface{}:
+			if _, ok := v["r"].(bool); !ok {
+				t.Errorf("%s: backoffSeconds returned %T (%v) — in Lua a number is "+
+					"TRUTHY, so this caller would skip the host", c.name, v["r"], v["r"])
+			}
+		default:
+			t.Errorf("%s: decision returned %T (%v); want a boolean", c.name, out, out)
+		}
+	}
+}
+
+// TestQuotaBackoffRecordsConsecutiveFailures covers the recording guard through
+// the running plugin, because that part IS observable from outside.
+//
+// The guard used to be `not HOST_MEMO[host]`, true only on a host's first
+// failure. After that the memo held a table, the guard was false, nothing was
+// written, and the streak stayed pinned at 1 forever — measured by reading back
+// {at:... streak:1} after four consecutive failures.
+//
+// Only two things are asserted, because only two are reachable in one process:
+// a first failure records streak 1, and a SKIP does not grow it. A skipped probe
+// is not a failure, and a skip that counted would let a never-probed host grow
+// without end and never be retried.
+func TestQuotaBackoffRecordsConsecutiveFailures(t *testing.T) {
+	srv := newCountingServer(t)
+	ps := newQuotaPlugins(t, srv.URL)
+
+	args := map[string]interface{}{
+		"sources": []interface{}{map[string]interface{}{
+			"name": "dead", "base_url": srv.URL + "/v1", "api_key": "k",
+		}},
+		"name": "dead",
+	}
+	streak := func() float64 {
+		st, _ := ps.State("quota").(map[string]interface{})
+		memo, _ := st["probe_memo"].(map[string]interface{})
+		for _, v := range memo {
+			if e, ok := v.(map[string]interface{}); ok {
+				if f, ok := e["streak"].(float64); ok {
+					return f
+				}
+			}
+		}
+		t.Fatalf("no negative memo recorded: %#v", memo)
+		return 0
+	}
+
+	if _, err := ps.Call("quota", "collect_source", args); err != nil {
+		t.Fatal(err)
+	}
+	if got := streak(); got != 1 {
+		t.Fatalf("streak after the first failure = %.0f, want 1", got)
+	}
+
+	before := srv.Hits()
+	if _, err := ps.Call("quota", "collect_source", args); err != nil {
+		t.Fatal(err)
+	}
+	if served := srv.Hits() - before; served != 0 {
+		t.Errorf("a host inside its backoff window still made %d requests — the "+
+			"whole feature is to not make any", served)
+	}
+	if got := streak(); got != 1 {
+		t.Errorf("a SKIP changed the streak to %.0f — a skipped probe is not a failure", got)
+	}
+}
+
+// TestQuotaMemoKeyIsTheHost pins the key the memo is written and read under.
+//
+// This is the foundation the whole memoisation rests on, and it was broken in the
+// most total way possible: hostOf was written with string.gmatch and a capture
+// group, gmatch yields whole matches rather than captures, and with `^` the
+// iterator never advanced — so it returned nil and every lookup and every write
+// went to HOST_MEMO[nil]. The memo therefore never returned anything for any
+// host, and the negative memo, the TTL and the backoff on top of them were all
+// dead code that looked alive.
+//
+// Measured on this gateway before the fix: hostOf(key) matched 0 of the 19
+// entries in production state.
+//
+// The check is on the shipped source plus its runtime behaviour, because the
+// failure mode is invisible from outside — a memo that never matches anything
+// behaves exactly like a memo that has nothing to remember.
+func TestQuotaMemoKeyIsTheHost(t *testing.T) {
+	src := readShippedPlugin(t, "quota.lua")
+	if !strings.Contains(src, "string.match(u,") {
+		t.Error("hostOf does not use string.match — string.gmatch yields whole " +
+			"matches, so the capture group is ignored and the function returns nil")
+	}
+
+	ps := quotaWithDecision(t)
+	for _, c := range []struct{ in, want string }{
+		{"https://qijiar.top/v1", "qijiar.top"},
+		{"https://api.deepseek.com", "api.deepseek.com"},
+		{"http://127.0.0.1:19900", "127.0.0.1:19900"},
+		{"https://s2a.frankhan.top:14514/v1", "s2a.frankhan.top:14514"},
+	} {
+		out, err := ps.Call("quota", "__hostof", map[string]interface{}{"u": c.in})
+		if err != nil {
+			t.Fatalf("hostOf(%s): %v", c.in, err)
+		}
+		mm, _ := out.(map[string]interface{})
+		got, _ := mm["h"].(string)
+		if got != c.want {
+			t.Errorf("hostOf(%q) = %q, want %q", c.in, got, c.want)
+		}
+		if got == "" {
+			t.Errorf("hostOf(%q) returned nothing — every lookup would be "+
+				"HOST_MEMO[nil]", c.in)
+		}
+	}
+
+	// The same host under two paths must produce the SAME key, which is the whole
+	// reason the memo is keyed this way.
+	out, err := ps.Call("quota", "__hostof", map[string]interface{}{"u": "https://x.example/a/v1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	out2, err := ps.Call("quota", "__hostof", map[string]interface{}{"u": "https://x.example/b/v1"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	m1, _ := out.(map[string]interface{})
+	m2, _ := out2.(map[string]interface{})
+	if m1["h"] != m2["h"] {
+		t.Errorf("two paths on one host produced different keys: %v and %v — "+
+			"they would never share a memo", m1["h"], m2["h"])
+	}
+}
+
+// --- helpers ---
+
+// readShippedPlugin returns the bundled source, so the tests exercise the file
+// that actually ships rather than a copy that can drift from it.
+func readShippedPlugin(t *testing.T, name string) string {
+	t.Helper()
+	b, err := bundledAdapters.ReadFile("plugins/" + name)
+	if err != nil {
+		t.Fatalf("read embedded %s: %v", name, err)
+	}
+	return string(b)
+}
+
+// quotaWithDecision loads quota.lua with its local backoffSeconds exposed as an
+// export. The plugin's own exports list is appended to, so the export allowlist is
+// satisfied exactly as it is in production.
+func quotaWithDecision(t *testing.T) *Plugins {
+	t.Helper()
+	src := readShippedPlugin(t, "quota.lua")
+	shim := strings.Replace(src, "return PLUGIN",
+		"PLUGIN.exports[#PLUGIN.exports+1] = '__decision'\n"+
+			// Wrapped, not exported raw: backoffSeconds takes (at, streak)
+			// positionally, and the test calls it with a table because that is the
+			// shape a Go caller can pass. Exporting the raw function made every
+			// assertion read `at = <the argument table>, streak = nil`, so the
+			// function took its "no streak" path and returned false for every input
+			// — which looked exactly like a backoff that never fires.
+			"PLUGIN.__decision = function(q) return { r = backoffSeconds(q.at, q.streak) } end\n"+
+			"PLUGIN.exports[#PLUGIN.exports+1] = '__hostof'\n"+
+			"PLUGIN.__hostof = function(q) return { h = hostOf(q.u) } end\n"+
+			"return PLUGIN", 1)
+	if shim == src {
+		t.Fatal("could not inject the decision shim — quota.lua no longer ends with 'return PLUGIN'")
+	}
+	return newQuotaPluginsWith(t, shim, "127.0.0.1")
+}
+
+// quotaSkip asks the plugin whether a host that failed `streak` times and was
+// last heard from `age` seconds ago should be skipped.
+func quotaSkip(t *testing.T, ps *Plugins, streak, age int) bool {
+	out, err := ps.Call("quota", "__decision", map[string]interface{}{
+		"at":     float64(time.Now().Unix() - int64(age)),
+		"streak": float64(streak),
+	})
+	if err != nil {
+		t.Fatalf("decision: %v", err)
+	}
+	// A single-field table collapses to the bare value on encode, so both shapes
+	// mean the same thing here.
+	switch v := out.(type) {
+	case bool:
+		return v
+	case map[string]interface{}:
+		b, ok := v["r"].(bool)
+		if !ok {
+			t.Fatalf("decision returned %T, not a boolean: %#v", v["r"], v)
+		}
+		return b
+	default:
+		t.Fatalf("decision returned %T (%v), not a boolean", out, out)
+		return false
+	}
+}
+
+// newQuotaPlugins boots the shipped quota.lua against a temp plugin dir.
+func newQuotaPlugins(t *testing.T, baseURL string) *Plugins {
+	t.Helper()
+	return newQuotaPluginsWith(t, readShippedPlugin(t, "quota.lua"), baseURL)
+}
+
+func newQuotaPluginsWith(t *testing.T, src, baseURL string) *Plugins {
+	t.Helper()
+	dir := t.TempDir()
+	pluginDir := filepath.Join(dir, "plugins")
+	if err := os.MkdirAll(pluginDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	vm := NewVM(filepath.Join(dir, "adapters"))
+	if err := vm.Start(); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(vm.Stop)
+	ps := NewPlugins(vm, pluginDir)
+	t.Cleanup(ps.Close)
+	ps.SetOutboundHosts([]string{"127.0.0.1"})
+	if err := os.WriteFile(filepath.Join(pluginDir, "quota.lua"), []byte(src), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := ps.LoadDir(); err != nil {
+		t.Fatal(err)
+	}
+	p := ps.find("quota")
+	if p == nil {
+		t.Fatal("quota plugin did not load")
+	}
+	if p.LoadError != "" {
+		t.Fatalf("quota plugin failed to compile: %s", p.LoadError)
+	}
+	_ = baseURL
+	return ps
+}
+
+// countingServer answers 404 on every path and counts what it is asked, so a test
+// can tell "the plugin skipped" (no requests) from "the plugin probed and found
+// nothing" (requests, no data).
+//
+// It used to answer 200 with a non-balance body, and the commandcode adapter
+// accepted that — so the host was memoised as a SUCCESS and the negative path was
+// never reached. The test then failed for the right reason, and would have
+// passed for the wrong one had the assertion been looser.
+type countingServer struct {
+	*httptest.Server
+	hits int
+}
+
+func (c *countingServer) Hits() int { return c.hits }
+
+func newCountingServer(t *testing.T) *countingServer {
+	t.Helper()
+	cs := &countingServer{}
+	cs.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		cs.hits++
+		w.WriteHeader(http.StatusNotFound)
+		_, _ = w.Write([]byte(`{"error":"not found"}`))
+	}))
+	t.Cleanup(cs.Close)
+	return cs
+}
+
+// TestQuotaDueSourcesAndMemoRecordCoverTheShardGap pins the split that makes the
+// backoff work at all under sharding.
+//
+// The conflict, measured on a live 20-source gateway:
+//   - probing is network-bound and Plugins.Call holds the plugin mutex for the
+//     whole Lua call, so /call/ serialises (29s)
+//   - so the page uses ShardCall — and a shard has its own PLUGIN.state
+//   - and the backoff lives in PLUGIN.state: fanout left the memo at 0 entries,
+//     so dead hosts were re-probed forever
+//
+// The resolution is to put the DECISION and the RECORDING back on the shared VM
+// (microseconds, no network) and shard only the network wait. These two exports
+// are that seam, and without them the fanout path silently defeats the backoff.
+func TestQuotaDueSourcesAndMemoRecordCoverTheShardGap(t *testing.T) {
+	src := readShippedPlugin(t, "quota.lua")
+	for _, want := range []string{"due_sources", "memo_record"} {
+		if !strings.Contains(src, want) {
+			t.Errorf("quota.lua does not export %s — a sharded pass cannot see the "+
+				"backoff state, nor leave any behind", want)
+		}
+		if !strings.Contains(src, `"`+want+`"`) {
+			t.Errorf("%s is not listed in PLUGIN.exports, so the page cannot call it", want)
+		}
+	}
+	// due_sources must not probe. If it did, the whole point of asking it first
+	// would be lost: the cost would be paid before any shard runs.
+	if strings.Contains(src, "due_sources = function(payload)") &&
+		strings.Contains(src, "due_sources") {
+		// The body is small; check it does not call collectSource.
+		i := strings.Index(src, "due_sources = function(payload)")
+		j := strings.Index(src[i:], "\nend\n")
+		if j > 0 {
+			body := src[i : i+j]
+			if strings.Contains(body, "collectSource(") {
+				t.Error("due_sources calls collectSource — the filter must make no " +
+					"network request, or it costs the same as probing everything")
+			}
+		}
+	}
+}
+
+// TestQuotaMemoRecordGrowsTheStreakFromShards covers the recording half: a host
+// that failed inside a shard must end up in the SHARED memo with a streak, or the
+// next refresh has nothing to back off from.
+func TestQuotaMemoRecordGrowsTheStreakFromShards(t *testing.T) {
+	srv := newCountingServer(t)
+	ps := newQuotaPlugins(t, srv.URL)
+	host := hostOnly(srv.URL)
+
+	read := func() map[string]interface{} {
+		st, _ := ps.State("quota").(map[string]interface{})
+		memo, _ := st["probe_memo"].(map[string]interface{})
+		return memo
+	}
+
+	// A shard reports a host that answered with nothing.
+	for i := 1; i <= 3; i++ {
+		out, err := ps.Call("quota", "memo_record", map[string]interface{}{
+			"results": []interface{}{
+				map[string]interface{}{"host": srv.URL + "/v1"},
+			},
+		})
+		if err != nil {
+			t.Fatalf("memo_record %d: %v", i, err)
+		}
+		if out != true {
+			t.Fatalf("memo_record returned %#v", out)
+		}
+	}
+	memo := read()
+	e, ok := memo[host].(map[string]interface{})
+	if !ok {
+		t.Fatalf("memo has no entry for %s: %#v", host, memo)
+	}
+	if got, _ := e["streak"].(float64); got != 3 {
+		t.Errorf("streak after three shard failures = %.0f, want 3", got)
+	}
+
+	// A host that answers is recorded as a positive memo, which resets the streak.
+	if _, err := ps.Call("quota", "memo_record", map[string]interface{}{
+		"results": []interface{}{
+			map[string]interface{}{"host": srv.URL + "/v1", "vendor": "deepseek"},
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if got := read()[host]; got != "deepseek" {
+		t.Errorf("a host that answered is memoised as %#v, want the vendor name — "+
+			"the streak must not survive a success", got)
+	}
+}
+
+// TestQuotaDueSourcesSkipsBackedOffHosts is the behaviour the page depends on:
+// the filter must keep a backed-off host out of the sharded set.
+func TestQuotaDueSourcesSkipsBackedOffHosts(t *testing.T) {
+	srv := newCountingServer(t)
+	ps := newQuotaPlugins(t, srv.URL)
+	host := hostOnly(srv.URL)
+
+	// One failure, which is enough to put the host inside its window.
+	if _, err := ps.Call("quota", "memo_record", map[string]interface{}{
+		"results": []interface{}{map[string]interface{}{"host": srv.URL + "/v1"}},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	// A SECOND host, so the two cases are genuinely independent. The memo is keyed
+	// by host, so two paths on one host are one entry — using the same server here
+	// would have asserted that a per-host memo behaves per-path.
+	other := newCountingServer(t)
+	srcs := []interface{}{
+		map[string]interface{}{"name": "dead", "base_url": srv.URL + "/v1", "api_key": "k"},
+		map[string]interface{}{"name": "fresh", "base_url": other.URL + "/v1", "api_key": "k"},
+	}
+	out, err := ps.Call("quota", "due_sources", map[string]interface{}{"sources": srcs})
+	if err != nil {
+		t.Fatal(err)
+	}
+	m, ok := out.(map[string]interface{})
+	if !ok {
+		t.Fatalf("due_sources returned %#v", out)
+	}
+	due, _ := m["due"].([]interface{})
+	backing, _ := m["backing_off"].([]interface{})
+	if len(backing) != 1 {
+		t.Errorf("backing_off = %#v, want exactly the host that failed", backing)
+	} else if b, _ := backing[0].(map[string]interface{}); b["host"] != host {
+		t.Errorf("backing off %v, want %s — the key must be the host, or it will "+
+			"never match on the next refresh", b["host"], host)
+	}
+	if len(due) != 1 {
+		t.Errorf("due = %#v, want only the unprobed host", due)
+	} else if d, _ := due[0].(map[string]interface{}); d["name"] != "fresh" {
+		t.Errorf("due contains %v, want fresh", d["name"])
+	}
+
+	// And it must make no network request — the whole saving depends on it.
+	if n := srv.Hits(); n != 0 {
+		t.Errorf("due_sources made %d requests; it must decide without probing", n)
+	}
+}
+
+// hostOnly mirrors the plugin's own hostOf. Used only to CHECK what the plugin
+// wrote — seeding a memo under a different key produces a memo the plugin never
+// looks up, which cost this file a couple of rounds before it was noticed.
+func hostOnly(baseURL string) string {
+	u := strings.TrimPrefix(strings.TrimPrefix(baseURL, "https://"), "http://")
+	if i := strings.Index(u, "/"); i >= 0 {
+		u = u[:i]
+	}
+	return u
+}
+
+// TestQuotaBackoffRecordIsSkippedAndFlagged pins the shape the page depends on.
+//
+// Two things went wrong in the same field, and both were silent:
+//
+//  1. The back-off record had no `skipped` flag, so the page's
+//     `filter(r => !r.skipped)` let it through to the card renderer. That
+//     renderer matches on known fields, and a record carrying only a reason
+//     matches none — so it fell through to a default that said "有数据".
+//     Thirteen dead upstreams were rendered as though they had answered.
+//  2. Once it did carry `skipped`, the page grouped skipped sources by reading
+//     the REASON TEXT. Translating that string would have silently merged
+//     "backing off" into "exposes no balance endpoint" — a different claim
+//     about the upstream. The grouping keys on an explicit flag.
+func TestQuotaBackoffRecordIsSkippedAndFlagged(t *testing.T) {
+	srv := newCountingServer(t)
+	ps := newQuotaPlugins(t, srv.URL)
+
+	// One failure, enough to put the host inside its window.
+	if _, err := ps.Call("quota", "memo_record", map[string]interface{}{
+		"results": []interface{}{map[string]interface{}{"host": srv.URL + "/v1"}},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	out, err := ps.Call("quota", "due_sources", map[string]interface{}{
+		"sources": []interface{}{map[string]interface{}{
+			"name": "dead", "base_url": srv.URL + "/v1", "api_key": "k",
+		}},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	m, _ := out.(map[string]interface{})
+	backing, _ := m["backing_off"].([]interface{})
+	if len(backing) != 1 {
+		t.Fatalf("backing_off = %#v, want one record", m["backing_off"])
+	}
+	rec, _ := backing[0].(map[string]interface{})
+	if rec["skipped"] != true {
+		t.Errorf("back-off record has skipped=%v — the page filters cards on this "+
+			"field, and without it the record is rendered as a source with data",
+			rec["skipped"])
+	}
+	if rec["backing_off"] != true {
+		t.Errorf("back-off record has no explicit backing_off flag; the page groups " +
+			"by it instead of by reason text, so translating the message cannot " +
+			"merge this with 'no balance endpoint'")
+	}
+	if s, _ := rec["reason"].(string); s == "" {
+		t.Error("back-off record carries no reason; the card renders empty")
+	}
+}
+
+// TestQuotaRenderCallSitesPassRecordsNotNames guards a bug that only appeared
+// because ONE of two identical call sites was fixed.
+//
+// render() groups skipped sources by the backing_off flag. A call site that maps
+// the records to their names first ("skipped.map(r => r.name)") therefore hands
+// render an array of strings: the flag is gone, every entry falls into the
+// "exposes no balance endpoint" bucket, and the footer lists N unnamed sources.
+//
+// The first fix changed only the refresh path. The snapshot path kept mapping to
+// names, and the symptom was a footer reading "未列出 13 个源：、、、、…" — which
+// reads like a rendering bug rather than a contract mismatch between two call
+// sites of one function.
+func TestQuotaRenderCallSitesPassRecordsNotNames(t *testing.T) {
+	src := readShippedPlugin(t, "quota.lua")
+	if strings.Contains(src, ".map(function(r){ return r.name; })") ||
+		strings.Contains(src, ".map(function(r){return r.name;})") {
+		// A benign occurrence would be fine, but there is no other reason for a
+		// skipped list to be reduced to names, so treat any as the bug.
+		for _, line := range strings.Split(src, "\n") {
+			if strings.Contains(line, "skipped") && strings.Contains(line, ".map(") &&
+				strings.Contains(line, "r.name") {
+				t.Errorf("a skipped list is mapped to names before reaching render(): %s\n"+
+					"render() groups on the backing_off flag, which a name cannot carry",
+					strings.TrimSpace(line))
+			}
+		}
+	}
+	// Both call sites must exist and both must hand over records.
+	if n := strings.Count(src, ".filter(function(r){return r.skipped;})") +
+		strings.Count(src, ".filter(function(r){ return r && r.skipped; })"); n < 2 {
+		t.Errorf("found %d skipped-list filters, want 2 (snapshot and refresh paths)", n)
+	}
+}

packaging/upload-release.sh

@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+# Upload release assets to a GitCode release.
+#
+# The endpoint shapes are asymmetric and both differ from the "obvious" guess:
+#
+#   GET  /releases/tags/{tag}            → read a release
+#   GET  /releases/{tag}/upload_url      → get a presigned PUT   (NO "tags/")
+#   PUT  {presigned url}                 → upload, REPLAYING the headers verbatim
+#
+# Two mistakes cost real time on v1.9.5 before this script existed:
+#
+#   1. POSTing straight to /releases/{tag}/assets returns 404 — that endpoint
+#      does not exist. It reads as "authentication failed" if you only check
+#      the status code.
+#   2. PUTting without the headers fails with an opaque storage error, not a
+#      permission one, so it looks like a size limit or a corrupt file.
+#
+# The x-obs-* headers are part of the signature: dropping any one of them turns
+# the request into an unsigned one.
+set -uo pipefail
+
+TOKEN="$(tr -d '\n\r' < /tmp/gc_token)"
+REPO="JianFeeeee/ModelRouter"
+TAG="${1:?usage: upload-release.sh <tag> <file>...}"
+shift
+
+API="https://gitcode.com/api/v5/repos/${REPO}"
+
+for f in "$@"; do
+  name="$(basename "$f")"
+  size="$(stat -c%s "$f")"
+  printf '  %-46s %10s bytes  ' "$name" "$size"
+
+  # 1) presigned upload location. The file name must be URL-encoded; the v1.9.4
+  #    assets show "~" surviving in a stored name, so encoding is per-component.
+  enc="$(python3 -c 'import urllib.parse,sys; print(urllib.parse.quote(sys.argv[1]))' "$name")"
+  resp="$(curl -sS -m 60 -H "Authorization: Bearer ${TOKEN}" \
+    "${API}/releases/${TAG}/upload_url?file_name=${enc}" 2>&1)"
+  url="$(printf '%s' "$resp" | python3 -c 'import json,sys
+try:
+    d = json.load(sys.stdin)
+    print(d.get("url") or d.get("uploadUrl") or "")
+except Exception:
+    print("")' 2>/dev/null)"
+  if [ -z "$url" ]; then
+    echo "✗ 取得上传地址失败: ${resp:0:140}"
+    continue
+  fi
+
+  # 2) PUT with the headers replayed exactly as returned.
+  code="$(printf '%s' "$resp" | python3 -c 'import json,sys
+try:
+    print(json.dumps(json.load(sys.stdin).get("headers") or {}))
+except Exception:
+    print("{}")' 2>/dev/null | python3 -c '
+import json, subprocess, sys
+headers = json.load(sys.stdin)
+args = ["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}", "-X", "PUT"]
+for k, v in headers.items():
+    args += ["-H", f"{k}: {v}"]
+args += ["--data-binary", "@" + sys.argv[1], sys.argv[2]]
+print(subprocess.run(args, capture_output=True, text=True).stdout)
+' "$f" "$url" 2>/dev/null)"
+
+  if [ "$code" = "200" ] || [ "$code" = "201" ] || [ "$code" = "204" ]; then
+    echo "✓ HTTP=$code"
+  else
+    echo "✗ HTTP=$code"
+  fi
+  sleep 1
+done