ModelRouter/internal/lua/plugins/quota.lua

-- Provider quota — plan balance across upstreams, one card per source.
--
-- This started as a Command Code plugin and grew, because only one vendor was
-- being asked. The shape that works across providers is a two-part contract:
--
--   1. A small set of PROBE ADAPTERS, each naming an endpoint, how to call it
--      and which JSON paths hold the numbers. Adapters are tried in order and
--      the first one that returns a usable answer wins.
--   2. A DECLARATIVE override, so a provider nobody anticipated can be added
--      from the plugin's own state (endpoints + path mapping) without touching
--      code.
--
-- What the adapters share, and why it is worth generalizing:
--
--   - The DeepSeek family (deepseek, and the many re-sells that copied its
--     dashboard) answers GET /user/balance with
--     {is_available, balance_infos:[{currency,total_balance,granted,topped_up}]}.
--     Granted vs topped-up matters: gifted credit expiring on a schedule is a
--     different operational fact from prepaid money.
--   - The OpenAI-compatible dashboard shape answers
--     /v1/dashboard/billing/subscription plus /v1/dashboard/billing/usage,
--     which is what most re-sells of an OpenAI console expose. Note these
--     report LIMITS and SPEND, not a balance: the provider here has a hard limit
--     of 1e8 and 660 spent, so "remaining" must not be computed as
--     limit - spend and presented as money left — it is a ceiling nobody is
--     near. Reporting that honestly matters more than a green bar.
--   - Command Code exposes rolling 5-hour / weekly / monthly windows plus a
--     billing period, which no other provider here does.
--
-- A provider with no working adapter is SKIPPED, silently. Half these upstreams
-- are small re-sells whose dashboard endpoints return an HTML page, not JSON; a
-- list full of "unknown" rows trains the operator to ignore the page entirely,
-- which is worse than showing only what is real. Failures that are NOT "no such
-- endpoint" — a 401 on a key we hold, say — are surfaced, because those are the
-- ones worth fixing.
--
-- Only hosts of CONFIGURED SOURCES are reachable (__llmsproxy_http enforces
-- that), so adding a probe never widens what the gateway can talk to.

local PLUGIN = {}

PLUGIN.name = "quota"
PLUGIN.version = "1.0.0"
PLUGIN.description = "多源套餐余量（余额 / 用量 / 滚动窗口）"

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
-- no saved state still has an empty table to write into, rather than nil.
--
-- It is initialised at the TOP on purpose. Every collect path below touches it
-- (to restore the probe memo on first use, to persist it afterwards), and a
-- declaration further down would be nil for the reads that run first.
PLUGIN.state = {}

-- ---------------------------------------------------------------- helpers

local function tonum(v)
  return tonumber(v) or 0
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.match(u, "^https?://([^/]+)")) or u
end

-- get walks a dotted path through nested tables, returning nil if any hop is
-- missing. A missing field must stay nil rather than becoming 0: "this provider
-- did not report it" and "this provider reported zero" are different, and
-- collapsing them shows a healthy $0.00 for a provider that never answered.
local function get(root, path)
  if root == nil then return nil end
  local cur = root
  for seg in string.gmatch(path, "[^%.]+") do
    if type(cur) ~= "table" then return nil end
    cur = cur[seg]
    if cur == nil then return nil end
  end
  return cur
end

local function getnum(root, path)
  local v = get(root, path)
  if v == nil then return nil end
  if type(v) == "boolean" then return v and 1 or 0 end
  local n = tonumber(v)
  return n
end

