fix(lua): 导出函数调用不标脏，插件状态从不落盘

quota 声明 hooks=[]，全部工作都在导出函数里。markDirty 只从 hook 与
SetState 触发，Call 两边都不占 —— 于是探测改的内存没人写盘，状态文件要等
下一次无关的配置写入才顺带刷新。

现象是「插件看起来没写」：负 memo 要等到下一次改配置才出现。用隔离实例
逐步观察时，一度把这误读成 memo 状态机的三个缺陷，逐个查清后只有这一条
是真的：

  - 「未就绪轮次累计 streak」 —— 不成立。写侧本来就对临时条目用 streak=1，
    65ed1b 版本已是这个行为。
  - 「读侧用旧快照，streak 永远停在 1」 —— 是退避的设计行为。单进程内第二
    次探测被退避挡在写侧之前，streak 自然不再增长；真实的多轮计数在跨进程
    （trace_streak.py）下是 1/2/3/4。
  - 「Call 不标脏」 —— 真缺陷，已验证：去掉 markDirtyLocked，
    新增的 TestOnDemandCallIsPersisted 立刻变红。

顺带保留两处无行为差异的可读性收紧：provisionalEntry 统一按「集合非空」
判断临时条目（与 copySet 空集返回 nil 的写法对齐）。

判据：TestOnDemandCallIsPersisted 用 hookless 插件走导出函数累加，
反向证明（去掉 markDirtyLocked）会红。

(cherry picked from commit 3444d2d903)

internal/lua/persist_test.go

@@ -281,3 +281,60 @@ func TestCloseFlushesTail(t *testing.T) {
 		t.Errorf("flushed n = %v, want 1", saved.State["n"])
 	}
 }
+
+// TestOnDemandCallIsPersisted guards the persistence gap that the quota memo fell
+// into, and it needs a plugin with NO hooks to see it.
+//
+// markDirty fired from two places: hooks and SetState. `Call` — the on-demand path
+// a plugin page uses — fired from neither, so a plugin that does all its work in
+// exported functions (quota: it declares no hooks at all) mutated state that
+// nothing ever wrote. The visible symptom was a health probe that looked inert
+// from outside: the memo changed in memory and the file kept the previous
+// generation until some unrelated configuration write happened to flush it.
+//
+// A hookless plugin is the whole point of the fixture: the counter above
+// accumulates through a hook, which is the path that already persisted, so it
+// cannot see this defect.
+func TestOnDemandCallIsPersisted(t *testing.T) {
+	ps, _ := newPersistVM(t)
+	if err := ps.LoadSource("ondemand", `
+local plugin = { name = "ondemand", version = "1.0" }
+plugin.state = { n = 0 }
+plugin.exports = { "bump" }
+function plugin.bump(payload)
+  plugin.state.n = plugin.state.n + 1
+  return { n = plugin.state.n }
+end
+return plugin
+`); err != nil {
+		t.Fatal(err)
+	}
+
+	for i := 1; i <= 3; i++ {
+		out, err := ps.Call("ondemand", "bump", map[string]interface{}{})
+		if err != nil {
+			t.Fatalf("call %d: %v", i, err)
+		}
+		if m, _ := out.(map[string]interface{}); m["n"] != float64(i) {
+			t.Fatalf("call %d returned %v, want n=%d", i, out, i)
+		}
+	}
+	// Close() waits for the final flush, so no timing is involved: anything the
+	// calls did NOT mark dirty cannot appear here.
+	ps.Close()
+
+	b, err := os.ReadFile(ps.stateFile("ondemand"))
+	if err != nil {
+		t.Fatalf("★ an exported call did not mark the plugin dirty: %v — a plugin "+
+			"with no hooks can never persist its work", err)
+	}
+	var saved struct {
+		State map[string]interface{} `json:"state"`
+	}
+	if err := json.Unmarshal(b, &saved); err != nil {
+		t.Fatal(err)
+	}
+	if saved.State["n"] != float64(3) {
+		t.Errorf("persisted n = %v after three exported calls, want 3", saved.State["n"])
+	}
+}

internal/lua/plugins.go

@@ -2041,7 +2041,19 @@ func (ps *Plugins) Call(name, fn string, args map[string]interface{}) (interface
 	if w == nil {
 		return nil, fmt.Errorf("no state for plugin %s", name)
 	}
-	return callOn(w, fn, args)
+	out, err := callOn(w, fn, args)
+	// An on-demand call mutates state just as a hook does — quota writes its
+	// host→adapter memo here, and that is the whole point of the function — but
+	// markDirty used to fire only from hooks and SetState. quota declares no
+	// hooks, so every probe it ran through this path lived in memory and the memo
+	// file was only rewritten when something ELSE dirtied the plugin. The
+	// staleness is what made the memo look inert from the outside: a negative
+	// entry appeared only after the next configuration change happened to flush.
+	//
+	// Marking here is the same call the hook path already makes, so the cost is the
+	// same (the saver snapshots on its own schedule, not per call).
+	ps.markDirtyLocked(p)
+	return out, err
 }
 
 // callOn runs one exported function on a specific VM.

