Files
ModelRouter/internal/lua/plugins.go
JianFeeeee af8b93bd7c fix(内核+插件): 测试写生产账号池 —— 13 个账号被覆盖成 0 个(真实数据事故)
用户报"完全不可用"。逐平台查 account/list,发现 13 个账号里只有 1 个有凭据,
其余是占位壳子 —— 再查发现**账号池被清空了**,而保险库里凭据还在
(症状正是"账号凭空消失")。本提交修的是这个数据事故的根因。

## 根因(全程实测,非推断)

channels 插件用 `storeDir()` 决定自己的存储位置,它有多级兜底:
env → settings → **读 /etc/llmsproxy/config.yaml 的 plugin_dir** → 硬编码。
测试台没设 `LLMSPROXY_PLUGIN_DIR` 时,后两级直接命中**生产**:

    测试插件目录 = /tmp/TestXXX/002/plugins
    storeDir 结果 = /etc/llmsproxy/plugins          ← 实测输出

于是测试加载真 channels.lua 时**读的是生产存储、写的也是生产存储**。
一次全量 `go test ./internal/lua` 就把
`/etc/llmsproxy/plugins/.channels-store.json` 从 1730 覆盖成 1986 字节、
账号池清空(inode 都换了)。

同类前科:曾把 26 个测试账号写进生产 store。两次同一类错误。

## 修法(两层,第二层才是根治)

1. **内核把插件的真实目录交给插件**(ground truth,不再靠猜):
   新增 `__llmsproxy_plugin_dir` 全局,在 bootPluginState 里注册。
   `storeDir()` 优先用它 —— 这样**即使没有环境变量**也不会落到生产路径。
   实测:撤掉 env 后 storeDir 仍落在临时目录。
2. 测试台补隔离:`runtimePlugins` 设 env + `DisableStatePersistence()`
   (插件声明了 schedule 时内核会自动起调度器,**后台 goroutine** 比
   `t.Setenv` 生命周期长,测试结束后仍可能落盘)。

## 判据 + 变异验证

`internal/lua/store_dir_isolation_test.go`:
- storeDir 绝不能是 /etc/llmsproxy 前缀
- 必须落在临时目录
- env 必须真的被设上(机制而非巧合)

**变异验证**:把隔离设成生产路径 → 判据变红并打印
「storeDir 指向了**生产**目录」;恢复 → 变绿。

全量回归:9 个包全绿,且 `go test ./...` 后生产存储**未被改动**(此前必被改)。
2026-10-14 11:31:41 +08:00