-- request performs one GET with the standard retry. Retrying is not
-- politeness here: several of these hosts sit behind DNS that intermittently
-- refuses, and a single 000 turned into "provider unknown" on every poll.
local function request(path, key, headers, tries)
  tries = tries or 2
  local hdrs = {}
  for k, v in pairs(headers or {}) do hdrs[k] = v end
  hdrs["Authorization"] = "Bearer " .. tostring(key or "")
  local lastErr = "no attempt"
  for _ = 1, tries do
    local res, err = __llmsproxy_http({
      -- 4s, not 10s: probing is network-bound and the whole page shares ONE
      -- Lua state under a write lock, so a slow upstream is not just a slow card
      -- — it stalls every request_end hook (i.e. billing) behind it.
      url = path, method = "GET", headers = hdrs, timeout_ms = 4000,
    })
    if res ~= nil then
      if res.status >= 200 and res.status < 300 then
        local ok, decoded = pcall(json.decode, res.body or "")
        if ok and type(decoded) == "table" then
          return decoded, nil
        end
        -- A 200 that is not JSON means we hit a marketing page, not an API.
        return nil, "not json (HTTP " .. tostring(res.status) .. ")"
      end
      if res.status == 401 or res.status == 403 then
        -- Retrying a rejected credential only wastes the operator's time.
        return nil, "HTTP " .. tostring(res.status) .. " (key rejected)"
      end
      lastErr = "HTTP " .. tostring(res.status)
    else
      lastErr = tostring(err or "request failed")
    end
  end
  return nil, lastErr
end

-- Monthly credit caps per planId. The credits endpoint never reports a cap, so
-- this is a lookup against the vendor's published pricing ($10 Go, $70
-- GOAT/Pro/Max), cross-checked against monthly_used + balance. Declared before
-- the adapters that read it — a Lua local is nil until its assignment runs, and
-- the Command Code probe hit that as a runtime nil-index.
local PLAN_CAPS = {
  ["individual-go"]   = 10,
  ["individual-goat"] = 70,
  ["individual-pro"]  = 70,
  ["individual-max"]  = 70,
  ["team-goat"]       = 70,
}

-- ---------------------------------------------------------------- adapters

-- Each adapter returns (report, nil) when it produced something usable, or
-- (nil, reason) to let the next one try. `skip` is the quiet reason — a 404 or
-- an HTML page means "this provider does not expose balance here", which is not
-- worth showing. Everything else becomes a visible failure.
local function deepseekBalance(base, key)
  local r, err = request(base .. "/user/balance", key)
  if r == nil then
    if err and (err:match("not json") or err:match("HTTP 404")) then
      return nil, "skip", err
    end
    return nil, "error", err
  end
  local infos = get(r, "balance_infos")
  if type(infos) ~= "table" or #infos == 0 then return nil, "skip", "no balance_infos" end
  local first = infos[1]
  local rep = {
    vendor = "deepseek-style",
    currency = tostring(first.currency or "USD"),
    balance = tonum(first.total_balance),
    granted = getnum(first, "granted_balance"),
    topped_up = getnum(first, "topped_up_balance"),
    available = get(r, "is_available") == true,
  }
  return rep, nil
end

-- openaiDashboard reports LIMITS and SPEND. There is deliberately no
-- `remaining` here: hard_limit_usd is a ceiling (1e8 on the live provider), and
-- ceiling - spend would read as "$99,999,339 left", which is a number nobody
-- should ever believe.
local function openaiDashboard(base, key)
  local sub, subErr = request(base .. "/v1/dashboard/billing/subscription", key, nil, 1)
  if sub == nil then
    if subErr and (subErr:match("not json") or subErr:match("HTTP 404")) then
      return nil, "skip", subErr
    end
  end
  local use, useErr = request(base .. "/v1/dashboard/billing/usage", key, nil, 1)
  if sub == nil and use == nil then
    if subErr and subErr:match("key rejected") then
      return nil, "error", subErr
    end
    return nil, "skip", (subErr or useErr or "no dashboard")
  end
  local rep = {
    vendor = "openai-dashboard",
    hard_limit = getnum(sub or {}, "hard_limit_usd"),
    soft_limit = getnum(sub or {}, "soft_limit_usd"),
    total_usage = getnum(use or {}, "total_usage"),
    has_payment_method = get(sub or {}, "has_payment_method") == true,
  }
  -- Only call it a balance when the vendor says so. A hard limit with no usage
  -- endpoint is a ceiling with nothing spent against it; that still renders, but
  -- as a limit rather than as money left.
  rep.balance = nil
  if rep.hard_limit and rep.hard_limit < 1000000 and rep.total_usage then
    rep.balance = rep.hard_limit - rep.total_usage
    rep.cap = rep.hard_limit
    rep.used = rep.total_usage
  end
  return rep, nil
end