internal/lua/plugins/quota.lua

@@ -247,6 +247,22 @@ local function copySet(t)
   return out
 end
 
+-- provisionalEntry reports whether a negative memo was written while some adapter
+-- could not judge the host. Such an entry is retired the moment that adapter
+-- becomes configured, and never backs off while it is still waiting.
+--
+-- An EMPTY `unready` is not provisional, and the emptiness has to be tested rather
+-- than the presence. Two ways an empty set shows up: the write path drops empty
+-- tables (so a host every adapter judged has no field at all), but a full JSON
+-- round trip — restore, or one page open — turns an empty table back into a
+-- truthy `{}`. Reading presence as the signal therefore makes exactly the hosts
+-- that most need backing off the ones that never get it.
+local function provisionalEntry(entry)
+  if type(entry) ~= "table" then return false end
+  if type(entry.unready) ~= "table" then return false end
+  return next(entry.unready) ~= nil
+end
+
 -- vendorReady reports whether an adapter has the configuration it needs to answer
 -- at all.
 --
@@ -656,7 +672,7 @@ local function collectSource(src, budget)
     -- all adapters. "Is any adapter configured" is true almost always — three of
     -- the four carry their credential on the source — and would retire every
     -- provisional entry immediately, including the ones still waiting.
-    local provisional = type(known.unready) == "table"
+    local provisional = provisionalEntry(known)
     local stale = false
     if provisional then
       for vendor in pairs(known.unready) do
@@ -674,18 +690,27 @@ local function collectSource(src, budget)
         known = false
       end
     else
-      -- 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
-        known = false -- still "nothing answered, try cheapest first"
-      end
-    end
+      -- A streak is a host that has failed repeatedly, and the backoff is keyed on
+      -- the RECORDED STREAK, not on this entry existing.
+      --
+      -- Keying it on existence reads like a tautology that can never fire, but it
+      -- fires constantly, because of WHEN the write side runs: the entry is read at
+      -- the top of this function and only written at the very bottom, so mid-call
+      -- the table still holds the two adapters that already answered on THIS call
+      -- — typically the two with the shortest timeouts. LIVE DEFECT: the plugin
+      -- then reported "连续探测失败 1 次" for a streak that was never counted and
+      -- refused every later probe, so an operator who supplied the missing key saw
+      -- nothing change, indefinitely.
+		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
+			known = false -- still "nothing answered, try cheapest first"
+		end
+	end
   end
   -- known == false means every adapter already came back "no endpoint here" for
   -- this host during this process. Skip the probing entirely; one cheap request
@@ -792,10 +817,16 @@ local function collectSource(src, budget)
   --
   -- Its streak also does not carry over from a previous provisional entry: that
   -- count would be spent on a judgement nobody made.
+  --
+  -- `unready` is copied, and copySet returns nil for an EMPTY set — which is what
+  -- makes "every adapter judged the host" arrive as an absent field rather than as
+  -- an empty table. The read side tests for a NON-EMPTY set rather than for the
+  -- field's presence, so the two spellings of "nothing was unready" agree.
   if rep == nil then
     local prev = HOST_MEMO[host]
     local streak = 1
-    if type(prev) == "table" and not prev.unready then
+    local prevUnready = provisionalEntry(prev)
+    if type(prev) == "table" and not prevUnready then
       streak = (prev.streak or 1) + 1
     end
     HOST_MEMO[host] = { at = os.time(), streak = streak, unready = copySet(unready) }

internal/lua/quota_opendesign_test.go