2549 lines
94 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package lua
// Plugin runtime.
//
// A plugin is a single .lua file, loaded from its own directory, that extends
// the gateway in two ways:
//
// 1. Hooks: it registers callbacks on the request pipeline's stages
// (request_start, response_end, …). A hook receives a JSON table and
// returns either nil (no opinion) or a JSON object.
// 2. UI: at boot it returns HTML/CSS/JS fragments that the kernel injects
// into the WebUI — either as a whole new page, or as an extra element on
// an existing page.
//
// Design notes that are load-bearing (each one cost something to learn):
//
// - Plugins run in a SEPARATE VM from adapters, and each plugin gets its own
// elastic pool, exactly like an adapter. Sharing one state would let a
// plugin's globals corrupt an adapter's protocol translation (or vice
// versa), and a plugin is third-party code while an adapter is core.
//
// - A plugin that errors must NEVER break request forwarding. Hook calls are
// wrapped so a plugin error is logged and the original value is returned
// unchanged. A broken plugin is a missing feature, not an outage — the same
// reason transform_stream_chunk swallows errors today.
//
// - The billing plugin therefore cannot be trusted to be the source of truth
// for anything the gateway must enforce. It reads usage off the hook
// payload and accumulates in its own state; the gateway's own quota
// accounting (internal/gateway/stats.go) stays authoritative for limits.
// Two accounting paths that disagree is worse than one that is slightly
// less featureful, so the split is explicit and documented.
import (
"encoding/json"
"fmt"
"log"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"sync/atomic"
"time"
golua "github.com/aarzilli/golua/lua"
)
// Stage identifies a point in the request pipeline. Plugins may register a
// function for any stage; unknown stages are ignored at call time so an old
// plugin survives a gateway that grew new stages.
type Stage string
// The pipeline stages. The order here is the order they fire in; it is the
// contract plugins are written against.
const (
// StageRequestStart fires after the gateway has parsed and authorized a
// request but BEFORE any upstream slot is chosen. payload:
// stage, type ("chat"|"stream"|"image"), model (as requested),
// key (masked gateway key id), role, source (empty), stream (bool),
// messages_count, tools_count, ts (unix seconds).
StageRequestStart Stage = "request_start"
// StageRouted fires once a (source, model) slot has been chosen and before
// the upstream call. payload adds: source, model (the resolved one),
// tier (AUTO tier, -1 for the direct path), stream.
StageRouted Stage = "routed"
// StageChainStep fires ONCE PER STEP of an AUTO chain walk, and only on the
// AUTO path (a direct request has no chain and therefore emits nothing).
//
// This exists because StageRouted cannot express degradation: it fires once,
// after the walk, with the slot that finally won. "tier 1 was cooling so we
// dropped to tier 3" and "tier 1 served it" were indistinguishable. That
// distinction is the whole point of a priority chain, and it is what an
// operator debugging "why did my expensive model not get used" needs.
//
// payload:
// kind "tier_skip" | "slot_fail" | "tier_busy" | "selected"
// tier the AUTO tier this step belongs to (1 = highest priority)
// source / model set for slot_fail and selected
// reason human-readable cause, for tier_skip and tier_busy
// error the underlying error text, for slot_fail
// attempt 1-based slot attempt within this walk
//
// Ordering: every step precedes StageRouted, and the "selected" step is the
// last one. A plugin accumulating the walk therefore has the full picture
// by the time request_end arrives.
//
// These events are OBSERVATION ONLY — see the accounting note in
// docs/plugins.md: nothing here feeds back into routing, cooldown or quota.
StageChainStep Stage = "chain_step"
// StageRequestEnd fires exactly once per request, after the client response
// has been produced (or after a failure was recorded). payload adds:
// source, model, ok, status, latency_ms, first_byte_ms, prompt_tokens,
// completion_tokens, cache_hit_tokens, cache_miss_tokens, image_count,
// error ("" when ok).
//
// This is the stage a billing plugin should read: it carries the final
// accounting for the request, including the upstream's own usage numbers.
StageRequestEnd Stage = "request_end"
)
// AllStages is the firing order, used by the docs and by the hook listing.
var AllStages = []Stage{
StageRequestStart,
StageChainStep,
StageRouted,
StageRequestEnd,
}
// UIExtension is what a plugin contributes to the WebUI at boot.
type UIExtension struct {
// Page is a whole new sidebar entry + pane. Requires PageID and Title.
// The kernel renders Page's HTML into a pane whose id is "tab-"+PageID and
// adds a sidebar button with data-tab="<PageID>".
Page *UIPage `json:"page,omitempty"`
// Pages is the multi-page form of the same thing. A plugin with several
// distinct screens (billing totals vs. the price rules that produced
// them) would otherwise have to cram both into one pane behind tabs, or
// smuggle the second one in as a hidden element. Both make the sidebar
// lie about what the plugin contributes.
//
// Page and Pages merge: a plugin may use either or both.
Pages []*UIPage `json:"pages,omitempty"`
// Elements are snippets injected into EXISTING pages, keyed by target page
// id (e.g. "status", "keys"). Order within a target is plugin load order.
Elements []UIElement `json:"elements,omitempty"`
// Visibility is the plugin-wide default audience, applied to every page and
// element this plugin contributes. A per-page Audience narrows it further.
Visibility string `json:"visibility,omitempty"`
// StateScope splits the plugin's published state into global and per-caller
// buckets, for a plugin that is readable by more than one role but must not
// hand one caller another's data.
StateScope *StateScope `json:"state_scope,omitempty"`
}
// UIPage is a plugin-provided page.
type UIPage struct {
PageID string `json:"page_id"` // kebab-case; becomes data-tab and #tab-<id>
Title string `json:"title"` // sidebar label
Icon string `json:"icon"` // optional inline SVG or short glyph
Order int `json:"order"` // sidebar sort key (default 100)
// Audience limits which API-key roles may see this page. "" and "all" mean
// every authenticated role; "admin" hides it from user keys; a comma list
// ("admin,user") names the roles allowed.
//
// This exists because a plugin page is not automatically public data. The
// billing page shows GLOBAL cost across every source, and the quota page
// shows upstream balances — neither belongs to whoever happens to hold a
// user key, yet both were served to them because /api/ui-inject applied no
// role filter at all.
//
// The plugin declares its own default and the gateway may override it (see
// SetPluginVisibility), because the plugin author cannot know the operator's
// policy but the operator must be able to impose one.
Audience string `json:"audience,omitempty"`
// Mount is the page body. It may contain <script> and <style>; the kernel
// executes scripts AFTER injecting the HTML so the DOM exists, and exposes
// `pluginAPI` to them (see docs/plugins.md).
Mount string `json:"mount"`
}
// UIElement is a snippet injected into an existing page.
type UIElement struct {
// Target is the id of the pane to inject into: "status", "chat", "keys",
// "sort", "sources" or "adapters".
Target string `json:"target"`
// Audience is the same role filter as UIPage.Audience. An injected tile can
// leak exactly what a whole page would — the billing status tile shows total
// cost — so the field exists here too rather than only on pages.
Audience string `json:"audience,omitempty"`
// Anchor is where in the target pane: "top", "bottom" or "before:<sel>" /
// "after:<sel>" for a CSS selector. Empty = "bottom".
Anchor string `json:"anchor,omitempty"`
Order int `json:"order,omitempty"` // sort key within the target
Mount string `json:"mount"`
}
// PluginInfo is the manifest a plugin declares.
type PluginInfo struct {
Name string `json:"name"`
Version string `json:"version"`
Description string `json:"description,omitempty"`
Author string `json:"author,omitempty"`
}
// Plugin is one loaded plugin.
type Plugin struct {
Info PluginInfo
Hooks map[Stage]string // stage -> exported function name
UI *UIExtension
// LoadError is non-empty when the plugin failed to compile or register. Such
// a plugin is listed in the UI with its error but is never called.
LoadError string
// Disabled marks a plugin the operator switched off. It stays on disk and
// keeps its name, hooks and UI declared (so the UI can show it greyed out
// and report what it WOULD contribute), but Fire never calls it and its
// UI extension is excluded from the inject payload.
//
// Disabling is deliberately NOT deleting: a plugin that breaks a live
// gateway is often one line away from being fixed, and an operator needs a
// way to take it out of the request path without losing it. That is the same
// reasoning as a systemd unit being masked rather than removed.
Disabled bool
dir string
// http lists endpoints this plugin mounts under /api/plugins/http/. Read at
// load time, never mutated afterwards.
http []HTTPHandler
// exports lists the functions this plugin allows the WebUI to call through
// POST /api/plugins/{name}/call/{fn}. It is read once at load from
// plugin.exports and never changes, so Call can check it without taking a
// Lua read on the hot path.
exports []string
// script is the plugin's source, kept so a reload can rebuild its state.
script string
// Builtin marks a plugin that shipped with the gateway. The UI shows it as
// such so an operator can tell "example I can read" from "mine", and a
// delete of a builtin is allowed but re-seeds on a fresh plugin dir.
Builtin bool
// state is the plugin's SINGLE authoritative Lua state, guarded by mu.
//
// A plugin deliberately does NOT use the adapter's elastic pool. Adapters
// are per-worker stateless transforms, so N independent states are correct
// (and necessary) for them. A plugin's hook, however, typically ACCUMULATES
// into `plugin.state` — the billing plugin's totals live there — so two
// states would mean two divergent sets of totals, and whichever worker a
// hook happened to get would see a different number. That was a real bug
// found by a test: SetState wrote prices to one worker, the hook then ran on
// another and priced everything at zero.
//
// A single state is sufficient because every entry point (Fire, State,
// SetState) serializes on mu, and a plugin hook is a short synchronous call.
// A hook that blocks for seconds would stall every other plugin's hook,
// which is the real cost — documented as a rule in docs/plugins.md.
mu sync.Mutex
state *worker
// persisted marks this plugin's state as having been restored from disk.
// It gates the first save: without it, a freshly loaded plugin whose state
// is still the compiled-in default would immediately overwrite the file it
// was supposed to inherit from.
persisted bool
// ExtraHosts lists outbound hosts this plugin declared in its own source
// (`plugin.hosts`) beyond the operator-configured sources. A REQUEST, not a
// grant: the operator approves each one in the settings page.
ExtraHosts []string
// KernelDeclared lists the kernel actions this plugin asks for in its own
// source (`plugin.kernel = {...}`). It is a REQUEST, shown to the operator
// beside what was actually granted; it grants nothing by itself.
KernelDeclared []string
// Schedule is the periodic work this plugin declared, and Runtime is the
// scheduler that runs it on THIS plugin's own VM (p.state) under p.mu — the
// same state and the same lock a request hook uses. That is what makes a
// token rotated by a task immediately visible to inference traffic.
//
// Runtime is nil when the plugin declared nothing — the common case stays
// exactly as cheap as before: one VM, no extra goroutine. The point of the
// schedule is auto check-in / token rotation: work that must happen with no
// browser open and nobody clicking anything.
Schedule []ScheduledTask
Runtime *PluginRuntime
}
// Plugins is the loaded plugin set, owned by the VM.
type Plugins struct {
mu sync.RWMutex
vm *VM
dir string // plugin directory; "" disables plugin loading entirely
plugins []*Plugin // load order; the index is the stable plugin id
// ui caches the merged UI extensions so the boot payload is computed once.
ui atomic.Pointer[UIExtension]
// stageFuncs is the precomputed stage -> []hookCall, in plugin load order.
stageFuncs map[Stage][]hookCall
// hookErr records per-stage plugin failures so a silently broken plugin is
// visible in /api/status rather than merely missing.
hookErr *hookErrors
// logf, when set, receives persistence warnings. It is a field rather than a
// direct log call so the plugin package stays free of a logging dependency
// and tests can capture the warnings.
logf func(format string, args ...interface{})
// visibility holds the operator's per-plugin audience overrides, keyed by
// plugin name. Consulted by UIForRole and the call endpoint's authorisation.
visibility visibilityOverrides
// adapterOwn records which plugin installed each adapter, so a plugin can
// only update adapters it owns (see plugin_adapter_bridge.go for why this
// is state and not a convention).
adapterOwn *adapterOwnership
adapterOwnOnce sync.Once
// routeWinners records the operator's arbitration for contested HTTP service
// paths: a path two+ plugins declared, resolved to one winning plugin. Keyed
// by normalized path. See route_conflicts.go — a contested path that is NOT
// in here is unresolved and the dispatcher returns 409 rather than silently
// picking a load-order winner. Persisted in .route-conflicts.json.
routeWinners map[string]string
// adapterWinners records the operator's arbitration for an adapter NAME two
// things claimed (user requirement B): "incumbent" keeps the built-in / first
// plugin, "candidate:<plugin>" lets that plugin's version win. Persisted in
// .route-conflicts.json beside routeWinners. See route_conflicts.go.
adapterWinners map[string]string
// adapterConflicts holds adapter-name installs currently SUSPENDED pending an
// operator decision: a plugin tried to install an adapter whose name an
// incumbent already holds. The incumbent keeps serving; the candidate waits
// here (with its code) until arbitrated. In-memory only — a restart replays
// the install, which re-suspends unless a persisted winner says otherwise.
adapterConflicts map[string]*AdapterConflict
// 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
// allowHTTP bounds plugin outbound requests to operator-declared hosts.
// nil means deny-all: an unconfigured gateway must not silently grant a
// plugin unrestricted network access.
allowHTTP *httpAllowlist
// kernelAPI is the injected kernel action surface (nil when unwired).
// See kernel_api.go: a plugin calling __llmsproxy_kernel reaches the core
// through this interface, never through a direct import (core depends on
// lua, not the other way round).
kernelAPI KernelAPI
// secretBox seals the credential vault at rest. nil = write plaintext (tests
// / embedders with no runtime dir). Injected by core via SetSecretBox: it is
// the deployment's ONE master key, the same one that seals config.yaml, so a
// lost master.key loses the plugin tokens too — loud, consistent, not a
// second key to manage.
secretBox secretSealer
// kernelPerms records what the OPERATOR granted each plugin. Separate from
// what a plugin DECLARES it wants (Plugin.KernelDeclared): a declaration is
// a request, a grant is a privilege. Persisted in .access.json.
kernelPerms kernelPermissionSet
// sourceHosts remembers the source-derived allowlist so a rebuild can fold it
// together with any granted login hosts.
sourceHosts []string
// extraHosts is the per-plugin set of APPROVED extra outbound hosts. Separate
// from what a plugin declared (Plugin.ExtraHosts): a declaration is a request.
extraHosts map[string][]string
// save coalesces state writes into one background goroutine.
//
// A hook must NOT write synchronously: the billing plugin mutates state on
// every single request, and serializing the whole state per request would put
// a file write (plus a full JSON encode) on the hot path — measured at
// 14.6us per hook call already, a write would dominate it. Instead hooks mark
// the plugin dirty and this saver flushes, so N requests between two flushes
// cost one write.
save *stateSaver
}
// stateSaver coalesces state writes: one goroutine, a minimum interval between
// flushes, and a dirty set. Under load the number of writes is bounded by the
// timer rather than by the request rate.
type stateSaver struct {
mu sync.Mutex
ps *Plugins
dirty map[string]bool
wake chan struct{}
stop chan struct{}
done chan struct{}
// wakeDelay is how long a burst suppresses the next tick. A separate field
// (not the constant 2s) so tests can shorten it: Close() waits for the saver
// goroutine, so a test with the production interval pays that interval on
// every teardown.
wakeDelay time.Duration
once sync.Once
// interval is the minimum gap between flushes.
interval time.Duration
// writes counts completed file writes; tests read it to assert that N
// mutations did NOT become N writes.
writes atomic.Int64
}
// setWakeDelayForTest shortens the debounce so tests do not pay it on teardown.
func (s *stateSaver) setWakeDelayForTest(d time.Duration) { s.wakeDelay = d }
func newStateSaver(ps *Plugins) *stateSaver {
s := &stateSaver{
ps: ps,
dirty: map[string]bool{},
wake: make(chan struct{}, 1),
stop: make(chan struct{}),
done: make(chan struct{}),
interval: 2 * time.Second,
wakeDelay: 2 * time.Second,
}
go s.loop()
return s
}
// markDirty records that a plugin's state changed. Never blocks: the channel is
// buffered, and a full buffer means a flush is already pending, which is exactly
// the state we want.
func (s *stateSaver) markDirty(name string) {
if s == nil {
return
}
s.mu.Lock()
s.dirty[name] = true
s.mu.Unlock()
select {
case s.wake <- struct{}{}:
default:
}
}
func (s *stateSaver) loop() {
defer close(s.done)
t := time.NewTicker(s.interval)
defer t.Stop()
for {
select {
case <-s.stop:
// Final flush so a clean shutdown does not lose the tail — losing
// the last interval of spend is the same bug in a smaller window.
s.flush()
return
case <-t.C:
s.flush()
case <-s.wake:
// Debounce a burst of marks into one write.
time.Sleep(s.wakeDelay)
s.flush()
}
}
}
// flush writes every dirty plugin's state.
func (s *stateSaver) flush() {
if s == nil {
return
}
s.mu.Lock()
if len(s.dirty) == 0 {
s.mu.Unlock()
return
}
names := make([]string, 0, len(s.dirty))
for n := range s.dirty {
names = append(names, n)
}
s.dirty = map[string]bool{}
s.mu.Unlock()
// Snapshot + write here, off the request path. This is the ONLY place the
// plugin's Lua tables are walked for persistence.
for _, n := range names {
s.ps.persistByName(n)
s.writes.Add(1)
}
}
// Close stops the saver AFTER its final flush completes.
//
// It must WAIT for that flush, not just signal it. Signalling and returning
// leaves the write to a goroutine that the caller is about to tear down
// (vm.Stop() frees the plugin states; the test process is exiting), so the
// final state is silently lost — which is the very defect persistence was added
// to fix. Close is on the shutdown path, where a few milliseconds is free.
func (s *stateSaver) Close() {
if s == nil {
return
}
s.once.Do(func() { close(s.stop) })
<-s.done
}
type hookCall struct {
pluginIdx int
plugin string
fn string
}
// hookErrors counts hook failures per stage, surfaced in /api/status so a
// silently broken plugin is visible instead of just missing.
type hookErrors struct {
mu sync.Mutex
counts map[Stage]int
last map[Stage]string
}
func newHookErrors() *hookErrors {
return &hookErrors{counts: map[Stage]int{}, last: map[Stage]string{}}
}
func (h *hookErrors) note(s Stage, msg string) {
h.mu.Lock()
h.counts[s]++
if len(msg) > 200 {
msg = msg[:200]
}
h.last[s] = msg
h.mu.Unlock()
}
func (h *hookErrors) snapshot() map[string]map[string]interface{} {
h.mu.Lock()
defer h.mu.Unlock()
out := map[string]map[string]interface{}{}
for s, n := range h.counts {
if n == 0 {
continue
}
out[string(s)] = map[string]interface{}{"count": n, "last_error": h.last[s]}
}
return out
}
// pluginGlobal is the Lua global the plugin's returned table is stored under,
// mirroring adapterGlobal for adapters.
const pluginGlobal = "__llmsproxy_plugin"
// NewPlugins creates the plugin registry for a VM. dir is the plugin directory;
// a missing directory is not an error (plugins are optional).
func NewPlugins(vm *VM, dir string) *Plugins {
ps := &Plugins{
vm: vm,
stageFuncs: map[Stage][]hookCall{},
hookErr: newHookErrors(),
dir: dir,
// Deny-by-default. NewPlugins is called during wiring, before any
// source list is known; SetOutboundHosts replaces this with the real
// set. A plugin loaded in that window simply cannot make requests.
allowHTTP: newHTTPAllowlist(nil),
}
// The saver is started even with no dir: a gateway can be given a plugin dir
// 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
}
// Close stops the background state saver, flushing once more first.
//
// The gateway MUST call this on shutdown. Without it the last flush interval's
// accumulation is lost, which is the same "restart loses the books" defect this
// persistence exists to fix, just scoped to a few seconds instead of the whole
// process lifetime.
// DisableStatePersistence turns persistence off (tests/benchmarks only).
func (ps *Plugins) DisableStatePersistence() { ps.stateDirDisabled = true }
// SetOutboundHosts declares which hosts plugins may contact via
// __llmsproxy_http. The gateway passes the host of every configured source:
// those are operator-declared upstreams, and reusing that set keeps a plugin's
// reach no wider than the deployment already trusts. An empty list means no
// outbound host is allowed at all — plugins then still load, they just cannot
// make requests.
func (ps *Plugins) SetOutboundHosts(hosts []string) {
ps.mu.Lock()
defer ps.mu.Unlock()
// Remembered so a later extra-host grant can rebuild the allowlist as
// sources ∪ granted-extra-hosts instead of losing the source half.
ps.sourceHosts = append([]string(nil), hosts...)
ps.rebuildOutboundAllowlistLocked()
}
// outboundAllowlist returns the current allowlist. bootPluginState reads it
// before any concurrent reload can swap it, and a nil result is treated as
// "deny everything" rather than "allow everything".
func (ps *Plugins) outboundAllowlist() *httpAllowlist {
ps.mu.RLock()
defer ps.mu.RUnlock()
return ps.allowHTTP
}
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()
}
// LoadDir loads every .lua file in dir as a plugin. Files are loaded in
// lexical order so a plugin's UI order is stable across restarts.
//
// A plugin that fails to compile or register is NOT fatal: it is kept with its
// LoadError so the UI can show it, and it is never called. This is the same
// posture as adapters, except adapters are core while plugins are not.
func (ps *Plugins) LoadDir() error {
if ps.dir == "" {
return nil
}
// Overrides first: they are keyed by plugin name and must be in place before
// anything reads them, and a reload must not blank a restriction the
// operator set earlier in this same process.
if err := ps.LoadAccessOverrides(); err != nil {
log.Printf("[core] plugin access overrides not loaded: %v", err)
}
// Kernel grants load beside them and for the same reasons: an in-memory map
// would make "this plugin may reload config" look enforced right up to the
// next restart, at which point the grant silently vanished.
if err := ps.LoadKernelPermissions(); err != nil {
log.Printf("[core] plugin kernel permissions not loaded: %v", err)
}
// Route-conflict arbitrations are operator policy too: who wins a service
// path two plugins both declared. Must be in place before the first request
// is dispatched, or a contested path would 409 until the operator re-chose.
if err := ps.LoadRouteConflicts(); err != nil {
log.Printf("[core] plugin route conflicts not loaded: %v", err)
}
if _, err := os.Stat(ps.dir); os.IsNotExist(err) {
return nil
}
entries, err := os.ReadDir(ps.dir)
if err != nil {
return err
}
var names []string
for _, e := range entries {
if e.IsDir() || !strings.HasSuffix(e.Name(), ".lua") {
continue
}
names = append(names, e.Name())
}
sort.Strings(names)
for _, n := range names {
path := filepath.Join(ps.dir, n)
code, err := os.ReadFile(path)
if err != nil {
continue
}
if err := ps.LoadSource(strings.TrimSuffix(n, ".lua"), string(code)); err != nil {
// LoadSource records the error on the plugin itself; keep going so
// one bad plugin does not stop the others from loading.
continue
}
}
ps.rebuild()
// Persisted host grants must be folded into the allowlist HERE, not only in
// the WebUI grant path.
//
// Why (live defect): LoadKernelPermissions above loads `jp.Hosts` into
// ps.extraHosts but deliberately does NOT fold it — its comment says "the
// caller owns that fold". On the boot path nobody did. SetOutboundHosts had
// already run (core.New, before LoadDir), so the allowlist held the source
// hosts and nothing else, and every approved extra host (the plugin's whole
// declared set: zcode.z.ai, copilot.tencent.com, …) was refused after a
// restart with "host ... is not an allowed outbound destination".
//
// Symptom is the classic "restart loses the operator's decision": it clears
// the moment an operator opens the plugin page and re-saves the grants. It
// also broke the scheduled work — refresh_models fires 5s after boot, inside
// that window, and reported the grant as missing for every platform.
ps.mu.Lock()
ps.rebuildOutboundAllowlistLocked()
ps.mu.Unlock()
return nil
}
// SeedBundled writes the plugins shipped with the gateway into dir when the
// directory does not exist yet, mirroring the adapter seeding rule: once the
// directory exists it is authoritative, so deleting or editing a shipped plugin
// is a real action that survives restarts.
// SeedBundled writes the plugins shipped with the gateway into dir when they are
// MISSING.
//
// This used to be "write everything only when the directory does not exist",
// which silently stopped delivering plugins added to the binary LATER: the
// directory had existed since the first install, so a newly embedded plugin
// (e.g. channels.lua) never landed on an upgraded deployment — it was in the
// binary and completely invisible in /api/plugins. The directory's mere
// existence is not evidence that its contents are current.
//
// The rule is now per-file: seed a bundled plugin only when a file of that name
// is absent. An OPERATOR'S file always wins, so customising or replacing a
// bundled plugin survives an upgrade — the same rule adapters already use.
func (ps *Plugins) SeedBundled() error {
if ps.dir == "" {
return nil
}
return writeBundledPlugins(ps.dir)
}
// bootPluginState compiles one plugin into a fresh Lua state and stores its
// returned table under pluginGlobal. It is the plugin counterpart of
// adapterPool.boot, minus the pooling: a plugin keeps exactly one state.
func bootPluginState(code, name string, allow *httpAllowlist, ps *Plugins) (*worker, error) {
L := golua.NewState()
L.OpenLibs()
setupGlobals(L)
// Only plugins get __llmsproxy_http. Adapters are operator-supplied config
// and go through setupGlobals too, so installing the bridge here — not in
// setupGlobals — is what keeps the capability scoped to installed .lua
// files.
if allow != nil {
registerFn(L, httpGlobal, luaHTTPFn(L, allow))
}
// R4: plugins — and only plugins — can install/update/remove their own
// adapters at runtime. Scoped here (not in setupGlobals) for the same reason
// as __llmsproxy_http: adapters share setupGlobals and must not get it.
if ps != nil {
registerAdapterBridge(L, ps, name)
// R4-d: the credential vault and the kernel action surface. Both are
// plugin-only, for the same reason as the bridges above — an adapter
// shares setupGlobals and must not be able to read credentials or drive
// the kernel.
registerSecretBridge(L, ps, name)
registerKernelBridge(L, ps, name)
// The plugin's own directory, as ground truth (see pluginDirGlobal).
registerPluginDirBridge(L, ps.dir)
}
if err := L.DoString(code); err != nil {
L.Close()
return nil, fmt.Errorf("compile plugin %s: %w", name, err)
}
if L.Type(-1) != golua.LUA_TTABLE {
L.Close()
return nil, fmt.Errorf("plugin %s must return a table", name)
}
// SetGlobal POPS, so this is the only global it can set (see vm.go boot()).
L.SetGlobal(pluginGlobal)
L.SetTop(0)
return &worker{L: L}, nil
}
// LoadSource loads one plugin from source text. name is the plugin id (the
// file's base name). It returns an error only for conditions the caller should
// see; a plugin that merely registers nothing is not an error.
func (ps *Plugins) LoadSource(name, code string) error {
if name == "" {
return fmt.Errorf("plugin name required")
}
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{},
dir: ps.dir,
}
// A plugin runs in its OWN Lua state, separate from every adapter and every
// other plugin, so an error in one cannot corrupt another. The returned
// table is stored under pluginGlobal, exactly like adapterGlobal.
//
// One state, held for the plugin's lifetime (see Plugin.state): a hook that
// accumulates into plugin.state would otherwise split its totals across
// whichever worker it happened to run on.
w, err := bootPluginState(code, name, ps.outboundAllowlist(), ps)
if err != nil {
p.LoadError = err.Error()
ps.append(p)
ps.rebuild()
return err
}
p.script = code
p.state = w
// Builtin = the shipped source is byte-identical to what this file was
// loaded from. That is stronger than "the name matches a bundled plugin": an
// operator who EDITED billing.lua must not be told their copy is builtin,
// or the UI would offer to overwrite their changes.
if orig, err := ReadBundledPlugin(name); err == nil && orig == code {
p.Builtin = true
}
// ---- manifest ----
w.L.GetGlobal(pluginGlobal)
if !w.L.IsNil(-1) {
w.L.GetField(-1, "name")
if s := w.L.ToString(-1); s != "" {
p.Info.Name = s
}
w.L.SetTop(-2)
w.L.GetField(-1, "version")
if s := w.L.ToString(-1); s != "" {
p.Info.Version = s
}
w.L.SetTop(-2)
w.L.GetField(-1, "description")
if s := w.L.ToString(-1); s != "" {
p.Info.Description = s
}
w.L.SetTop(-2)
w.L.GetField(-1, "author")
if s := w.L.ToString(-1); s != "" {
p.Info.Author = s
}
w.L.SetTop(-2)
}
w.L.SetTop(0)
// ---- hooks ----
for _, st := range AllStages {
if fn, ok := pluginHookName(w.L, string(st)); ok {
p.Hooks[st] = fn
}
}
// ---- schedule ----
//
// A plugin that declares `schedule` gets a kernel-driven scheduler running
// ON ITS OWN VM (the same p.state a hook uses), so its periodic work (auto
// check-in, token rotation, quota polling) happens with no browser open and
// no operator clicking anything. Without this the only periodic work a
// plugin could do was a front-end poll, which is exactly the "nobody is
// looking at the dashboard" case the feature exists for.
//
// The parse happens here (right after boot) so a malformed schedule is
// visible at load time rather than silently ignored.
p.Schedule = parseSchedule(w.L).tasks
w.L.SetTop(0)
if len(p.Schedule) > 0 {
p.Runtime = ps.startRuntime(p, scheduleSpec{tasks: p.Schedule})
if p.Runtime == nil {
// The scheduler could not start. That must NOT silently disable the
// plugin — hooks still work — but the operator needs to know their
// scheduled work will never run.
log.Printf("[scheduler] plugin %s: declared %d task(s) but no runtime could be started",
p.Info.Name, len(p.Schedule))
}
}
// ---- UI ----
if ui, ok := readPluginUI(w.L, p.Info.Name); ok {
p.UI = ui
}
// ---- declared kernel requests ----
// A declaration is a REQUEST, not a grant (see kernel_api.go). Reading it at
// load time lets the settings page show "this plugin wants add_source"
// beside what was actually granted.
p.KernelDeclared = readPluginKernelDeclared(w.L)
w.L.SetTop(0)
// ---- declared extra hosts ----
// Some plugins need hosts that are not sources (account portals, claim
// endpoints), which the outbound allowlist blocks by default. The plugin
// declares what it needs; the operator approves each host. See plugin_outbound.go.
p.ExtraHosts = readPluginExtraHosts(w.L)
w.L.SetTop(0)
// ---- http endpoints ----
p.http = readPluginHTTP(w.L)
// ---- 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
// to decide whether a call is allowed.
p.exports = readPluginExports(w.L)
// Restore persisted state AFTER the plugin compiled and registered its
// hooks, so the restore overwrites the compiled-in defaults instead of being
// overwritten by them. Doing it earlier would mean a restart resets the
// totals back to whatever the .lua source initialises them to.
ps.restore(p)
ps.append(p)
// Rebuild here rather than only in LoadDir: LoadSource is also the single-
// plugin entry point (the WebUI upload path), and a caller that loads one
// plugin and immediately fires a stage must not silently get nothing.
ps.rebuild()
return nil
}
// pluginHookName returns the exported function name a plugin registered for a
// stage. A plugin registers either `hooks = {request_end = "on_end"}` or a
// direct `request_end = function(...) end` on the returned table; both forms are
// accepted because the table form keeps the manifest tidy while the direct form
// is shorter for a single-hook plugin.
//
// For the anonymous-function forms the value is re-keyed onto the plugin table
// under a synthetic name so the hot path can fetch every hook by name.
//
// STACK DISCIPLINE (this binding aborts the PROCESS on a bad index — SIGABRT,
// not a Go panic, so nothing can recover it):
//
// - GetField/SetField take the table by ABSOLUTE index, so the plugin table's
// index must be re-read after every SetTop, since SetTop(0) invalidates it.
// - Therefore each form re-pushes the plugin table and re-reads its index,
// instead of caching one index across a reset. Getting this wrong was a
// real crash found by running the test, not by reading the code.
func pluginHookName(L *golua.State, stage string) (string, bool) {
L.SetTop(0)
L.GetGlobal(pluginGlobal)
if L.IsNil(-1) {
L.SetTop(0)
return "", false
}
plug := L.GetTop()
// form 1: hooks = { request_end = "fn" } (or = function)
L.GetField(plug, "hooks")
if L.Type(-1) != golua.LUA_TNIL {
hooksIdx := L.GetTop()
L.GetField(hooksIdx, stage)
switch L.Type(-1) {
case golua.LUA_TSTRING:
name := L.ToString(-1)
L.SetTop(0)
if name != "" {
return name, true
}
return "", false
case golua.LUA_TFUNCTION:
// An anonymous function: key it onto the plugin table under a stable
// per-stage name so invoke() can fetch it like any other hook.
name := "__hook_" + stage
L.SetField(plug, name)
L.SetTop(0)
return name, true
}
L.SetTop(0)
}
// form 2: request_end = function(...) end directly on the table.
// Re-push and re-read the index: SetTop(0) above invalidated `plug`.
L.GetGlobal(pluginGlobal)
if L.IsNil(-1) {
L.SetTop(0)
return "", false
}
plug = L.GetTop()
L.GetField(plug, stage)
if L.Type(-1) == golua.LUA_TFUNCTION {
name := "__hook_" + stage
L.SetField(plug, name)
L.SetTop(0)
return name, true
}
L.SetTop(0)
return "", false
}
// readPluginUI reads the optional ui extension block.
// readPluginExports collects plugin.exports, the list of functions the WebUI is
// allowed to invoke through POST /api/plugins/{name}/call/{fn}.
//
// It accepts either an array of names or a map whose KEYS are the names — both
// shapes read naturally in Lua and authors reach for either. A missing or
// 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 {
// 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
}
return decStringList(map[string]interface{}{"v": raw}, "v")
}
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, field)
if L.Type(-1) == golua.LUA_TNIL {
L.SetTop(0)
return nil, false
}
raw := luaTableToGoValue(L, -1)
L.SetTop(0)
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 pg := decTable(m, "page"); pg != nil {
ui.Page = uipageFromValue(pg)
}
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) {
ps.mu.Lock()
ps.plugins = append(ps.plugins, p)
ps.mu.Unlock()
}
// rebuild recomputes the stage dispatch table and the merged UI payload. It is
// called after any load so the hot path (Fire) is a slice walk with no map
// lookups or locking beyond one RLock.
func (ps *Plugins) rebuild() {
ps.mu.Lock()
defer ps.mu.Unlock()
stageFuncs := map[Stage][]hookCall{}
for i, p := range ps.plugins {
// Disabled plugins are excluded from BOTH the dispatch table and the
// merged UI below. Including them in the UI would render a page whose
// refresh calls go to a plugin that is never consulted.
if p.LoadError != "" || p.Disabled {
continue
}
for _, st := range AllStages {
if fn, ok := p.Hooks[st]; ok {
stageFuncs[st] = append(stageFuncs[st], hookCall{pluginIdx: i, plugin: p.Info.Name, fn: fn})
}
}
}
ps.stageFuncs = stageFuncs
merged := &UIExtension{}
seenPage := map[string]bool{}
for _, p := range ps.plugins {
if p.LoadError != "" || p.Disabled || p.UI == nil {
continue
}
// The single `page` field is folded into the same list as `pages`.
// Assigning it to its own slot meant the LAST plugin to declare a page
// silently replaced every earlier one — a second plugin contributing a
// page erased the first from the sidebar with no error. One list, one
// rule: first writer wins per page_id (a duplicate id would collide in
// the DOM, and the plugin that loaded first is the better answer than
// whichever happened to load last).
var declared []*UIPage
if p.UI.Page != nil {
declared = append(declared, p.UI.Page)
}
declared = append(declared, p.UI.Pages...)
for _, pg := range declared {
if pg == nil || seenPage[pg.PageID] {
continue
}
seenPage[pg.PageID] = true
merged.Pages = append(merged.Pages, pg)
}
merged.Elements = append(merged.Elements, p.UI.Elements...)
}
// Stable order by declared Order then page id, so the sidebar does not
// reshuffle when a plugin is reloaded.
sort.SliceStable(merged.Pages, func(i, j int) bool {
if merged.Pages[i].Order != merged.Pages[j].Order {
return merged.Pages[i].Order < merged.Pages[j].Order
}
return merged.Pages[i].PageID < merged.Pages[j].PageID
})
ps.ui.Store(merged)
}
// DiskEntry is one .lua file in the plugin directory, whether or not it loaded.
// The management UI lists these rather than only the loaded set, so an operator
// can see (and fix) a plugin that failed to compile instead of finding it
// missing from the list.
type DiskEntry struct {
Name string `json:"name"`
Path string `json:"path"`
Size int64 `json:"size"`
Loaded bool `json:"loaded"`
Disabled bool `json:"disabled"`
Builtin bool `json:"builtin"`
Error string `json:"error,omitempty"`
// Version/Description are read from the loaded plugin when available.
Version string `json:"version,omitempty"`
Description string `json:"description,omitempty"`
Hooks int `json:"hooks"`
// Pages lists the UI page ids the plugin contributes. The desktop panel
// shows it so an operator can tell at a glance that a plugin actually
// renders something, without opening its source.
Pages []string `json:"pages,omitempty"`
}
// pageNames returns the plugin's contributed UI page ids, de-duplicated.
// A plugin may declare `page` or `pages` or both, and the same id can appear
// twice; the loader's merge keeps the first, so this mirrors that order.
func pageNames(p *Plugin) []string {
if p == nil || p.UI == nil {
return nil
}
declared := make([]*UIPage, 0, len(p.UI.Pages)+1)
if p.UI.Page != nil {
declared = append(declared, p.UI.Page)
}
declared = append(declared, p.UI.Pages...)
var out []string
seen := map[string]bool{}
for _, pg := range declared {
if pg == nil || seen[pg.PageID] {
continue
}
seen[pg.PageID] = true
out = append(out, pg.PageID)
}
return out
}
// OnDisk lists every .lua file in the plugin directory with its load state.
func (ps *Plugins) OnDisk() []DiskEntry {
out := []DiskEntry{}
if ps.dir == "" {
return out
}
entries, err := os.ReadDir(ps.dir)
if err != nil {
return out
}
for _, e := range entries {
if e.IsDir() || !strings.HasSuffix(e.Name(), ".lua") {
continue
}
name := strings.TrimSuffix(e.Name(), ".lua")
info, _ := e.Info()
de := DiskEntry{Name: name, Path: e.Name()}
if info != nil {
de.Size = info.Size()
}
ps.mu.RLock()
if p := ps.findLocked(name); p != nil {
de.Loaded = p.LoadError == ""
de.Disabled = p.Disabled
de.Builtin = p.Builtin
de.Error = p.LoadError
de.Version = p.Info.Version
de.Description = p.Info.Description
de.Hooks = len(p.Hooks)
// A plugin may declare a single page or several; both forms end
// up in Pages (see the merge at load time), so reporting that one
// field covers either.
de.Pages = pageNames(p)
} else {
// On disk but not in the running set: either it failed so badly
// that LoadSource never produced a record, or the dir was written
// after startup. Mark it by comparing with the bundled source.
if code, err := os.ReadFile(filepath.Join(ps.dir, e.Name())); err == nil {
if orig, err := ReadBundledPlugin(name); err == nil && orig == string(code) {
de.Builtin = true
}
}
}
ps.mu.RUnlock()
out = append(out, de)
}
sort.Slice(out, func(i, j int) bool {
// builtins first, then alphabetical: the example plugin an operator
// is most likely to want to read should not be buried under whatever
// they installed most recently.
if out[i].Builtin != out[j].Builtin {
return out[i].Builtin
}
return out[i].Name < out[j].Name
})
return out
}
// Count returns how many plugins loaded (including ones with LoadError).
func (ps *Plugins) Count() int {
ps.mu.RLock()
defer ps.mu.RUnlock()
return len(ps.plugins)
}
// List returns a JSON-friendly view of every loaded plugin, for the status API.
func (ps *Plugins) List() []map[string]interface{} {
ps.mu.RLock()
defer ps.mu.RUnlock()
out := make([]map[string]interface{}, 0, len(ps.plugins))
for _, p := range ps.plugins {
stages := make([]string, 0, len(p.Hooks))
for _, st := range AllStages {
if _, ok := p.Hooks[st]; ok {
stages = append(stages, string(st))
}
}
row := map[string]interface{}{
"name": p.Info.Name,
"version": p.Info.Version,
"description": p.Info.Description,
"author": p.Info.Author,
"hooks": stages,
"loaded": p.LoadError == "",
"disabled": p.Disabled,
"builtin": p.Builtin,
}
if p.LoadError != "" {
row["error"] = p.LoadError
}
if p.UI != nil {
ui := map[string]interface{}{}
// Every contributed page, listed under `pages`, INCLUDING the single
// `page` one. Two keys for the same thing looks redundant until you
// try to render the count: the management panel asked for `pages`,
// got an empty list for a plugin that plainly has a page, and showed
// "contributes 0 pages" next to a working screen. `page` stays as the
// backwards-compatible single-page accessor.
ids := make([]string, 0, len(p.UI.Pages)+1)
if p.UI.Page != nil && p.UI.Page.PageID != "" {
ids = append(ids, p.UI.Page.PageID)
}
for _, pg := range p.UI.Pages {
if pg != nil && pg.PageID != "" {
ids = append(ids, pg.PageID)
}
}
if len(ids) > 0 {
ui["pages"] = ids
if p.UI.Page != nil {
ui["page"] = p.UI.Page.PageID
}
}
if len(p.UI.Elements) > 0 {
ui["elements"] = len(p.UI.Elements)
}
row["ui"] = ui
}
out = append(out, row)
}
return out
}
// HookErrors returns per-stage hook failure counts (empty when all is well).
func (ps *Plugins) HookErrors() map[string]map[string]interface{} {
return ps.hookErr.snapshot()
}
// UI returns the merged UI extensions to inject into the WebUI.
// audienceAllows reports whether a role is permitted by an audience string.
//
// The grammar is deliberately small, because it is authored in two places (a
// .lua file and config.yaml) and a richer language would only give someone a way
// to express a rule nobody audits:
//
// "" or "all" everyone
// "admin" admin role only
// "user" user role only
// "admin,user" either
//
// "*" 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 {
return ParseAccessRule(want).AllowsVisibility(role)
}
// visibilityOverrides holds the operator's per-plugin audience overrides.
//
// A plugin declares its own default and the operator may force one. Both sides
// need to be able to: the plugin author does not know the deployment's policy,
// and the operator must not have to edit a bundled .lua — which the next
// upgrade would overwrite anyway.
type visibilityOverrides map[string]string
// seedShardConfig copies the shared plugin's `prices` table onto a shard VM.
//
// `prices` is CONFIGURATION, as opposed to `state` (accumulated history). A shard
// must not write state — its copy is thrown away — but it must be able to READ
// the operator's configuration, or a plugin that branches on its configuration
// behaves one way through Plugins.Call and another through ShardCall.
//
// Found by the OpenDesign adapter, which reads prices.control_key: without this, a
// fanout probe saw an empty table, concluded "not configured", and reported a
// skip for every host — while the identical call through the shared VM worked.
// The page uses the fanout path, so the feature would have been dead in
// production and alive in every test that called the shared VM.
//
// Called on every checkout (boot and reuse), so a configuration change reaches
// pooled VMs too. The push is a copy: shards get the same values, never a live
// reference to the shared table.
func (ps *Plugins) seedShardConfig(name string, w *worker) {
if ps == nil || w == nil || w.L == nil {
return
}
ps.mu.RLock()
p := ps.find(name)
ps.mu.RUnlock()
if p == nil {
return
}
p.mu.Lock()
if p.state == nil || p.state.L == nil || p.state.L == w.L {
p.mu.Unlock()
return
}
prices := snapshotField(p.state.L, "prices")
p.mu.Unlock()
if prices == nil {
return
}
L := w.L
L.SetTop(0)
defer L.SetTop(0)
L.GetGlobal(pluginGlobal)
if L.IsNil(-1) {
return
}
plug := L.GetTop()
pushGoValue(L, prices)
L.SetField(plug, "prices")
}
// SetPluginVisibility records an operator override for a plugin's audience. An
// empty value clears the override and restores the plugin's own default.
func (ps *Plugins) SetPluginVisibility(plugin, audience string) {
ps.mu.Lock()
defer ps.mu.Unlock()
if ps.visibility == nil {
ps.visibility = visibilityOverrides{}
}
plugin = strings.TrimSpace(plugin)
audience = strings.TrimSpace(audience)
if audience == "" {
delete(ps.visibility, plugin)
} else {
ps.visibility[plugin] = audience
}
ps.saveAccessOverridesLocked()
}
// VisibilityOverride returns the operator's audience override for a plugin,
// or "" when none is set. Exposed so /api/plugins can show which visibility is
// imposed rather than only the one in effect.
func (ps *Plugins) VisibilityOverride(plugin string) string {
ps.mu.RLock()
defer ps.mu.RUnlock()
return ps.visibility[plugin]
}
// uiAudience resolves the audience in force for a plugin: the operator's
// override if set, else the plugin's declared default.
func (ps *Plugins) uiAudience(p *Plugin, declared string) string {
ps.mu.RLock()
override := ps.visibility[p.Info.Name]
ps.mu.RUnlock()
if override != "" {
return override
}
return declared
}
// UIForRole returns the UI a given role may see.
//
// Filtering happens HERE, not in the HTTP layer, so every consumer sees the same
// view — the WebUI boot payload, the Electron shell and the tests. Filtering at
// one call site means the others quietly keep serving everything.
//
// A plugin's page/element can narrow visibility further than its own default
// (a public summary page beside an admin-only one), so the role check runs on
// the resolved plugin audience AND on each item's own declaration.
func (ps *Plugins) UIForRole(role string) *UIExtension {
full := ps.UI()
if full == nil {
return &UIExtension{}
}
// An admin sees everything, including anything a plugin hides from itself
// by mistake: anything else makes an override unverifiable from the browser.
if role == "admin" {
return full
}
out := &UIExtension{}
// The plugin-level default is per plugin, so pages are filtered by walking
// plugins rather than pages: a page id alone does not say who owns it.
for _, p := range ps.plugins {
if p.UI == nil || p.LoadError != "" {
continue
}
aud := ps.uiAudience(p, p.UI.Visibility)
if !audienceAllows(aud, role) {
continue
}
for _, pg := range p.UI.Pages {
if pg == nil || !audienceAllows(pg.Audience, role) {
continue
}
out.Pages = append(out.Pages, pg)
}
for _, el := range p.UI.Elements {
if !audienceAllows(el.Audience, role) {
continue
}
out.Elements = append(out.Elements, el)
}
// Page (singular) is the other half of the same declaration and both
// merge in UI(). Handling only Pages made every plugin that used the
// single-page form disappear for a user key — silently, with no error,
// because "this plugin contributes nothing to you" is indistinguishable
// from "this plugin is not installed".
if p.UI.Page != nil && audienceAllows(p.UI.Page.Audience, role) {
if out.Page == nil {
out.Page = p.UI.Page
}
if !containsPage(out.Pages, p.UI.Page.PageID) {
out.Pages = append(out.Pages, p.UI.Page)
}
}
}
return out
}
func (ps *Plugins) UI() *UIExtension {
if u := ps.ui.Load(); u != nil {
return u
}
return &UIExtension{}
}
// State returns a plugin's own published state. A plugin publishes it by
// setting `plugin.state = {...}` inside its hook; that is the only way a hook's
// numbers reach the WebUI, because request_end is the LAST pipeline stage and
// has no downstream consumer to hand a return value to.
//
// Returns nil when the plugin does not exist or has not published anything.
func (ps *Plugins) State(name string) interface{} {
ps.mu.RLock()
p := ps.find(name)
ps.mu.RUnlock()
if p == nil {
return nil
}
p.mu.Lock()
w := p.state
if w == nil {
p.mu.Unlock()
return nil
}
defer p.mu.Unlock()
L := w.L
L.SetTop(0)
defer L.SetTop(0)
L.GetGlobal(pluginGlobal)
if L.IsNil(-1) {
return nil
}
plug := L.GetTop()
L.GetField(plug, "state")
if L.Type(-1) == golua.LUA_TNIL {
return nil
}
var out interface{}
if err := luaToJSON(L, -1, &out); err != nil {
return nil
}
return out
}
// Prices returns the plugin's current price table (the `prices` field, kept
// separate from `state`).
//
// Exposed because the billing rules editor has to show what is actually in
// effect. Reconstructing it from config alone would be wrong the moment a
// price was injected directly through SetState or the two drifted, and an
// editor that displays a different table from the one billing uses is worse
// than no editor at all.
func (ps *Plugins) Prices(name string) interface{} {
ps.mu.RLock()
p := ps.find(name)
ps.mu.RUnlock()
if p == nil {
return nil
}
p.mu.Lock()
defer p.mu.Unlock()
if p.state == nil || p.state.L == nil {
return nil
}
return snapshotField(p.state.L, "prices")
}
// SetState replaces a plugin's published state (admin API). It is how a
// configuration change (a new price for a model) reaches the plugin without
// reloading it.
//
// CONTRACT: a `prices` key in the payload is applied to the plugin's SEPARATE
// `prices` field and STRIPPED from `state`. That split is deliberate: prices are
// configuration while state is accumulated history, and a single replaceable
// field would make a price update wipe the totals (or make the totals carry a
// stale price table). A plugin that keeps its config elsewhere can ignore the
// convention and read the whole payload from `state` instead.
func (ps *Plugins) SetState(name string, state interface{}) error {
ps.mu.RLock()
p := ps.find(name)
ps.mu.RUnlock()
if p == nil {
return fmt.Errorf("plugin %s not loaded", name)
}
p.mu.Lock()
w := p.state
if w == nil {
p.mu.Unlock()
return fmt.Errorf("plugin %s has no state", name)
}
defer p.mu.Unlock()
L := w.L
L.SetTop(0)
defer L.SetTop(0)
L.GetGlobal(pluginGlobal)
if L.IsNil(-1) {
return fmt.Errorf("plugin %s has no table", name)
}
plug := L.GetTop()
// Pull `prices` out of the payload before storing the rest as state.
body := state
if m, ok := state.(map[string]interface{}); ok {
if prices, has := m["prices"]; has {
pushGoValue(L, prices)
L.SetField(plug, "prices")
rest := make(map[string]interface{}, len(m))
for k, v := range m {
if k != "prices" {
rest[k] = v
}
}
if len(rest) == 0 {
// A prices-ONLY payload is a configuration change, not a state
// reset. Leaving `state` untouched is what makes repricing safe:
// replacing it with an empty table would silently erase every
// accumulated total, so the next request would start from zero
// and the dashboard would show a sudden drop in spend.
//
// The prices still have to be persisted here: returning before
// the write below would leave the new price table in memory
// only, and a restart would silently revert to the old prices
// while the operator believed the change took effect.
var curState, curPrices interface{}
curState = snapshotField(L, "state")
curPrices = snapshotField(L, "prices")
if ps.dir != "" && !ps.stateDirDisabled && curState != nil {
writeJSONAtomic(ps.stateFile(name), map[string]interface{}{
"version": 1,
"state": curState,
"prices": curPrices,
})
}
return nil
}
body = rest
}
}
pushGoValue(L, body)
L.SetField(plug, "state")
L.SetTop(0)
// A state replacement is exactly the kind of thing an operator restarts the
// gateway for, so it must survive the restart. The snapshot is taken from
// the values just written — the caller already holds p.mu, so calling
// persistNow here would deadlock on that same non-reentrant mutex.
writeJSONAtomic(ps.stateFile(name), map[string]interface{}{
"version": 1,
"state": body,
"prices": readPricesFromPayload(state),
})
return nil
}
// readPricesFromPayload recovers the prices table a caller passed to SetState,
// which SetState stores on the plugin (not in state). Used only for the
// persistence record, so that a restart restores configuration and history to
// the two fields they belong in rather than collapsing them.
func readPricesFromPayload(state interface{}) interface{} {
if m, ok := state.(map[string]interface{}); ok {
return m["prices"]
}
return nil
}
// markDirty snapshots a plugin's state and hands it to the saver.
//
// It is called from the hook path, so it must not block on file I/O — that is
// the saver's job. It DOES walk the Lua tables, because that has to happen
// while the caller still holds p.mu and the VM is guaranteed alive; the flush
// (Snapshotting in the flush goroutine instead was a use-after-free: vm.Stop()
// frees the Lua states. See persistByName for why the walk is safe here.)
//
// Cost is one JSON conversion per request, which the billing plugin would pay
// anyway inside its own hook.
func (ps *Plugins) markDirtyLocked(p *Plugin) {
if ps == nil || p == nil || ps.stateDirDisabled || p.state == nil || ps.dir == "" {
return
}
// NOTE: this only records THAT the state changed. It deliberately does not
// snapshot it.
//
// Snapshotting here cost 264us per request — the snapshot walks the plugin's
// Lua tables through luaValueToGo + json.Marshal + json.Unmarshal (three
// conversions), and the billing plugin does it on every single request. That
// turned a 14.6us hook into a 385us one and put the gateway's hot path
// behind a bookkeeping step. The saver pulls the snapshot on its own schedule
// instead, so N requests between flushes cost ONE snapshot.
ps.save.markDirty(p.Info.Name)
}
// persistByName snapshots one plugin and writes its state file.
//
// It is called from the saver goroutine, which means it walks a Lua state that
// vm.Stop() may be about to free. That was a real crash: the first version had
// the saver read the tables itself and the process died with SIGSEGV inside
// golua once shutdown raced a flush.
//
// Two things make it safe now:
// - Plugins.Close() stops the saver AND waits for its final flush BEFORE the
// caller stops the VM (see Plugins.Close), so no flush can be in flight when
// the Lua states go away.
// - p.mu is held while the tables are read, so a hook cannot be mutating them
// underneath. A hook that arrives after the VM stopped cannot run either,
// because the request path is already torn down by then.
func (ps *Plugins) persistByName(name string) {
if ps == nil || ps.stateDirDisabled || ps.dir == "" {
return
}
ps.mu.RLock()
p := ps.find(name)
ps.mu.RUnlock()
if p == nil || p.state == nil {
return
}
ps.persistNow(p)
}
// ---------- state persistence ----------
//
// A plugin's state lives in the Lua VM, which dies with the process. For the
// billing plugin that means a restart silently zeroes every total — measured on
// this gateway: 218,241 requests and 26.2M prompt tokens gone after one
// `systemctl restart`. For a REPORTING plugin whose whole purpose is the number
// it accumulates, that is not a rounding error, it is the feature not working.
//
// Three rules shape this:
//
// 1. `prices` (configuration) and `state` (accumulated history) are persisted
// to SEPARATE files. Restoring them together would let a price edit look
// like a state reset, or a state restore resurrect stale prices — SetState
// already keeps them apart in memory, and disk has to agree.
// 2. Writes are atomic (temp file + rename). A crash mid-write must leave the
// previous state readable, not a truncated JSON file that fails to parse on
// the next boot and loses the total anyway.
// 3. A corrupt or unreadable state file is a WARNING, never a startup error.
// Forwarding must not depend on a plugin's bookkeeping surviving.
// stateFile is the per-plugin state path, kept beside the plugin source so an
// operator can find (and delete) it next to the plugin it belongs to.
func (ps *Plugins) stateFile(name string) string {
return filepath.Join(ps.dir, "."+name+".state.json")
}
// persistNow writes a plugin's state and prices to disk synchronously. Callers
// on the hot path must use ps.markDirtyLocked instead; this is the flush worker and
// the admin-state path, where durability matters more than latency.
//
// It is called after every
// state mutation, so it must be cheap enough not to matter: the billing plugin
// mutates on every request, and writing the whole state per request would put a
// file write on the hot path.
func (ps *Plugins) persistNow(p *Plugin) {
if ps == nil || p == nil || ps.stateDirDisabled || p.state == nil || ps.dir == "" {
return
}
p.mu.Lock()
state, prices := readStateAndPrices(p.state.L)
p.mu.Unlock()
if state == nil {
return
}
writeJSONAtomic(ps.stateFile(p.Info.Name), map[string]interface{}{
"version": 1,
"state": state,
"prices": prices,
})
}
// readStateAndPrices pulls both tables out of the Lua state. The caller holds
// p.mu. Returns nils when the plugin table is missing, which happens for a
// plugin that failed to compile.
//
// Each field is read by REBUILDING the stack from the global, because the
// conversion helper (luaToJSON) ends with L.SetTop(0) on every path — it treats
// the whole stack as its own. Reusing a saved index across two conversions
// addresses a slot that no longer exists, and this binding aborts the process
// (SIGABRT) instead of reporting a bad index. The first version did exactly
// that and crashed inside the Lua C layer on every hook call.
func readStateAndPrices(L *golua.State) (state, prices interface{}) {
state = snapshotField(L, "state")
prices = snapshotField(L, "prices")
L.SetTop(0)
return state, prices
}
// snapshotField converts plugin.<field> into Go values, leaving the stack clean.
// Caller holds p.mu; caller is responsible for the Lua state being alive.
func snapshotField(L *golua.State, field string) interface{} {
L.SetTop(0)
defer L.SetTop(0)
L.GetGlobal(pluginGlobal)
if L.IsNil(-1) {
return nil
}
L.GetField(L.GetTop(), field)
if L.Type(-1) != golua.LUA_TTABLE {
return nil
}
var out interface{}
if err := luaToJSON(L, -1, &out); err != nil {
return nil
}
return out
}
// writeJSONAtomic writes v to path via a temp file + rename, so a reader (or a
// crash) never observes a half-written file.
func writeJSONAtomic(path string, v interface{}) {
b, err := json.MarshalIndent(v, "", " ")
if err != nil {
return
}
dir := filepath.Dir(path)
if err := os.MkdirAll(dir, 0755); err != nil {
return
}
tmp, err := os.CreateTemp(dir, filepath.Base(path)+".tmp*")
if err != nil {
return
}
tmpName := tmp.Name()
if _, err := tmp.Write(b); err != nil {
tmp.Close()
os.Remove(tmpName)
return
}
if err := tmp.Close(); err != nil {
os.Remove(tmpName)
return
}
if err := os.Rename(tmpName, path); err != nil {
os.Remove(tmpName)
}
}
// restore re-applies a persisted state (and prices) onto a freshly compiled
// plugin. Called once per plugin after it compiles and registers its hooks.
//
// Ordering matters: this runs AFTER compilation so the plugin's own defaults
// exist, and it OVERWRITES them, so a restart continues from the saved totals
// rather than from the values baked into the .lua source.
func (ps *Plugins) restore(p *Plugin) {
if ps == nil || p == nil || p.state == nil || ps.stateDirDisabled {
return
}
b, err := os.ReadFile(ps.stateFile(p.Info.Name))
if err != nil {
return // no saved state yet: the compiled-in default stands
}
var saved struct {
Version int `json:"version"`
State interface{} `json:"state"`
Prices interface{} `json:"prices"`
}
if err := json.Unmarshal(b, &saved); err != nil {
// A corrupt file must not stop the gateway: the plugin keeps its
// compiled-in defaults and the operator sees a warning in the log.
ps.logf("plugin %s: ignoring unreadable state file %s: %v",
p.Info.Name, ps.stateFile(p.Info.Name), err)
return
}
p.mu.Lock()
defer p.mu.Unlock()
L := p.state.L
L.SetTop(0)
defer L.SetTop(0)
L.GetGlobal(pluginGlobal)
if L.IsNil(-1) {
return
}
plug := L.GetTop()
if saved.Prices != nil {
pushGoValue(L, saved.Prices)
L.SetField(plug, "prices")
}
if saved.State != nil {
pushGoValue(L, saved.State)
L.SetField(plug, "state")
}
p.persisted = true
}
// SetEnabled turns a plugin's dispatch on or off without touching its file.
//
// The state is on the Plugin record (not derived from disk) so a disable survives
// as long as the process lives and is trivially re-enabled; it deliberately does
// NOT persist across restarts, because a "disable" that silently outlives the
// operator's intent is its own surprise. An operator who wants it permanent
// moves the file out of the plugin dir.
func (ps *Plugins) SetEnabled(name string, enabled bool) error {
ps.mu.Lock()
p := ps.findLocked(name)
ps.mu.Unlock()
if p == nil {
return fmt.Errorf("plugin %s not loaded", name)
}
if p.LoadError != "" {
return fmt.Errorf("plugin %s failed to load (%s); fix the file before enabling it", name, p.LoadError)
}
p.mu.Lock()
p.Disabled = !enabled
p.mu.Unlock()
ps.rebuild()
return nil
}
// Enabled reports whether a plugin is currently dispatching.
func (ps *Plugins) Enabled(name string) bool {
ps.mu.RLock()
p := ps.findLocked(name)
ps.mu.RUnlock()
return p != nil && p.LoadError == "" && !p.Disabled
}
// Unload removes a plugin from the running set. Its states are closed so the
// memory goes back; a subsequent LoadSource with the same name works again.
func (ps *Plugins) Unload(name string) error {
ps.mu.Lock()
idx := -1
for i, p := range ps.plugins {
if p.Info.Name == name || strings.HasSuffix(filepath.Base(p.dir), name+".lua") {
idx = i
break
}
}
if idx < 0 {
ps.mu.Unlock()
return fmt.Errorf("plugin %s not loaded", name)
}
p := ps.plugins[idx]
ps.plugins = append(ps.plugins[:idx], ps.plugins[idx+1:]...)
ps.mu.Unlock()
// Stop the scheduler BEFORE touching the state. Stop() closes the loop and
// waits for it, so after it returns no NEW task can be dispatched — the set
// of in-flight tasks is now frozen, which is what makes the check below
// meaningful rather than racy.
inFlight := false
if p.Runtime != nil {
p.Runtime.Stop()
inFlight = p.Runtime.InFlight()
p.Runtime = nil
}
// A task holds p.mu for the whole duration of its Lua call. If one is still
// in flight we must NOT take p.mu (a wedged task never releases it, so Lock
// would hang the unload forever) and must NOT close the state underneath it
// (that is a use-after-free in the Lua C layer, which aborts the process
// instead of returning an error). Leaking this one state is the bounded,
// non-crashing outcome; the operator's action is to remove the file and
// restart, which is what a plugin wedged by its own task requires anyway.
if inFlight {
log.Printf("[scheduler] plugin %s: unloaded while a task was still running; "+
"its Lua state is left open to avoid a use-after-free", name)
ps.rebuild()
return nil
}
// Lifecycle hook: if the plugin exported on_unload, call it BEFORE closing
// the state. This is generic plugin lifecycle (part of "调度插件"), not any
// feature: a plugin that spawned OS processes or holds other external
// resources is the one that knows how to release them, and Unload is the
// only guaranteed point where it still can. Best-effort by design — a
// hook that errors must not block an unload the operator explicitly asked
// for (a wedged plugin has its own path above).
p.mu.Lock()
if p.state != nil && p.state.L != nil {
p.state.L.GetGlobal(pluginGlobal)
if L := p.state.L; L.Type(-1) == golua.LUA_TTABLE {
L.GetField(-1, "on_unload")
if L.Type(-1) == golua.LUA_TFUNCTION {
if err := callFnOn(L, "on_unload"); err != nil {
log.Printf("[plugins] %s: on_unload: %v", name, err)
}
}
L.Pop(2) // adapter/plugin table + field remnant
} else {
L.Pop(1)
}
p.state.L.Close()
p.state = nil
}
p.mu.Unlock()
ps.rebuild()
return nil
}
// find returns a loaded plugin by declared name, taking the read lock.
func (ps *Plugins) find(name string) *Plugin {
ps.mu.RLock()
defer ps.mu.RUnlock()
return ps.findLocked(name)
}
// findLocked returns a loaded plugin by declared name. The caller holds ps.mu.
func (ps *Plugins) findLocked(name string) *Plugin {
for _, p := range ps.plugins {
if p.Info.Name == name {
return p
}
}
return nil
}
// Fire calls every plugin registered for a stage, in plugin load order.
//
// A plugin may MUTATE the payload by returning a JSON object: any keys it
// returns are merged into the payload for the next hook and returned to the
// caller. Returning nil or an empty table means "no opinion". This lets a
// plugin add fields (the billing plugin adds `cost_usd`) without the gateway
// having to know about them.
//
// Errors are contained: a plugin that throws is logged against the stage and
// skipped. Forwarding never depends on plugin health.
func (ps *Plugins) Fire(stage Stage, payload map[string]interface{}) map[string]interface{} {
ps.mu.RLock()
calls := ps.stageFuncs[stage]
ps.mu.RUnlock()
if len(calls) == 0 {
return payload
}
// Resolve the plugins and drop the lock BEFORE running any hook.
//
// A hook can install, disable or remove a plugin (the admin API is reachable
// from a hook that has the key), and those paths take ps.mu for write. Taking
// a per-iteration RLock like the single-plugin case did works but serializes
// on a shared cache line under load; a snapshot plus one lookup is cheaper and
// — more importantly — it means the parallel section below never holds ps.mu,
// so a hook cannot deadlock against a concurrent reload.
type target struct {
p *Plugin
hc hookCall
}
targets := make([]target, 0, len(calls))
for _, hc := range calls {
ps.mu.RLock()
p := ps.plugins[hc.pluginIdx]
ps.mu.RUnlock()
if p == nil || p.LoadError != "" {
continue
}
// A plugin disarmed by an overrunning task still holds its own lock (the
// task never returned), so dispatching to it would block THIS request
// behind a lock that is never released. Skip it and record why: the
// operator needs to see that inference traffic is being dropped and which
// plugin is responsible, not watch requests hang.
if p.Runtime != nil && p.Runtime.Disarmed() {
ps.hookErr.note(stage, p.Info.Name+": "+errPluginWedged.Error())
continue
}
targets = append(targets, target{p: p, hc: hc})
}
if len(targets) == 0 {
return payload
}
// One plugin is the overwhelmingly common case (a gateway with the bundled
// billing plugin has exactly one). Paying for goroutines there would be pure
// overhead, so it takes the direct path.
if len(targets) == 1 {
if out := ps.runHook(stage, targets[0].p, targets[0].hc.fn, payload); len(out) > 0 {
for k, v := range out {
payload[k] = v
}
}
return payload
}
// Parallel across plugins. Each hook receives the SAME payload snapshot, and
// the returned tables are merged afterwards IN PLUGIN LOAD ORDER, so the
// documented contract ("a returned table's keys are merged into the payload")
// still holds deterministically.
//
// What changes: a hook no longer sees the keys another hook just added. The
// sequential behaviour made that possible, and docs/plugins.md described it
// ("the payload is passed to the next plugin unchanged"). It was never used —
// the bundled billing plugin returns nil on every stage, with a comment saying
// nobody downstream reads it — but it IS a contract change and is called out
// there rather than left as a surprise.
//
// Why this is safe for the per-plugin lock: each plugin has its own Lua state
// and its own p.mu, so two plugins never touch the same state. What is shared
// is the payload, and it is only READ here — merging happens after every hook
// has returned, on the caller's goroutine.
outs := make([]map[string]interface{}, len(targets))
var wg sync.WaitGroup
for i := range targets {
wg.Add(1)
go func(i int) {
defer wg.Done()
// A hook that panics would otherwise take the whole process with it,
// and in the sequential version a panic could not escape Fire either.
// recover() here restores that: the plugin is skipped and noted.
defer func() {
if rec := recover(); rec != nil {
ps.hookErr.note(stage, fmt.Sprintf("%s: panic in hook: %v", targets[i].hc.plugin, rec))
}
}()
outs[i] = ps.runHook(stage, targets[i].p, targets[i].hc.fn, payload)
}(i)
}
wg.Wait()
// Merge in load order so the result does not depend on goroutine scheduling.
// A later plugin's value wins on a key collision, exactly as it did when the
// hooks ran one after another.
for _, out := range outs {
for k, v := range out {
payload[k] = v
}
}
return payload
}
// runHook invokes one hook and folds its returned table into payload. It
// contains the plugin's error handling so both the sequential and the parallel
// path behave identically on failure.
func (ps *Plugins) runHook(stage Stage, p *Plugin, fn string, payload map[string]interface{}) map[string]interface{} {
out, err := ps.invoke(p, fn, payload)
if err != nil {
ps.hookErr.note(stage, p.Info.Name+": "+err.Error())
return nil
}
return out
}
// invoke runs one plugin hook on that plugin's own state, under its pool's
// concurrency cap. The plugin's returned table is re-fetched each call because
// the pool is elastic: a plugin may have several states, and the hook function
// lives in each.
func (ps *Plugins) invoke(p *Plugin, fn string, payload map[string]interface{}) (map[string]interface{}, error) {
p.mu.Lock()
w := p.state
if w == nil {
p.mu.Unlock()
return nil, fmt.Errorf("no state for plugin %s", p.Info.Name)
}
defer p.mu.Unlock()
L := w.L
L.SetTop(0)
defer L.SetTop(0)
L.GetGlobal(pluginGlobal)
if L.IsNil(-1) {
return nil, fmt.Errorf("plugin table missing")
}
plug := L.GetTop() // absolute, so nothing below shifts
// The hook is called THROUGH a pcall guard (see hookGuardSrc) so a script
// error returns as values instead of raising into golua, whose error path
// SIGSEGVs the process. Desired stack before the payload push:
// [pluginGlobal, guard, hookfn]. Build it in that order — GetGlobal(guard)
// then GetField(hookfn) lands the function exactly above the guard, with no
// Remove/Insert juggling. (The first version reordered with Remove/Insert
// and ended up calling pluginGlobal as if it were the hook, producing
// "attempt to call a table value" and silently zeroed every total.)
L.GetGlobal(hookGuardName)
if L.IsNil(-1) {
// Guard absent (a state built before this existed): drop the nil so the
// stack is [pluginGlobal, hookfn] and the hook is called directly.
// Correct plugins still work; only their errors stop being survivable.
L.Pop(1)
}
L.GetField(plug, fn)
if !L.IsFunction(-1) {
L.SetTop(0)
return nil, fmt.Errorf("hook %s missing", fn)
}
// The hook is called with a DECODED table, not the raw JSON string: the
// adapter protocol passes JSON text to its transforms (they decode it
// themselves), but a plugin hook receives a table so it can read
// payload.model directly. Passing the string made every hook fail with
// "attempt to index local 'payload' (a string value)".
//
// Decoding is done WITHOUT JSON. json.Marshal + json.Unmarshal here cost
// 9.6us of the 14.6us hook — 67% of the call spent re-deriving a tree that
// pushGoValue walks natively anyway. plainForLua does the same conversion
// by type-switching, and falls back to JSON only for a type it does not
// model, so an exotic payload still arrives instead of vanishing.
pushGoValue(L, plainForLua(payload))
if err := L.Call(2, 2); err != nil {
return nil, err
}
// The guard returns TWO values: the hook's value, then an error string
// (nil on success). Drop the error slot unconditionally so the value is
// left on top. Peeking instead of popping was the first bug here: on a
// SUCCESSFUL call the top is nil, so "the error is non-nil" is false, the
// pop was skipped, and the code then read the (always-present) nil error as
// if it were the hook's answer. Every plugin's returned table silently
// became "no opinion" while the tests that only checked totals kept
// passing.
errMsg := ""
if L.GetTop() >= 2 {
if !L.IsNil(-1) {
errMsg = L.ToString(-1)
}
L.Pop(1) // the error slot, nil or not
}
if errMsg != "" {
// A hook that raised partway may already have mutated plugin.state, so
// mark it dirty before bailing: the accounting it managed to do is
// still real and should be persisted.
ps.markDirtyLocked(p)
return nil, fmt.Errorf("plugin %s: %s", p.Info.Name, errMsg)
}
// Read the return value FIRST, then snapshot state for persistence.
//
// The order is load-bearing. markDirtyLocked walks the Lua tables and resets
// the stack, so calling it before the return value is read destroyed the
// hook's answer — every plugin that returned a table silently became a
// plugin that "had no opinion". Reading first costs nothing and keeps the
// documented merge contract intact.
var out map[string]interface{}
if L.GetTop() < 1 || L.IsNil(-1) {
ps.markDirtyLocked(p)
return nil, nil
}
if err := luaToJSON(L, -1, &out); err != nil {
ps.markDirtyLocked(p)
return nil, nil // not a table: treat as "no opinion"
}
// A hook that ran at all may have mutated plugin.state, whether or not it
// returned anything. Marking here (not only on a returned table) is what
// makes an accumulating plugin like billing durable: its totals change on
// every call and it returns nil every time.
ps.markDirtyLocked(p)
return out, nil
}
// Call invokes a plugin function that the plugin declared in plugin.exports.
//
// The gate is the declaration, not the caller: a plugin lists the handful of
// entry points its own UI is allowed to reach, and anything absent is refused.
// Without that list this endpoint would be a remote Lua evaluator — it could
// call the plugin's private helpers, hand them crafted arguments, and read
// whatever the plugin can see, which for a plugin holding an API key means
// handing that key to anyone who can POST to the gateway.
//
// The return value is JSON-shaped rather than restricted to a map, because a
// 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
for _, c := range ps.plugins {
if c.Info.Name == name {
p = c
break
}
}
ps.mu.RUnlock()
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)
}
p.mu.Lock()
defer p.mu.Unlock()
w := p.state
if w == nil {
return nil, fmt.Errorf("no state for plugin %s", name)
}
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.
//
// 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)
L.GetGlobal(pluginGlobal)
if L.IsNil(-1) {
return nil, fmt.Errorf("plugin table missing")
}
plug := L.GetTop()
L.GetGlobal(hookGuardName)
if L.IsNil(-1) {
L.Pop(1)
}
L.GetField(plug, fn)
if !L.IsFunction(-1) {
return nil, fmt.Errorf("export %s missing", fn)
}
if args == nil {
args = map[string]interface{}{}
}
pushGoValue(L, plainForLua(args))
if err := L.Call(2, 2); err != nil {
return nil, err
}
errMsg := ""
if L.GetTop() >= 2 {
if !L.IsNil(-1) {
errMsg = L.ToString(-1)
}
L.Pop(1)
}
if 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, err
}
return out, nil
}
// pluginExportAllowed reports whether fn is in the plugin's exports list.
// An absent list means nothing is callable — the safe reading, since silently
// allowing all would turn an undeclared plugin into a remote evaluator.
func pluginExportAllowed(p *Plugin, fn string) bool {
for _, e := range p.exports {
if e == fn {
return true
}
}
return false
}
// StateVisibleTo reports whether a role may read a plugin's published state.
//
// The audience is resolved exactly as for the UI: operator override first, then
// the plugin's declaration. A plugin with neither is readable by any
// authenticated role, which is the pre-existing behaviour and the reason a
// newly installed widget keeps working without an edit.
// HTTPHandler is one endpoint a plugin registers on the gateway's router.
//
// A plugin that only receives hooks (billing) has nothing to expose. A plugin
// that computes something the operator may want to query from OUTSIDE the
// browser — a health probe, a CI check, another tool — needs a route, and
// hard-coding one in the gateway would mean every such plugin needs a gateway
// change. So the plugin declares the path, the method and the role it requires,
// and the gateway mounts it.
//
// The handler is a Lua function taking (request) and returning (response). Both
// are plain tables: request carries method/path/query/headers/body, response
// carries status/headers/body/json. Nothing here can reach the Go side except
// through those two values, so the surface a plugin can express is bounded by
// what a table can say.
type HTTPHandler struct {
// Path is the route suffix under /api/plugins/http/. It may contain ":name"
// segments, captured into request.params.
Path string `json:"path"`
// Method is the HTTP method; "" or GET. Only GET/POST/PUT/DELETE/PATCH are
// accepted — an arbitrary verb would need a switch in the router for no gain.
Method string `json:"method,omitempty"`
// Roles are the API-key roles allowed to call it. EMPTY MEANS EVERYONE, which
// is the wrong default for an endpoint a plugin invented: an unauthenticated
// internet scan should not be able to ask it questions. RequireRole exists
// so a plugin that really wants public access has to say so.
Roles []string `json:"roles,omitempty"`
// Keys names SPECIFIC gateway keys allowed to call it, by their configured
// name (the `name:` on a gateway_keys entry), never by the secret.
//
// Names rather than secrets for three reasons: a secret is rotated without
// notice and a rule written against it would silently stop matching; putting
// secrets in a plugin file scatters credentials across the install; and the
// operator reads names. A key with no name cannot be named here.
//
// When Keys is non-empty it is the WHOLE rule, not an addition to Roles. An
// OR would mean an operator writing a two-key allowlist had also granted
// every user key, which is the opposite of what they meant.
Keys []string `json:"keys,omitempty"`
// RequireAuth defaults to true in practice: the gateway refuses a handler
// with neither Roles nor an explicit opt-out.
Public bool `json:"public,omitempty"`
// Handler is the name of the plugin function to invoke.
Handler string `json:"handler"`
// Description is shown in /api/plugins so an operator can see what a plugin
// has exposed without reading the .lua.
Description string `json:"description,omitempty"`
// Plugin is filled in by HTTPHandlers: which plugin registered this route.
// The gateway dispatcher needs it to know which Lua state to enter, and
// deriving it from the path would mean trusting the plugin's own prefix.
Plugin string `json:"-"`
}
// routePath renders the handler's full mount path.
func (h HTTPHandler) routePath() string {
return "/api/plugins/http/" + strings.Trim(h.Path, "/")
}
// allows reports whether a caller may invoke this handler.
//
// 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 {
return accessRuleFor(h.Roles, h.Keys, h.Public).AllowsCapability(role, keyName)
}
func (ps *Plugins) StateVisibleTo(name, role string) bool {
if role == "admin" {
return true
}
ps.mu.RLock()
defer ps.mu.RUnlock()
for _, p := range ps.plugins {
if p.Info.Name != name {
continue
}
override := ps.visibility[name]
if override != "" {
return audienceAllows(override, role)
}
declared := ""
if p.UI != nil {
declared = p.UI.Visibility
}
return audienceAllows(declared, role)
}
// An unknown plugin name: no visibility can be established, and this endpoint
// answers with an empty state rather than an error, so refusing here would
// change behaviour for a name that may simply not be installed.
return true
}
func containsPage(pages []*UIPage, id string) bool {
for _, p := range pages {
if p != nil && p.PageID == id {
return true
}
}
return false
}
// readPluginHTTP collects plugin.http, the endpoints the plugin asks the gateway
// to mount.
//
// A malformed entry is DROPPED rather than failing the plugin: a plugin with one
// bad route should still load its hooks and pages, and the bad route simply does
// 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 {
// `http` is a LIST of declarations, each naming the handler function by
// STRING:
//
// 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 {
h, ok := it.(map[string]interface{})
if !ok {
continue
}
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"),
}
// 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 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. 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
}
// uipageFromValue builds a UIPage from a decoded Lua table.
func uipageFromValue(m map[string]interface{}) *UIPage {
pg := &UIPage{
PageID: asString(m["page_id"]),
Title: asString(m["title"]),
Icon: asString(m["icon"]),
Mount: asString(m["mount"]),
Audience: asString(m["audience"]),
}
if n, ok := asFloat(m["order"]); ok {
pg.Order = int(n)
}
return pg
}