-- commandCode keeps its own shape: rolling windows plus a billing period.
local function commandCode(base, key)
  local H = {
    ["x-command-code-version"] = "1.54.2",
    ["x-cli-environment"] = "production",
    ["User-Agent"] = "quota-plugin/1.0.0",
  }
  local origin = base:gsub("/provider/v1/?$", "")
  local credits, err = request(origin .. "/alpha/billing/credits", key, H)
  if credits == nil then
    if err and err:match("key rejected") then return nil, "error", err end
    return nil, "skip", err
  end
  local rep = {
    vendor = "commandcode",
    currency = "USD",
    balance = tonum(get(credits, "credits.monthlyCredits"))
      + tonum(get(credits, "credits.purchasedCredits"))
      + tonum(get(credits, "credits.freeCredits")),
    below_threshold = get(credits, "credits.belowThreshold") == true,
    limited = get(credits, "windowLimits.limited") == true,
    exceeded = get(credits, "windowLimits.exceeded"),
  }
  local fh = get(credits, "windowLimits.fiveHour")
  if type(fh) == "table" then
    rep.windows = rep.windows or {}
    rep.windows[#rep.windows + 1] = {
      label = "5 小时", used = tonum(fh.used), cap = tonum(fh.cap),
      exceeded = fh.exceeded == true, reset_at = tonum(fh.resetAt),
    }
  end
  local wk = get(credits, "windowLimits.weekly")
  if type(wk) == "table" then
    rep.windows = rep.windows or {}
    rep.windows[#rep.windows + 1] = {
      label = "每周", used = tonum(wk.used), cap = tonum(wk.cap),
      exceeded = wk.exceeded == true, reset_at = tonum(wk.resetAt),
    }
  end

  -- Subscriptions and usage are enrichment: their absence must not blank the
  -- balance that just came back.
  local subs = request(origin .. "/alpha/billing/subscriptions", key, H, 1)
  if subs ~= nil then
    rep.plan_id = get(subs, "data.planId")
    rep.plan_status = get(subs, "data.status")
    rep.period_end = get(subs, "data.currentPeriodEnd")
    rep.plan_cap = PLAN_CAPS[tostring(rep.plan_id or "")]
    rep.cancel_at_period_end = get(subs, "data.cancelAtPeriodEnd") == true
  end
  local usage = request(origin .. "/alpha/usage/summary", key, H, 1)
  if usage ~= nil then
    rep.monthly_used = getnum(usage, "totalMonthlyCredits")
    rep.requests = getnum(usage, "totalCount")
    rep.tokens = getnum(usage, "totalTokens")
    rep.success_rate = getnum(usage, "successRate")
  end
  if rep.plan_cap and rep.monthly_used and rep.monthly_used > 0 then
    local implied = rep.monthly_used + rep.balance
    if math.abs(implied - rep.plan_cap) > (rep.plan_cap * 0.02 + 0.5) then
      rep.cap_mismatch = true
      rep.implied_cap = implied
    end
  end
  return rep, nil
end

-- Ordered probes. DeepSeek-family first because it is a single cheap endpoint,
-- then the OpenAI console shape, then Command Code's four.
local ADAPTERS = {
  { vendor = "commandcode",    probe = commandCode },
  { vendor = "deepseek",       probe = deepseekBalance },
  { vendor = "openai-dash",    probe = openaiDashboard },
}

-- The probe memo remembers, per host, which adapter answered. Persisted in the
-- plugin's state so the expensive part — discovering that 10 of 21 upstreams
-- expose no balance endpoint at all — is paid once, not on every page open.
--
-- This matters more than it looks. The answers depend only on the host, never
-- on the key, and re-discovering them cost 21 sources x 3 adapters x 2 attempts
-- of timeouts on the first load.

-- HOST_MEMO maps host -> adapter vendor (or false when nothing answered). It
-- is restored from the plugin's own state on the first collect of a process, so
-- a restart does not re-pay the cost of discovering which upstreams are silent.
local HOST_MEMO = {}
-- HOST_MEMO_AT[host] = when that host last produced data. A negative memo older
-- than PROBE_MEMO_TTL is ignored, so a vendor that adds an endpoint is picked up
-- within the TTL without a restart.
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.
local PROBE_BUDGET_MS = 12000

-- ---------------------------------------------------------------- source scan