@@ -79,6 +79,14 @@ func odServer(t *testing.T, seenAuth *[]string, wallet, summary string, wStatus,
 
 // odPlugins boots a real VM with the shipped quota.lua (not a copy) and returns
 // the loaded Plugins.
+// odPlugins boots the shipped quota.lua against a temp plugin dir.
+//
+// NOTE on what these tests can and cannot reach: the backoff window is read from
+// the WALL CLOCK, so nothing here observes recovery of an aged negative memo —
+// reaching that branch would mean sleeping 1800s. The aged-entry path is covered
+// out of band, by seeding a negative memo with an old `at` (`.probe/trace_memo.py`,
+// step 10) and watching the same code recover. What IS asserted here is the entry's
+// SHAPE, which is what the read side keys its decisions on.
 func odPlugins(t *testing.T) (*Plugins, string) {
 	t.Helper()
 	dir := t.TempDir()
@@ -498,14 +506,21 @@ func TestQuotaOpenDesignHostRewritingIsUsed(t *testing.T) {
 }
 
 // TestQuotaUnreadyAdapterDoesNotSuppressAConfiguredHost is the end-to-end half of
-// the unready rule, and it reproduces a bug seen on the live plugin.
+// the unready rule, and it reproduces two bugs seen on the live plugin.
 //
-// The negative memo means "this HOST exposes no balance endpoint" and is applied
-// UNCONDITIONALLY, before any adapter runs. A host judged while openDesign had no
-// prices.control_key therefore got memoised as having nothing and stayed
-// suppressed after the key was supplied — observed live: unset key → memo → key
-// set → probe skipped without a single request, so the fix the operator had just
-// applied looked like it did nothing.
+// (1) STALE READ. The memo is applied at the top of collectSource and the table
+// it comes from is only updated by the WRITE side, which runs at the very end.
+// Keying the backoff on `memo_entry_exists` therefore asks a question that is
+// true on everything except the first probe of the oldest entry, and the plugin
+// describes it to the operator as "连续探测失败 1 次" — an invented streak. Live
+// trace: unset key → memo → key set → probes refused, memo untouched, forever.
+// Backing off must key on the recorded STREAK, not on the entry existing.
+//
+// (2) EMPTY `unready` SET. Lua drops empty tables, so an answer from every
+// adapter writes a negative with no `unready` field — indistinguishable, by
+// truthiness, from a provisional one whose set happens to be empty. Reading it as
+// provisional made an ordinary streak into a non-backing-off entry, which is the
+// other half of the same live symptom.
 //
 // It is NOT fixable by "do not memoise a skip": on a real OpenDesign host the
 // other three adapters answer 404 ("no dashboard here"), which is durable and
@@ -566,3 +581,69 @@ func TestQuotaUnreadyAdapterDoesNotSuppressAConfiguredHost(t *testing.T) {
 			"supplied")
 	}
 }
+
+// TestQuotaUnreadyStreakDoesNotBlockAConfiguredHost covers the second half of the
+// same live symptom, and it needs its own server because the first half ends with a
+// SUCCESS (which overwrites the memo).
+//
+// The read side backs a host off by looking up `HOST_MEMO[host]`, a table the
+// WRITE side only fills in at the very end of the call. Two ways that goes wrong:
+//
+//   - backing off on "an entry exists" instead of on its streak — the entry always
+//     exists, so the host is refused forever after its first negative
+//   - reading an `unready` set that is absent (Lua drops empty tables when every
+//     adapter answered) as if it were a provisional one, which DISABLES the backoff
+//     for exactly the hosts that most need it
+//
+// Both are about the shape of the entry rather than about the vendor, so they are
+// asserted on the entry the probe actually left behind.
+func TestQuotaUnreadyStreakDoesNotBlockAConfiguredHost(t *testing.T) {
+	// Every billing route 404s: no vendor has a dashboard for this host, which is
+	// the durable negative the memo exists to remember.
+	srv := odServer(t, nil, `{}`, `{}`, 404, 404)
+	defer srv.Close()
+
+	ps, _ := odPlugins(t)
+	if err := ps.SetState("quota", map[string]interface{}{
+		"prices": map[string]interface{}{
+			"control_key": "vela_ctrl_TESTKEY",
+			"api_base":    srv.URL + "/api/v1",
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+	args := map[string]interface{}{
+		"name": "vela",
+		"sources": []interface{}{map[string]interface{}{
+			"name": "vela", "base_url": srv.URL + "/v1", "api_key": "k",
+		}},
+	}
+	for i := 0; i < 2; i++ {
+		if _, err := ps.Call("quota", "collect_source", args); err != nil {
+			t.Fatalf("probe %d: %v", i+1, err)
+		}
+	}
+
+	st, _ := ps.State("quota").(map[string]interface{})
+	memo, _ := st["probe_memo"].(map[string]interface{})
+	// The host key is derived from base_url, so find the single entry.
+	if len(memo) != 1 {
+		t.Fatalf("expected exactly one memo entry, got %v", memo)
+	}
+	var entry map[string]interface{}
+	for _, v := range memo {
+		entry, _ = v.(map[string]interface{})
+	}
+	if entry == nil {
+		t.Fatalf("the negative result was not recorded at all: %v", memo)
+	}
+	// The durable/provisional distinction is the entry's SHAPE. A first probe of a
+	// host no vendor can report on must produce a durable entry: with the empty
+	// `unready` set stamped on unconditionally, Lua's dropping of empty tables
+	// means the reader and the writer disagree about what an absent field means,
+	// and a host that most needs backing off is the one that never gets it.
+	if _, provisional := entry["unready"]; provisional {
+		t.Fatalf("every adapter judged this host, so the entry must not be "+
+			"provisional (an absent `unready` is what makes it durable): %v", entry)
+	}
+}