-- collectSource probes one source and returns a report, or nil when the source
-- simply has no balance endpoint (silent) — the caller distinguishes the two
-- through `skipped`.
local function collectSource(src, budget)
  local base = tostring(src.base_url or "")
  if base == "" then return nil end
  base = base:gsub("/+$", "")
  local key = tostring(src.api_key or "")
  local rep, kind, why
  -- Try the adapter this host answered with last time first. Same host with a
  -- different key (two Command Code seats) gives the same answers, so this turns
  -- 21 sources into a handful of distinct round trips after the first pass.
  local host = hostOf(base)
  local known = HOST_MEMO[host]
  -- A negative memo is a TABLE ({at=timestamp}), distinguishable from a vendor
  -- string. JSON round-trips it fine; a bare false would not survive the trip
  -- through the Go-side state as cleanly, and the two being different types is
  -- 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,
                 reason = "backing off after " .. tostring(known.streak) .. " failed probes" }
      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
  -- known == false means every adapter already came back "no endpoint here" for
  -- this host during this process. Skip the probing entirely; one cheap request
  -- confirms it is still the case, so a vendor that adds a dashboard later is
  -- picked up without a restart.
  -- A cheap liveness probe, NOT a skip: the hosts behind these adapters are
  -- slow (12s+ to answer a root request), so treating a slow answer as "down"
  -- lost five sources that do have real data. The probe only short-circuits the
  -- genuinely unreachable case; a reachable host still goes through the
  -- adapters.
  if known == false then
    local probe = __llmsproxy_http({
      url = base .. "/", method = "GET", timeout_ms = 2500,
      headers = { ["Authorization"] = "Bearer " .. tostring(key or "") },
    })
    if probe == nil then
      return { name = src.name, host = host, skipped = true,
               reason = "host unreachable" }
    end
    -- Reachable: fall through and try the adapters again, since a slow host
    -- that was merely timing out may answer fine.
  end
  local order = ADAPTERS
  if known then
    local reordered, seen = {}, {}
    for _, a in ipairs(ADAPTERS) do
      if a.vendor == known then
        reordered[#reordered + 1] = a
        seen[a.vendor] = true
      end
    end
    for _, a in ipairs(ADAPTERS) do
      if not seen[a.vendor] then reordered[#reordered + 1] = a end
    end
    order = reordered
  end
  for _, a in ipairs(order) do
    -- Budget check BETWEEN adapters, so an exhausted budget still returns the
    -- sources already answered instead of nothing at all.
    if budget and budget.exhausted() then
      return { name = src.name, host = host, skipped = true,
               reason = "probe budget exhausted" }
    end
    local r, k, e = a.probe(base, key)
    if r ~= nil then
      rep = r
      HOST_MEMO[host] = a.vendor
      break
    end
    kind, why = k, e
  end
  -- Remember the NEGATIVE result too. Only recording successes left every
  -- source that has no balance endpoint re-probing all three adapters (2
  -- attempts, 4s each) on every single page open — measured at 60s per source,
  -- unchanged between runs, because "nothing answered" was never recorded.
  --
  -- The value is the adapter that came closest (an error rather than a skip), so
  -- a host that later starts working is still retried first.
  -- Remember that this host answered nothing, but only as an ORDERING hint, and
  -- with a stamp so it expires. "This upstream has no balance endpoint" is a
  -- fact that can change — vendors ship dashboards — so a permanent negative
  -- 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.
  -- 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()
  end
  if rep == nil then
    if kind == "error" then
      return { name = src.name, vendor = "", error = tostring(why),
               skipped = false, host = hostOf(base) }
    end
    return { name = src.name, skipped = true, host = hostOf(base) }
  end
  rep.name = src.name
  rep.host = hostOf(base)
  rep.skipped = false
  rep.checked_at = os.time()
  return rep
end

-- collect_all walks every source. Prefer it for a SMALL number of sources.
--
-- The page calls collect_source concurrently instead: probing is almost entirely
-- network wait (three adapters, up to two attempts, ten second timeouts each),
-- 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
        backing[#backing + 1] = { name = s.name, host = hostOf(base),
                                  reason = "backing off after " .. tostring(known.streak) .. " failed probes" }
      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
    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
  -- 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.time() - t1) > lim1
  end
  local budget = { exhausted = spent }
  local out = {}
  for _, s in ipairs(srcs) do
    if type(s) == "table" then
      local rep = collectSource(s, budget)
      if rep ~= nil then out[#out + 1] = rep end
    end
  end
  PLUGIN.state.probe_memo = HOST_MEMO
  return out, nil
end

-- collect_source probes one named source.
PLUGIN.collect_source = 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
  -- A closure over the deadline, not a table method. `exhausted = function(self)`
  -- 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.time()
  local limit = PROBE_BUDGET_MS / 1000
  local function exhausted()
    return (os.time() - t0) > limit
  end
  local budget = { exhausted = exhausted }
  for _, s in ipairs(srcs) do
    if type(s) == "table" and s.name == payload.name then
      local rep = collectSource(s, budget)
      -- Persist the host→adapter map so the next PROCESS skips the dead ends,
      -- not just the next page open.
      PLUGIN.state.probe_memo = HOST_MEMO
      return rep, nil
    end
  end
  return nil, "source not found: " .. tostring(payload.name)
end

-- ---------------------------------------------------------------- page

PLUGIN.ui = {
  -- Every row names an upstream host and what it has spent. That is deployment
  -- information, not per-caller information, so it is admin-only by default; an
  -- operator who wants a shared read-only view can widen it from config without
  -- editing this file.
  visibility = "admin",
  page = {
    page_id = "quota",
    title = "Quota",
    order = 45,
    mount = [==[
<div id="qa-root" class="qa-root">
  <div class="qa-head">
    <div>
      <div class="qa-title">套餐余量</div>
      <div class="qa-sub" id="qa-sub">点击刷新</div>
    </div>
    <div class="grow"></div>
    <button class="small" id="qa-refresh">刷新</button>
  </div>
  <div id="qa-list"><div class="qa-empty">尚未拉取</div></div>
</div>
<style>
#qa-root{padding:16px;min-width:0;max-width:100%;overflow-x:auto}
.qa-head{display:flex;align-items:flex-start;gap:10px;margin:0 0 12px;flex-wrap:wrap}
.qa-title{font-size:15px;font-weight:600}
.qa-sub{font-size:12px;opacity:.66;margin-top:2px}
.qa-acc{border:1px solid var(--bd,rgba(127,127,127,.25));border-radius:10px;padding:12px 14px;margin:0 0 10px;min-width:0}
.qa-acc.err{border-color:rgba(220,80,80,.5)}
.qa-h{display:flex;align-items:center;gap:8px;flex-wrap:wrap;margin:0 0 4px}
.qa-n{font-weight:600;font-size:14px;overflow:hidden;text-overflow:ellipsis;white-space:nowrap;max-width:46%}
.qa-host{font-size:11.5px;opacity:.55;overflow:hidden;text-overflow:ellipsis;white-space:nowrap}
.qa-vendor{font-size:11px;padding:1px 7px;border-radius:999px;border:1px solid rgba(127,127,127,.35);opacity:.8;white-space:nowrap}
.qa-pill{font-size:11px;padding:1px 7px;border-radius:999px;border:1px solid rgba(127,127,127,.35);opacity:.85;white-space:nowrap}
.qa-pill.ok{color:#15803d;border-color:rgba(21,128,61,.45)}
.qa-pill.low{color:#b45309;border-color:rgba(180,83,9,.45)}
.qa-pill.out{color:#b91c1c;border-color:rgba(185,28,28,.45)}
.qa-bal{display:flex;align-items:baseline;gap:8px;margin:6px 0 8px;flex-wrap:wrap}
.qa-bal-n{font-size:19px;font-weight:600}
.qa-bal-l{font-size:12px;opacity:.66}
.qa-grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(170px,1fr));gap:9px 18px}
.qa-w{min-width:0}
.qa-w-h{display:flex;justify-content:space-between;gap:6px;font-size:12px;opacity:.85}
.qa-w-n{font-variant-numeric:tabular-nums;font-weight:600}
.qa-bar{height:6px;border-radius:3px;background:rgba(127,127,127,.22);overflow:hidden;margin:4px 0 2px}
.qa-fill{height:100%;background:#3b82f6;border-radius:3px}
.qa-fill.warn{background:#d97706}
.qa-fill.over{background:#dc2626}
.qa-w-d{font-size:11px;opacity:.6;font-variant-numeric:tabular-nums}
.qa-meta{display:flex;gap:13px;flex-wrap:wrap;font-size:11.5px;opacity:.7;margin-top:9px}
.qa-fail{font-size:11.5px;color:#b91c1c;margin-top:7px;word-break:break-word}
.qa-warnl{font-size:11px;color:#b45309;margin-top:3px}
.qa-empty{opacity:.6;font-size:13px;padding:8px 0}
.qa-skip{font-size:11.5px;opacity:.5;margin-top:8px;line-height:1.6}
</style>
<script>
(function(){
  var MONTHLY_CAP_NOTE = "月度上限按套餐表推导";
  function escH(s){return String(s==null?"":s).replace(/[&<>"]/g,function(c){
    return c==="&"?"&amp;":c==="<"?"&lt;":c===">"?"&gt;":"&quot;";});}
  // Money is formatted with the vendor's own currency; CNY and USD differ only
  // in symbol here, which is deliberate — guessing an exchange rate would be a
  // far worse lie than showing the raw number with its unit.
  function money(v, cur){
    cur = cur || "USD";
    var sym = cur === "CNY" ? "¥" : (cur === "USD" ? "$" : cur + " ");
    v = Number(v)||0;
    // A negative credit is a real debit, not "zero" — deepseek reports
    // topped_up_balance -0.02 against a grant, and flattening that to 0.00
    // hides the overdraft.
    if (v===0) return sym+"0.00";
    if (v<0) return "-"+sym+Math.abs(v).toFixed(2);
    return sym+(v<0.01?v.toFixed(4):v.toFixed(2));
  }
  function pct(u,c){ c=Number(c)||0; u=Number(u)||0; if(!(c>0)) return null;
    return Math.floor(u/c*100+0.5); }
  function resetTxt(ms){
    if(!(Number(ms)>0)) return "";
    var s=Math.floor(Number(ms)/1000-Date.now()/1000); if(s<=0) return "";
    var d=Math.floor(s/86400),h=Math.floor((s%86400)/3600),m=Math.floor((s%3600)/60);
    if(d>0) return d+"d "+h+"h"; if(h>0) return h+"h "+m+"m"; return m+"m";
  }
  // windowRow renders one used/cap bar. A window with no cap renders as a plain
  // "used" line rather than a bar of unknown width.
  function win(w){
    if(!w) return "";
    var cap=Number(w.cap)||0, used=Number(w.used)||0;
    var h='<div class="qa-w"><div class="qa-w-h"><span>'+escH(w.label);
    var r=resetTxt(w.reset_at); if(r) h+=' <span style="opacity:.6">↻'+escH(r)+'</span>';
    var p = pct(used,cap);
    if(p===null){
      h+='</span><span class="qa-w-n">'+used.toLocaleString()+'</span></div>'+
         '<div class="qa-w-d">已用（无上限数据）</div></div>';
      return h;
    }
    if(w.exceeded===true) p=100;
    var cls="qa-fill"+((w.exceeded===true||p>=100)?" over":(p>=80?" warn":""));
    h+='</span><span class="qa-w-n">'+p+'%</span></div>'+
       '<div class="qa-bar"><div class="'+cls+'" style="width:'+Math.min(100,p)+'%"></div></div>'+
       '<div class="qa-w-d">'+used.toLocaleString()+' / '+cap.toLocaleString()+'</div></div>';
    return h;
  }
  // monthRow is Command Code's monthly allowance, which has no vendor cap.
  function month(r){
    var cap=Number(r.plan_cap)||0, used=Number(r.monthly_used)||0;
    if(!(used>0) && !(cap>0)) return "";
    var h='<div class="qa-w"><div class="qa-w-h"><span>每月';
    var pe=r.period_end;
    if(pe){ var d=Math.floor((new Date(pe)-Date.now())/86400000);
      if(d>0) h+=' <span style="opacity:.6">↻'+d+'d</span>'; }
    var p=pct(used,cap);
    if(p===null){
      h+='</span><span class="qa-w-n">$'+used.toFixed(2)+'</span></div>'+
         '<div class="qa-w-d" title="'+escH(MONTHLY_CAP_NOTE)+'">已用 · 套餐上限未知</div></div>';
      return h;
    }
    h+='</span><span class="qa-w-n">'+p+'%</span></div>'+
       '<div class="qa-bar"><div class="qa-fill'+(p>=100?" over":(p>=80?" warn":""))+
       '" style="width:'+Math.min(100,p)+'%"></div></div>'+
       '<div class="qa-w-d">$'+used.toFixed(2)+' / $'+cap.toFixed(0)+'</div>';
    if(r.cap_mismatch){
      h+='<div class="qa-warnl" title="已用+余额 与套餐额度不符，比例仅供参考">⚠ 额度口径可能有变</div>';
    }
    return h+"</div>";
  }
  // balanceLine renders the headline number. Three cases, kept visually distinct
  // because they mean different things: money left, a spend against a ceiling,
  // or no number at all.
  function balLine(r){
    if(r.error) return "";
    if(r.balance!==undefined && r.balance!==null){
      var extra="";
      // granted / topped_up may be NEGATIVE (deepseek reports topped_up -0.02
      // against a positive grant), so the zero test is on the ABSOLUTE value.
      // `!== 0` on a string-typed value would have been true anyway, but on a
      // coerced number it hid a real debit behind a flat 0.00.
      var g=Number(r.granted), t2=Number(r.topped_up);
      if((!isNaN(g)&&g!==0)||(!isNaN(t2)&&t2!==0)){
        extra=" <span class='qa-bal-l'>（赠送 "+
          money(g,r.currency)+" · 充值 "+money(t2,r.currency)+"）</span>";
      }
      var pill="";
      if(r.available===false) pill=' <span class="qa-pill out">不可用</span>';
      else if(r.exceeded) pill=' <span class="qa-pill out">已超额</span>';
      else if(r.below_threshold) pill=' <span class="qa-pill low">低于阈值</span>';
      return '<div class="qa-bal"><span class="qa-bal-n">'+money(r.balance,r.currency)+
        '</span><span class="qa-bal-l">余额</span>'+pill+extra+'</div>';
    }
    if(r.hard_limit!==undefined && r.hard_limit!==null){
      // No balance is derivable: a 1e8 ceiling is a policy number, not money.
      var s='<div class="qa-bal"><span class="qa-bal-n">'+money(r.total_usage||0,r.currency)+
        '</span><span class="qa-bal-l">累计花费（限额 '+
        (r.hard_limit>=1000000?'∞':money(r.hard_limit,r.currency))+'）</span></div>';
      return s;
    }
    return "";
  }
  function state(r){
    if(r.error) return {cls:"err", pill:'<span class="qa-pill out">读取失败</span>'};
    if(r.available===false) return {cls:"out", pill:'<span class="qa-pill out">不可用</span>'};
    if(r.exceeded) return {cls:"out", pill:'<span class="qa-pill out">已超额</span>'};
    if(r.below_threshold) return {cls:"low", pill:'<span class="qa-pill low">低于阈值</span>'};
    if(r.balance!==undefined && r.balance!==null){
      var b=Number(r.balance);
      if(b<=0) return {cls:"out", pill:'<span class="qa-pill out">无余额</span>'};
      if(b<5) return {cls:"low", pill:'<span class="qa-pill low">余额偏低</span>'};
      return {cls:"ok", pill:'<span class="qa-pill ok">'+(r.plan_status||"正常")+'</span>'};
    }
    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>'};
  }
  function card(r){
    var st=state(r);
    var h='<div class="qa-acc'+(r.error?" err":"")+'"><div class="qa-h">'+
      '<span class="qa-n">'+escH(r.name)+'</span>'+
      '<span class="qa-host">'+escH(r.host||"")+'</span>'+
      '<span class="qa-vendor">'+escH(r.vendor||"?")+'</span>'+
      '<span class="grow"></span>'+st.pill+'</div>';
    if(r.error){
      h+='<div class="qa-fail">'+escH(r.error)+'</div></div>';
      return h;
    }
    h+=balLine(r);
    var rows="";
    rows+=month(r);
    if(r.windows) r.windows.forEach(function(w){ rows+=win(w); });
    if(rows) h+='<div class="qa-grid">'+rows+'</div>';
    var meta=[];
    if(r.plan_id) meta.push("套餐 "+String(r.plan_id).replace(/^individual-/,""));
    if(r.period_end) meta.push("周期至 "+String(r.period_end).slice(0,10));
    if(r.cancel_at_period_end) meta.push("到期不续费");
    if(r.requests) meta.push(r.requests.toLocaleString()+" 次请求");
    if(r.tokens) meta.push((r.tokens/1e9).toFixed(2)+"B tokens");
    if(r.success_rate) meta.push("成功率 "+r.success_rate.toFixed(2)+"%");
    if(r.has_payment_method===false) meta.push("未绑定支付方式");
    if(meta.length) h+='<div class="qa-meta">'+meta.map(escH).join(" · ")+'</div>';
    return h+"</div>";
  }
  // paintSnapshot renders the persisted snapshot WITHOUT touching the network.
  // Called on load so the page has content immediately.
  async function showSnapshot(){
    try{
      var st = await pluginAPI.fetchState("quota");
      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; });
      render(real, skipped);
      var sub=document.getElementById("qa-sub");
      if(sub && st.updated_at){
        sub.textContent = real.length+" 个源 · 快照 "+new Date(st.updated_at).toLocaleTimeString()+"（点刷新取实时）";
      }
      return true;
    }catch(e){ return false; }
  }
  function render(reports, skipped){
    var host=document.getElementById("qa-list"); if(!host) return;
    var real=(reports||[]).filter(function(r){return !r.skipped;});
    var bad=real.filter(function(r){return r.error;});
    if(!real.length){
      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>';
      }
    }
    var sub=document.getElementById("qa-sub");
    if(sub) sub.textContent = real.length+" 个源 · "+bad.length+" 个读取失败";
  }
  async function refresh(){
    var srcs;
    try{
      var r=await fetch("/api/sources",{credentials:"same-origin"});
      srcs=(await r.json()).sources||[];
    }catch(e){
      var s0=document.getElementById("qa-sub"); if(s0) s0.textContent="拉取源列表失败: "+e.message;
      return;
    }
    if(!window.pluginAPI || typeof pluginAPI.call!=="function"){
      var sb=document.getElementById("qa-sub"); if(sb) sb.textContent="pluginAPI.call 不可用（内核版本过旧）";
      return;
    }
    var btn=document.getElementById("qa-refresh"); if(btn) btn.disabled=true;
    var sub=document.getElementById("qa-sub"); if(sub) sub.textContent="探测中…";
    var out;
    try{
    // 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});
    var due = (plan && plan.due) || [];
    var out = (plan && 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.
    var memo = {};
    (out||[]).forEach(function(r){
      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.
      if(r.reason && String(r.reason).indexOf("backing off")===0) 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;
      return;
    }
    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;});
    // real may already contain errors; collect_all keeps them for visibility.
    render(real, skipped);
    if(btn) btn.disabled=false;
    // Persist both the snapshot and the probe map; the map is what lets the
    // next page open skip the upstreams that expose nothing.
    try{
      await pluginAPI.postState("quota",
        {reports:all, updated_at:Date.now(), probe_memo:(window.__qaMemo||{})});
    }catch(e){}
  }
  function mount(){
    var b=document.getElementById("qa-refresh");
    if(b && !b.__qaWired){ b.__qaWired=true; b.addEventListener("click",refresh); }
  }
  mount();
  showSnapshot();
  // Do NOT auto-refresh on load.
  //
  // Probing is network-bound and every request takes the plugin's write lock, so
  // a page open that fires one call per source serialises them: measured 19s to
  // first paint and 22-100s to finish, with the whole WebUI held in its loading
  // state meanwhile. A balance the operator did not ask for is not worth the
  // first paint of the page they opened.
  //
  // What IS shown automatically is the last snapshot, so an operator who hits
  // refresh once sees a card immediately and a fresh number behind it.
  if (window.pluginAPI && pluginAPI.onTabShown) pluginAPI.onTabShown(mount);
  window.__qaRefresh=refresh; window.__qaRender=render;
})();
</script>
]==],
  },
}

return PLUGIN