Files
ModelRouter/internal/lua/vm.go
root 2113bfc125 fix(channels): 四渠道修复 —— qoder 裸 token 领取 / trae+buddy 非流式 / opencode 目录 / 隐藏适配器
生产实测逐渠道排查(2026-10-19):

* qoder:PKCE 设备码流无 register 第二跳,落库凭据是裸 token 字符串;
  qoderClaim 硬要求结构化 table 导致「缺少凭据对象」(而 account_test
  走 token 路径正常 —— 自相矛盾状态)。改为非空裸 tok 就地包
  {access_token=tok};补 TestQoderClaimAcceptsBareTokenCredential。

* trae(SOLO 自装适配器):上游对非流式请求也回 SSE(event: 前缀帧),
  transform_response 不识别 → 整个 body 原样透传 → Go unmarshal 报
  invalid character 'e'。补 aggregate_solo_sse 聚合。

* buddy(CodeBuddy):上游 400 11101 只支持流式。transform_request 强制
  stream=true;transform_response 补 SSE 聚合(tool_calls 按 index 拼分片)。

* provider 流式:真实 tool-call 片段之后出现 finish_reason=stop 时归一为
  tool_calls(trae 实测如此发);buddy/trae 不再序列化空 tool_calls 表。

* opencode:远端目录端点改为 /zen/v1/models(裸 /v1/models 回 HTML);
  兜底目录同步 2026-10-19 实测(space-bunny-free 410 下架等 5 项变化)。
  opencodezen 适配器 v1.1.0:会话亲和(client_session / 首 user 消息指纹
  分桶),保留免费层门禁四要素。

* 内核:适配器可声明 adapter.hidden = true 退出用户可见列表;被源引用的
  隐藏适配器保持可见(源编辑器要回显当前值)。opencodezen 默认隐藏。

* channels UI:切渠道立即清空账号/模型区 + 响应代际校验,修复 qoder 账号
  闪现在 opencode 卡片下的渲染竞态。

* channels request_end hook:挂接丢失的 isModelGone/recordDeadModel ——
  真实 404(如 codearts GLM-5.3 "The model is not registered")现在会进
  deadModels,不再每轮 404 仍可选中。
2026-10-19 22:32:18 +08:00

1738 lines
55 KiB
Go
Raw Permalink 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 implements the adapter runtime: bundles/loads *.lua adapter
// scripts, exposes json/string helpers, and lets Go call the protocol
// transform functions plus a build_headers signature hook.
//
// The runtime uses LuaJIT (github.com/aarzilli/golua) so every adapter runs on
// an independent interpreter state. Because a lua.State is NOT goroutine-safe,
// each adapter owns a pool of workers (states) and a worker is checked out for
// the duration of a single hook call.
//
// Pools are ELASTIC: the sum of max_concurrent over the sources using an adapter
// is a ceiling, not a preallocation. States are booted on demand in batches
// sized by that ceiling, and reclaimed by a single background janitor at a rate
// scaled inversely to the adapter's live connection count. An idle gateway
// therefore holds close to zero Lua states regardless of how much aggregate
// concurrency is configured. See adapterPool for the sizing rules.
package lua
import (
"crypto/hmac"
"crypto/sha256"
"embed"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
"time"
// golua exposes the LuaJIT C API. Build with -tags luajit so the cgo
// LDFLAGS resolve to -lluajit-5.1 (see golua's lua.go).
golua "github.com/aarzilli/golua/lua"
)
//go:embed adapters/*.lua plugins/*.lua
var bundledAdapters embed.FS
// adapterGlobal is the reserved global holding the adapter table after the
// worker boots the script. A reserved name avoids clashing with user code.
const adapterGlobal = "__llmsproxy_adapter"
type APIAdapter struct {
Name string `json:"name"`
Version string `json:"version"`
}
// worker wraps one LuaJIT interpreter state. Not goroutine-safe; each worker
// is used by exactly one goroutine at a time, checked out from its pool.
type worker struct {
L *golua.State
}
func (w *worker) close() {
if w.L != nil {
w.L.Close()
}
}
// staticInfo caches the immutable fields of an adapter script, extracted once
// at load time. Endpoint() and the static headers fallback read from here so
// they never churn a pooled worker.
type staticInfo struct {
name string
version string
endpoint string
// endpointStream is the streaming counterpart of endpoint. It exists so the
// KERNEL never has to know a vendor's streaming path: Gemini streams over the
// same path with a different verb (:generateContent -> :streamGenerateContent),
// and the verb is part of the path, so the two cannot both be static.
//
// Before this field the kernel carried `strings.Replace(ep,
// ":generateContent", ":streamGenerateContent", 1)` — vendor protocol
// knowledge inside the dispatcher. It now lives here, where it belongs:
// the adapter declares both endpoints and the kernel just picks one.
// An adapter that omits endpoint_stream uses endpoint for both (most do).
endpointStream string
headers map[string]string
// registeredAs is the key the adapter was LOADED under, which can differ
// from name when a script declares its own name. Kept so a reload replaces
// the same pool entry rather than silently registering a second one.
registeredAs string
// ingress is the client-facing endpoint path this adapter CLAIMS to parse,
// e.g. "/v1/messages". It is the INBOUND dual of serialize_* (which is
// outbound): when an adapter declares it, the kernel YIELDS parsing of that
// endpoint's request body to the adapter's deserialize_request hook, instead
// of its own built-in parser. The kernel keeps a built-in parser as the
// fallback (R5), so an unconfigured deployment still serves /v1/messages —
// but once an adapter claims the endpoint, protocol iteration lives in the
// adapter, not the kernel. Empty means the adapter claims no ingress.
ingress string
// hidden marks an adapter that should NOT appear in the user-facing adapter
// list (GET /api/adapters → UI 的适配器页与源编辑器的适配器下拉)。隐藏的
// 适配器仍然正常加载、仍然可以被源引用——它只是不从选择列表里冒出来。
// 用途:厂商探针类/内部实验类适配器(如 opencodezen 的免费层门禁形态),
// 直接暴露会让运维误当成通用 OpenAI 适配器去选。声明方式:adapter.hidden = true。
hidden bool
}
// adapterPool owns the worker states for one adapter. Idle workers are held in
// a slice guarded by a cond; workers are created lazily on demand and reclaimed
// by the VM janitor once demand drops.
//
// Sizing is elastic rather than fixed (plan 阶段 2). Three inputs drive it:
//
// - maxW — the adapter's maximum concurrency (Σ max_concurrent of every source
// using it). It is the hard ceiling AND sets the growth step, so a hot,
// high-concurrency adapter ramps in batches while a low-concurrency one
// creeps up one state at a time.
// - created / len(idle) — how many states already exist, i.e. how much room is
// left to grow and how much is available to reclaim.
// - inUse — how many states are checked out right now, i.e. the live
// connection count for this adapter. It sets the shrink step: the fewer
// connections, the more aggressively idle states are released.
type adapterPool struct {
name string
script string
static staticInfo
mu sync.Mutex
cond *sync.Cond
idle []*worker
created int // live states: len(idle) + inUse (+ states being booted)
inUse int // states checked out right now == live connections
waiting int // goroutines blocked waiting for a free state
maxW int // hard ceiling = Σ max_concurrent of the sources using this adapter
used bool
closed bool
lastGrow time.Time
idleRounds int // consecutive janitor rounds that saw reclaimable slack
peakInUse int // high-water mark of inUse, for observability
// revision counts how many times this adapter's SOURCE was replaced (or the
// adapter was removed and re-added). Callers that cache a decision derived
// from the script's declared surface read it via AdapterRevision, so a
// runtime rewrite (a plugin installing a new version of its own adapter)
// cannot leave them acting on a stale answer.
revision uint64
// pluginMode makes boot() store the returned table under pluginGlobal
// instead of adapterGlobal. Everything else (elastic sizing, reclaim) is
// identical, which is why plugins reuse this pool rather than getting a
// second implementation.
pluginMode bool
}
const (
// growDivisor turns the adapter's max concurrency into the growth step CAP:
// step <= clamp(ceil(maxW / growDivisor), 1, growStepCap). The actual batch
// is additionally bounded by how many callers are queued (see growPlanLocked),
// so a wide adapter may ramp in big jumps but never warms states nobody is
// waiting for.
growDivisor = 8
growStepCap = 8
// growCooldown keeps a burst of misses from batching repeatedly while the
// previous batch is still booting.
growCooldown = time.Second
// maxPluginStates caps how many concurrent VM states ONE plugin may occupy.
// A plugin is third-party code on the request path, so its ceiling is much
// lower than an adapter's (which is sized from the sources'
// max_concurrent): a plugin hook is a short synchronous call, so a handful
// of states is already far more parallelism than any real hook needs, and a
// runaway plugin cannot balloon memory the way a per-source adapter pool can.
maxPluginStates = 4
// residentWorkers is how many states an adapter keeps warm once it has
// served at least one request. Booting is milliseconds, but keeping one warm
// removes that from the critical path of the next request. Adapters that
// were never used keep nothing.
residentWorkers = 1
// idleHeadroom is the spare idle state kept above the live connection count,
// so the next concurrent request does not have to boot.
idleHeadroom = 1
// shrinkInterval is how often the VM janitor reclaims idle states.
shrinkInterval = 30 * time.Second
// shrinkGraceRounds is how many consecutive janitor rounds must see slack
// before anything is released. It guards against tearing a pool down between
// two bursts — NOT against ordinary traffic: a pool serving one request at a
// time still has reclaimable slack, and requiring "no traffic at all" would
// pin a burst's leftover states forever on any busy gateway. The counter is
// therefore reset by GROWTH (real contention), not by a mere checkout.
shrinkGraceRounds = 2
)
// PoolStats is a snapshot of one adapter pool's elastic sizing, surfaced by the
// status API so the algorithm is observable instead of a black box.
type PoolStats struct {
Name string `json:"name"`
Created int `json:"created"`
Idle int `json:"idle"`
InUse int `json:"in_use"`
Waiting int `json:"waiting"`
Max int `json:"max"`
Resident int `json:"resident"`
GrowStep int `json:"grow_step"`
ShrinkStep int `json:"shrink_step"`
PeakInUse int `json:"peak_in_use"`
}
func newAdapterPool(name, script string, static staticInfo) *adapterPool {
p := &adapterPool{name: name, script: script, static: static, maxW: 1, revision: 1}
p.cond = sync.NewCond(&p.mu)
return p
}
// setMax updates the pool's hard ceiling (Σ max_concurrent of its sources).
// Lowering it does not kill live states: the janitor reclaims the excess.
func (p *adapterPool) setMax(n int) {
if n < 1 {
n = 1
}
p.mu.Lock()
p.maxW = n
p.cond.Broadcast()
p.mu.Unlock()
}
func ceilDiv(a, b int) int {
if b <= 0 {
return a
}
return (a + b - 1) / b
}
// growStepLocked is how many states to warm on a miss, derived from the
// adapter's maximum concurrency. Caller holds p.mu.
func (p *adapterPool) growStepLocked() int {
step := ceilDiv(p.maxW, growDivisor)
if step < 1 {
return 1
}
if step > growStepCap {
return growStepCap
}
return step
}
// residentLocked is the floor the janitor will not shrink below. Caller holds p.mu.
func (p *adapterPool) residentLocked() int {
if !p.used {
return 0
}
if residentWorkers > p.maxW {
return p.maxW
}
return residentWorkers
}
// shrinkStepLocked is how many idle states to release this round. The step is
// inversely proportional to the live connection count: with no connections the
// slack collapses in one round, while a busy adapter gives up one state at a
// time so the hot path keeps its warm states. Caller holds p.mu.
func (p *adapterPool) shrinkStepLocked() int {
excess := p.created - p.keepLocked()
if excess <= 0 {
return 0
}
step := ceilDiv(excess, 1+p.inUse)
if step > excess {
step = excess
}
if step > len(p.idle) {
step = len(p.idle)
}
if step < 0 {
return 0
}
return step
}
// keepLocked is the number of live states this pool should retain right now.
// Caller holds p.mu.
func (p *adapterPool) keepLocked() int {
keep := p.residentLocked()
if h := p.inUse + idleHeadroom; h > keep {
keep = h
}
if keep > p.maxW {
keep = p.maxW
}
return keep
}
func (p *adapterPool) stats() PoolStats {
p.mu.Lock()
defer p.mu.Unlock()
return PoolStats{
Name: p.name,
Created: p.created,
Idle: len(p.idle),
InUse: p.inUse,
Waiting: p.waiting,
Max: p.maxW,
Resident: p.residentLocked(),
GrowStep: p.growStepLocked(),
ShrinkStep: p.shrinkStepLocked(),
PeakInUse: p.peakInUse,
}
}
// boot creates a fresh LuaJIT state, registers the shared globals and executes
// the adapter script, storing the returned adapter table in the state.
func (p *adapterPool) boot() (*worker, error) {
L := golua.NewState()
L.OpenLibs()
setupGlobals(L)
if err := L.DoString(p.script); err != nil {
L.Close()
return nil, fmt.Errorf("compile adapter %s: %w", p.name, err)
}
if L.Type(-1) != golua.LUA_TTABLE {
L.Close()
return nil, fmt.Errorf("adapter %s must return a table", p.name)
}
// SetGlobal POPS the value off the stack, so it can only be called once per
// boot. Plugins therefore store under pluginGlobal only, and NOT under
// adapterGlobal: a second SetGlobal on the now-empty stack would assign nil.
if p.pluginMode {
L.SetGlobal(pluginGlobal)
} else {
L.SetGlobal(adapterGlobal)
}
L.SetTop(0)
return &worker{L: L}, nil
}
// growPlanLocked reserves capacity for a batch prewarm and returns how many
// EXTRA states to boot in the background (the caller already reserved one for
// itself).
//
// Two independent limits apply, and BOTH matter:
//
// - the adapter's max concurrency caps the step (growStepLocked), so a
// high-concurrency adapter is allowed to ramp in bigger jumps than a
// narrow one;
// - the number of goroutines actually BLOCKED waiting for a state bounds it
// to real demand.
//
// Sizing on the ceiling alone over-provisions badly: an adapter with
// max_concurrent=76 has a step of 8, so two concurrent requests would warm 8
// states, and the janitor would throw 7 of them away a minute later — boot,
// discard, repeat, with RSS oscillating for no benefit. Prewarming only for
// goroutines that are genuinely queued keeps the ramp cheap without the churn.
//
// Caller holds p.mu.
func (p *adapterPool) growPlanLocked(contended bool) int {
if !contended || p.waiting <= 0 {
return 0
}
if time.Since(p.lastGrow) < growCooldown {
return 0
}
extra := p.growStepLocked() - 1
if extra > p.waiting {
extra = p.waiting // never warm more than the queue needs
}
if extra <= 0 {
return 0
}
if room := p.maxW - p.created; extra > room {
extra = room
}
if extra <= 0 {
return 0
}
// Reserve immediately so concurrent acquires respect the ceiling while the
// batch is still booting.
p.created += extra
p.lastGrow = time.Now()
return extra
}
// prewarm boots n reserved states and publishes them as idle. A boot failure
// releases its reservation; it is not fatal, since the synchronous path
// surfaces adapter errors already.
func (p *adapterPool) prewarm(n int) {
for i := 0; i < n; i++ {
p.mu.Lock()
closed := p.closed
p.mu.Unlock()
if closed {
p.releaseReservation(n - i)
return
}
w, err := p.boot()
if err != nil {
p.releaseReservation(n - i)
return
}
p.mu.Lock()
if p.closed {
p.created--
p.mu.Unlock()
w.close()
p.releaseReservation(n - i - 1)
return
}
p.idle = append(p.idle, w)
p.cond.Signal()
p.mu.Unlock()
}
}
func (p *adapterPool) releaseReservation(n int) {
if n <= 0 {
return
}
p.mu.Lock()
p.created -= n
if p.created < 0 {
p.created = 0
}
p.cond.Broadcast()
p.mu.Unlock()
}
// acquireAcrossSwap checks out a worker, re-reading the registry whenever the
// pool it was handed has already been retired.
//
// Retry (rather than one-shot) because the caller can lose the race repeatedly
// when several goroutines rotate the same adapter at once: each retry can land
// on a pool that the NEXT rotation is already closing. The bound is a small
// number of attempts, not unbounded — a permanently broken adapter must still
// surface an error instead of spinning.
func acquireAcrossSwap(v *VM, name string, p *adapterPool) (*worker, error) {
// A rotation publishes the new pool BEFORE retiring the old one (see
// LoadAdapterSource), so a lookup here can only ever find the newest one.
// The wait is what makes this correct under sustained rotation: several
// goroutines rewriting the same adapter keep retiring the pool this one
// holds, and a caller that gave up at that moment would fail a request for
// no reason other than landing in a microsecond-wide window.
//
// The bound is wall-clock, not attempts, so the behaviour does not change
// with contention. It is generous enough to ride out a burst of rotations
// and short enough that a genuinely gone adapter still surfaces promptly.
deadline := time.Now().Add(2 * time.Second)
for {
w, err := p.acquire()
if err == nil {
return w, nil
}
if !errors.Is(err, ErrPoolClosed) {
return nil, err
}
fresh := v.pool(name)
if fresh == nil {
return nil, fmt.Errorf("adapter %s not loaded", name)
}
p = fresh
if time.Now().After(deadline) {
return nil, fmt.Errorf("adapter %s: pool kept being replaced for 2s: %w", name, err)
}
// Yield briefly: the new pool needs a moment to publish, and spinning
// here would burn the CPU that the rotation itself needs.
time.Sleep(time.Millisecond)
}
}
// ErrPoolClosed reports that the pool a caller was about to use from was
// retired between lookup and acquire — which happens on every adapter hot-swap
// (a plugin rotating a token rewrites the adapter, and LoadAdapterSource
// replaces its pool).
//
// It is a sentinel so callers can RETRY against the newly published pool
// instead of surfacing an error to the client. Without it, a token rotation
// under load produces a burst of "adapter X pool closed" failures — the
// rotation is the operator's intent, and the window is microseconds wide, so
// failing the request that happened to land there is strictly worse than
// re-reading the registry.
var ErrPoolClosed = errors.New("adapter pool closed")
func (p *adapterPool) acquire() (*worker, error) {
p.mu.Lock()
for {
if p.closed {
p.mu.Unlock()
return nil, fmt.Errorf("adapter %s: %w", p.name, ErrPoolClosed)
}
if n := len(p.idle); n > 0 {
w := p.idle[n-1]
p.idle = p.idle[:n-1]
p.checkoutLocked()
p.mu.Unlock()
return w, nil
}
if p.created < p.maxW {
// Every existing state busy on a miss = real concurrency, not a
// sequential caller reusing one warm state.
contended := p.created > 0 && p.inUse >= p.created
p.created++
// Growing is the signal that capacity is genuinely short, so any
// pending shrink decision is stale.
p.idleRounds = 0
extra := p.growPlanLocked(contended)
p.mu.Unlock()
if extra > 0 {
go p.prewarm(extra)
}
w, err := p.boot()
if err != nil {
p.mu.Lock()
p.created--
p.cond.Signal()
p.mu.Unlock()
return nil, err
}
p.mu.Lock()
p.checkoutLocked()
p.mu.Unlock()
return w, nil
}
p.waiting++
p.cond.Wait()
p.waiting--
}
}
// checkoutLocked accounts one state as handed out. Caller holds p.mu.
func (p *adapterPool) checkoutLocked() {
p.inUse++
p.used = true
if p.inUse > p.peakInUse {
p.peakInUse = p.inUse
}
}
func (p *adapterPool) release(w *worker) {
w.L.SetTop(0)
p.mu.Lock()
if p.inUse > 0 {
p.inUse--
}
if p.closed {
p.created--
p.mu.Unlock()
w.close()
return
}
// A lowered ceiling is honoured on the spot rather than waiting for the
// janitor: returning a state the pool may no longer keep closes it.
if p.created > p.maxW {
p.created--
p.cond.Signal()
p.mu.Unlock()
w.close()
return
}
p.idle = append(p.idle, w)
p.cond.Signal()
p.mu.Unlock()
}
// reclaim releases idle states down to the current keep target, using a step
// scaled by the live connection count. It returns how many states were closed.
func (p *adapterPool) reclaim() int {
p.mu.Lock()
if p.closed {
p.mu.Unlock()
return 0
}
step := p.shrinkStepLocked()
if step <= 0 {
p.idleRounds = 0
p.mu.Unlock()
return 0
}
p.idleRounds++
if p.idleRounds < shrinkGraceRounds {
p.mu.Unlock()
return 0
}
victims := append([]*worker(nil), p.idle[len(p.idle)-step:]...)
p.idle = p.idle[:len(p.idle)-step]
p.created -= step
p.idleRounds = 0
p.mu.Unlock()
for _, w := range victims {
w.close()
}
return step
}
func (p *adapterPool) shutdown() {
p.mu.Lock()
p.closed = true
idle := p.idle
p.idle = nil
p.created -= len(idle)
p.cond.Broadcast()
p.mu.Unlock()
for _, w := range idle {
w.close()
}
}
// VM aggregates per-adapter worker pools. All exported methods are safe for
// concurrent use.
type VM struct {
mu sync.RWMutex
dir string
pools map[string]*adapterPool
// hiddenRefs 记录"被源引用的隐藏适配器"。hidden=true 的适配器默认不进
// ListAdapters,但被源引用时必须保持可见(源编辑器要能回显当前值)。
// 由内核在配置装载/源变更时调用 SetHiddenRefs 维护;nil = 未标记。
hiddenRefs map[string]bool
// janitor drives elastic shrinking. One goroutine serves every pool (not
// one per adapter), so an idle gateway costs a single sleeping goroutine.
janitorOnce sync.Once
janitorStop chan struct{}
janitorDone chan struct{}
}
// SetHiddenRefs 供内核同步"哪些隐藏适配器正被源引用"。传 nil/空集 = 没有引用。
func (v *VM) SetHiddenRefs(names []string) {
v.mu.Lock()
defer v.mu.Unlock()
if len(names) == 0 {
v.hiddenRefs = nil
return
}
m := make(map[string]bool, len(names))
for _, n := range names {
m[n] = true
}
v.hiddenRefs = m
}
func NewVM(dir string) *VM {
return &VM{
dir: dir,
pools: map[string]*adapterPool{},
janitorStop: make(chan struct{}),
janitorDone: make(chan struct{}),
}
}
// startJanitor launches the single shrink loop. Idempotent.
func (v *VM) startJanitor() {
v.janitorOnce.Do(func() {
go v.janitorLoop(shrinkInterval)
})
}
func (v *VM) janitorLoop(every time.Duration) {
defer close(v.janitorDone)
t := time.NewTicker(every)
defer t.Stop()
for {
select {
case <-v.janitorStop:
return
case <-t.C:
v.reclaimIdle()
}
}
}
// reclaimIdle runs one shrink round over every pool. Exported behaviour is
// tested through ReclaimIdleNow.
func (v *VM) reclaimIdle() int {
v.mu.RLock()
pools := make([]*adapterPool, 0, len(v.pools))
for _, p := range v.pools {
pools = append(pools, p)
}
v.mu.RUnlock()
n := 0
for _, p := range pools {
n += p.reclaim()
}
return n
}
// ReclaimIdleNow forces one shrink round and reports how many worker states
// were closed. Intended for tests and for an explicit "release idle memory"
// action; normal operation relies on the janitor.
func (v *VM) ReclaimIdleNow() int { return v.reclaimIdle() }
// PoolStats returns the elastic sizing snapshot of every loaded adapter,
// sorted by name.
func (v *VM) PoolStats() []PoolStats {
v.mu.RLock()
pools := make([]*adapterPool, 0, len(v.pools))
for _, p := range v.pools {
pools = append(pools, p)
}
v.mu.RUnlock()
out := make([]PoolStats, 0, len(pools))
for _, p := range pools {
out = append(out, p.stats())
}
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
return out
}
func (v *VM) Start() error {
v.startJanitor()
if v.dir == "" {
return nil
}
// First-run seeding: a brand-new adapter dir is created and populated with
// the bundled adapters. If the dir already exists it is treated as
// authoritative and never rewritten — deleting a file there is a real delete.
firstRun := false
if _, err := os.Stat(v.dir); os.IsNotExist(err) {
firstRun = true
if err := os.MkdirAll(v.dir, 0755); err != nil {
return fmt.Errorf("mkdir adapter dir: %w", err)
}
}
if firstRun {
if err := v.writeBundledAdapters(); err != nil {
return err
}
}
entries, err := os.ReadDir(v.dir)
if err != nil {
return err
}
for _, e := range entries {
if filepath.Ext(e.Name()) != ".lua" {
continue
}
if err := v.LoadAdapter(filepath.Join(v.dir, e.Name())); err != nil {
fmt.Printf("[lua] preload %s: %v\n", e.Name(), err)
}
}
return nil
}
// bundledPluginsDir is the directory inside the embedded FS holding the
// plugins shipped with the gateway. It is separate from adapters/ on purpose:
// the two are loaded by different machinery into different kinds of Lua state
// (a protocol transform vs. request-pipeline hooks), and keeping them apart
// makes it obvious that dropping a file in one does not affect the other.
const bundledPluginsDir = "plugins"
// ReadBundledPlugin returns the source of a plugin shipped with the gateway.
// It exists so a test (or an operator tool) can load a bundled plugin without
// depending on whether seeding has already run for this directory.
func ReadBundledPlugin(name string) (string, error) {
data, err := bundledAdapters.ReadFile(bundledPluginsDir + "/" + name + ".lua")
if err != nil {
return "", fmt.Errorf("bundled plugin %s: %w", name, err)
}
return string(data), nil
}
// writeBundledPlugins seeds the plugin directory with the shipped plugins.
//
// It runs only when the directory does not exist yet (same rule as adapters):
// once the directory exists it is authoritative, so deleting a shipped plugin is
// a real delete and editing one survives restarts.
func writeBundledPlugins(dir string) error {
if dir == "" {
return nil
}
if err := os.MkdirAll(dir, 0755); err != nil {
return fmt.Errorf("mkdir plugin dir: %w", err)
}
entries, err := bundledAdapters.ReadDir(bundledPluginsDir)
if err != nil {
return nil // nothing embedded; not an error
}
for _, e := range entries {
if e.IsDir() || filepath.Ext(e.Name()) != ".lua" {
continue
}
dst := filepath.Join(dir, e.Name())
// Per-file seeding: a plugin already present is left ALONE. That is what
// makes an operator's own copy (or a locally edited one) survive an
// upgrade, and it is also why a newly embedded plugin still arrives —
// absence of THAT file, not of the directory, is the trigger.
if _, err := os.Stat(dst); err == nil {
continue
} else if !os.IsNotExist(err) {
return err
}
data, err := bundledAdapters.ReadFile(bundledPluginsDir + "/" + e.Name())
if err != nil {
continue
}
if err := os.WriteFile(dst, data, 0644); err != nil {
return err
}
}
return nil
}
func (v *VM) Stop() {
select {
case <-v.janitorStop:
// already stopped
default:
close(v.janitorStop)
}
v.mu.Lock()
pools := v.pools
v.pools = map[string]*adapterPool{}
v.mu.Unlock()
for _, p := range pools {
p.shutdown()
}
}
// ConfigureConcurrency sets each adapter's worker CEILING to the sum of
// max_concurrent of every source using it. It is a ceiling, not a preallocation:
// states are booted on demand and reclaimed when demand drops, so a config with
// a large aggregate concurrency no longer implies a large resident pool.
// Values below one are clamped to 1.
func (v *VM) ConfigureConcurrency(adapterConcurrency map[string]int) {
v.mu.RLock()
defer v.mu.RUnlock()
for name, n := range adapterConcurrency {
if p, ok := v.pools[name]; ok {
p.setMax(n)
}
}
}
func (v *VM) ListAdapters() []APIAdapter {
v.mu.RLock()
defer v.mu.RUnlock()
// hidden=true 的适配器不进用户可见列表(仍正常加载、仍可被源引用)。
// 例外:**正在被某个源引用**的隐藏适配器必须仍出现在列表里——否则源编辑器
// 打开时找不到当前值,下拉会被重置成空,造成"保存后适配器悄悄丢了"。
// 引用关系由内核持有(core 在配置装载时标记),VM 只看这个位。
keepHidden := v.hiddenKeep()
list := make([]APIAdapter, 0, len(v.pools))
for name, p := range v.pools {
if p.static.hidden && !keepHidden[name] {
continue
}
list = append(list, APIAdapter{Name: name, Version: p.static.version})
}
sort.Slice(list, func(i, j int) bool { return list[i].Name < list[j].Name })
return list
}
// hiddenKeep returns the set of hidden adapter names that must STAY visible
// because at least one source references them. nil 表示内核未标记(测试环境/
// 未装载配置),此时隐藏适配器一律隐藏——与"新适配器默认隐藏"的预期一致。
func (v *VM) hiddenKeep() map[string]bool {
if v.hiddenRefs == nil {
return nil
}
return v.hiddenRefs
}
// AdapterForIngress returns the registered name of the adapter that CLAIMS the
// given client-facing endpoint path (its static.ingress), or "" when none does.
//
// This is the kernel-yields decision for an inbound endpoint (user requirement:
// "/v1/messages is the kernel's fallback endpoint; an adapter may declare it and
// the kernel actively yields"). The kernel keeps its own built-in parser as the
// fallback (R5) — AdapterForIngress returning "" means "no adapter claimed it,
// use the built-in", so an unconfigured deployment still serves the endpoint.
//
// Ties (two adapters claiming the same path) resolve to the lexically-first
// name so the choice is deterministic rather than map-iteration order. A real
// deployment should not have two adapters claiming one ingress; when it does,
// the plugin conflict-arbitration surface is the place to make it explicit, but
// a parse endpoint must still pick SOMETHING rather than vary per request.
func (v *VM) AdapterForIngress(path string) string {
if path == "" {
return ""
}
v.mu.RLock()
defer v.mu.RUnlock()
chosen := ""
for name, p := range v.pools {
p.mu.Lock()
claims := p.static.ingress == path
p.mu.Unlock()
if claims && (chosen == "" || name < chosen) {
chosen = name
}
}
return chosen
}
// LoadAdapter compiles and registers an adapter script from a file.
//
// The path is validated for SAFETY before reading (R6): it must sit inside the
// VM's adapter dir and the derived name must be a legal adapter name (no path
// traversal, absolute paths, or whitespace). It does NOT reject built-in names
// here: an operator placing openai.lua in adapter_dir to OVERRIDE the bundled
// openai is the documented hot-override mechanism. Only PLUGIN installs are
// barred from shadowing a built-in (see InstallPluginAdapter), because a
// third-party plugin clobbering the kernel's shipped adapter is tampering,
// whereas the operator editing their own deployment is not.
func (v *VM) LoadAdapter(path string) error {
name, err := ValidateAdapterPath(path, v.dir, nil)
if err != nil {
return err
}
data, err := os.ReadFile(path)
if err != nil {
return fmt.Errorf("read adapter: %w", err)
}
return v.LoadAdapterSource(name, string(data))
}
// LoadAdapterSource registers an adapter from source code. Replacing an
// existing adapter shuts the old pool down and re-boots lazily, carrying the
// configured ceiling over so a hot-reload does not reset elastic sizing.
//
// The name is validated for path SAFETY only (no traversal/absolute/space).
// Built-in names are allowed: an operator override is legitimate. A plugin
// that must NOT shadow a built-in uses InstallPluginAdapter, which adds the
// reserved-name gate.
func (v *VM) LoadAdapterSource(name, code string) error {
if err := ValidateAdapterName(name, nil); err != nil {
return err
}
// 共享前导注入点 —— 只此一处。适配器源码有三个编译点
//(worker boot / HasFn 探针 / inspectScript),它们全部源自这里存进
// pool 的 script 文本,所以拼在这里就同时覆盖三条路径,且不改变 Lua
// 作用域语义(运行时 SetGlobal 会在热重载/池重建时丢失)。
// 见 adapter_prelude.go 的开篇:12 个适配器里有 12680 字节的逐字重复。
code = withAdapterPrelude(code)
static, err := inspectScript(name, code)
if err != nil {
return err
}
maxW := 1
v.mu.RLock()
if p, ok := v.pools[static.name]; ok {
p.mu.Lock()
maxW = p.maxW
p.mu.Unlock()
}
v.mu.RUnlock()
v.mu.Lock()
// A reload must also retire an adapter that was loaded under a DIFFERENT
// name than the one this script declares (a plugin can register under a key
// it controls while shipping an adapter that declares the name sources
// reference). Without this, updating that adapter would install a second
// pool and leave the old one serving traffic.
//
// The key can live under EITHER name (this is the "reload replaces the same
// pool entry rather than silently registering a second one" rule):
//
// - pools[static.name] — the pool filed under the DECLARED name (the
// common case, and the only case when a plugin installs under a key that
// matches its script's own name), or
// The pool registry is keyed by the DECLARED name (static.name), always:
// first install files pools[static.name], and a reload replaces that same
// slot. registeredAs only RECORDS the registration key for reverse lookups
// (EffectiveAdapterName scans for it); it is not a second key space.
//
// An earlier version redirected the key to prev.static.registeredAs here,
// assuming pools were filed under the registration key — but first installs
// file under the declared name, so that redirect created a SECOND pool and
// left the old one serving whenever key≠declared (a real bug: a v2 install
// via arbitration left v1 serving, and UpdateAdapter had the same hole).
key := static.name
old := v.pools[key]
// The new pool MUST be filed under the SAME slot as the old one (key), and
// named after that slot, so a later lookup by either the DECLARED name or
// the REGISTRATION KEY finds the one live pool. Naming it after static.name
// would move the pool to the declared-name slot when the incumbent lived
// under the registration key, leaving the registration-key slot pointing at
// the stale (old) pool that shutdown() then tears down — callers reading by
// the registration key see "adapter missing", callers by declared name see
// a pool that isn't the one serving. key is the stable identity; the
// declared name is reachable through static.name.
p := newAdapterPool(key, code, *static)
p.setMax(maxW)
// Carry the revision forward so the new source keeps counting ABOVE the old
// one. A cache keyed on (vm, revision, name) must see a value different
// from the previous generation's; restarting at 1 could collide with a
// value an earlier generation already published, which would resurrect a
// stale cache entry — exactly the bug this revision exists to prevent.
//
// The old pool is read while v.mu is HELD (a nested RLock here would
// self-deadlock: sync.RWMutex is not reentrant) and while its own mu is
// free, because shutdown() — which takes p.mu — happens after v.mu is
// released below. No other path can mutate p.revision while we hold v.mu,
// because every writer reaches the pool through v.mu first.
if old != nil {
old.mu.Lock()
p.revision = old.revision + 1
old.mu.Unlock()
}
v.pools[key] = p
v.mu.Unlock()
// shutdown() happens AFTER v.mu is released, and the new pool is already
// published, so a request arriving in between sees the new code. Shutting
// down first would open a window where the adapter is neither old nor new:
// concurrent rotations then fail with "adapter X pool closed" — the exact
// failure a token-rotation plugin would hit under load.
if old != nil {
old.shutdown()
}
v.startJanitor()
return nil
}
// RemoveAdapter evicts an adapter and shuts its workers down.
func (v *VM) RemoveAdapter(name string) {
v.mu.Lock()
p := v.pools[name]
delete(v.pools, name)
v.mu.Unlock()
if p != nil {
p.shutdown()
}
}
// poolGlobal returns the Lua global holding a pool's returned table: plugins
// store theirs under one global, adapters under another. Transform and friends
// must read the SAME one boot() wrote, or a plugin-mode pool answers
// "has no table" even though the state is perfectly healthy.
func poolGlobal(p *adapterPool) string {
if p != nil && p.pluginMode {
return pluginGlobal
}
return adapterGlobal
}
func (v *VM) pool(name string) *adapterPool {
v.mu.RLock()
defer v.mu.RUnlock()
return v.pools[name]
}
// bundledNames returns the set of adapter names that ship inside the kernel.
// It is derived from the embedded FS (same source writeBundledAdapters uses),
// so a new built-in adapter is automatically reserved and an operator cannot
// clobber it by importing a same-named external adapter.
// BundledAdapterNames is the exported counterpart of bundledNames for callers
// outside the lua package (core.UploadAdapter validates against it). It returns
// the set of adapter names shipped inside the kernel so an external import that
// shadows one is rejected up front (R6).
func (v *VM) BundledAdapterNames() []string { return v.bundledNames() }
func (v *VM) bundledNames() []string {
entries, err := bundledAdapters.ReadDir("adapters")
if err != nil {
return nil
}
out := make([]string, 0, len(entries))
for _, e := range entries {
if e.IsDir() {
continue
}
out = append(out, strings.TrimSuffix(e.Name(), ".lua"))
}
return out
}
func (v *VM) writeBundledAdapters() error {
// Derive the set from the embedded FS instead of a hardcoded name list so
// adding an adapter never requires maintaining a second list.
entries, err := bundledAdapters.ReadDir("adapters")
if err != nil {
return err
}
for _, e := range entries {
if e.IsDir() || !strings.HasSuffix(e.Name(), ".lua") {
continue
}
dst := filepath.Join(v.dir, e.Name())
if _, err := os.Stat(dst); err == nil {
continue
}
data, err := bundledAdapters.ReadFile("adapters/" + e.Name())
if err != nil {
continue
}
if err := os.WriteFile(dst, data, 0644); err != nil {
return err
}
}
return nil
}
// Transform runs adapter.Fn(raw) and returns the resulting string.
//
// One retry on ErrPoolClosed: the pool was swapped by a concurrent rewrite
// (plugin token rotation) between this lookup and the checkout. Re-reading the
// registry gets the new pool, which is exactly what the caller wants.
func (v *VM) Transform(name, fn, raw string) (string, error) {
p := v.pool(name)
if p == nil {
return "", fmt.Errorf("adapter %s not loaded", name)
}
w, err := acquireAcrossSwap(v, name, p)
if err != nil {
return "", err
}
defer p.release(w)
L := w.L
L.SetTop(0)
defer L.SetTop(0)
L.GetGlobal(poolGlobal(p))
if L.IsNil(-1) {
return "", fmt.Errorf("adapter %s has no table", name)
}
L.GetField(-1, fn)
if L.IsNil(-1) || !L.IsFunction(-1) {
return "", fmt.Errorf("adapter %s missing %s", name, fn)
}
L.PushString(raw)
if err := L.Call(1, 1); err != nil {
return "", fmt.Errorf("%s: %w", fn, err)
}
return L.ToString(-1), nil
}
// BuildHeaders executes adapter.build_headers(meta). If the adapter defines no
// build_headers hook, it falls back to the static adapter.headers table.
func (v *VM) BuildHeaders(name string, meta map[string]interface{}) (map[string]string, error) {
p := v.pool(name)
if p == nil {
return nil, fmt.Errorf("adapter %s not loaded", name)
}
w, err := p.acquire()
if err != nil {
return nil, err
}
defer p.release(w)
L := w.L
L.SetTop(0)
defer L.SetTop(0)
L.GetGlobal(adapterGlobal)
if L.IsNil(-1) {
return nil, fmt.Errorf("adapter %s has no table", name)
}
L.GetField(-1, "build_headers")
if !L.IsFunction(-1) {
// no hook: fall back to the static headers captured at load time
p.mu.Lock()
hdr := make(map[string]string, len(p.static.headers))
for k, h := range p.static.headers {
hdr[k] = h
}
p.mu.Unlock()
return hdr, nil
}
L.SetTop(0)
L.GetGlobal(adapterGlobal)
L.GetField(-1, "build_headers")
pushGoValue(L, meta)
if err := L.Call(1, 1); err != nil {
return nil, fmt.Errorf("build_headers: %w", err)
}
if L.Type(-1) != golua.LUA_TTABLE {
return nil, fmt.Errorf("build_headers returned non-table")
}
headers := map[string]string{}
luaTableToMap(L, -1, headers)
return headers, nil
}
// TransformError executes the optional adapter.transform_error(status, body)
// hook: per-source condensing of an upstream error response into a short
// client-facing reason. ok=false means the adapter defines no hook (or it
// failed) — the caller falls back to the generic Go-side condenser.
func (v *VM) TransformError(name string, status int, body string) (reason string, ok bool, err error) {
p := v.pool(name)
if p == nil {
return "", false, fmt.Errorf("adapter %s not loaded", name)
}
w, err := p.acquire()
if err != nil {
return "", false, err
}
defer p.release(w)
L := w.L
L.SetTop(0)
defer L.SetTop(0)
L.GetGlobal(adapterGlobal)
if L.IsNil(-1) {
return "", false, nil
}
L.GetField(-1, "transform_error")
if !L.IsFunction(-1) {
return "", false, nil
}
L.SetTop(0)
L.GetGlobal(adapterGlobal)
L.GetField(-1, "transform_error")
L.PushInteger(int64(status))
L.PushString(body)
if callErr := L.Call(2, 1); callErr != nil {
return "", false, fmt.Errorf("transform_error: %w", callErr)
}
if L.Type(-1) != golua.LUA_TSTRING {
return "", false, nil
}
return L.ToString(-1), true, nil
}
func (v *VM) Endpoint(name string) string {
p := v.pool(name)
if p == nil {
return ""
}
p.mu.Lock()
defer p.mu.Unlock()
return p.static.endpoint
}
// StreamEndpoint returns the adapter's streaming endpoint, or "" when it
// declares none (in which case the caller keeps using Endpoint).
//
// This is the whole mechanism that keeps vendor streaming verbs out of the
// kernel: an adapter whose stream path differs declares endpoint_stream, and
// the dispatcher only ever compares two opaque strings.
func (v *VM) StreamEndpoint(name string) string {
p := v.pool(name)
if p == nil {
return ""
}
p.mu.Lock()
defer p.mu.Unlock()
return p.static.endpointStream
}
// HasFn reports whether the named adapter declares a callable field fn,
// WITHOUT acquiring a worker and WITHOUT running the script.
//
// It exists for capability probing that runs on the request path: the
// out-bound serializer asks "does this adapter declare serialize_chunk?" once
// per adapter, not once per chunk. Probing by calling Transform would be
// wrong twice over — it costs a pool checkout, and a MISSING hook comes back
// as an error string rather than a boolean.
//
// The probe runs on a throwaway state built from the SAME script text, so the
// answer reflects what the script declares. A script whose table is built
// conditionally on its environment could differ per state; that is already
// true of every other staticInfo field (endpoint, headers), so the pool does
// not get a special exemption here.
func (v *VM) HasFn(name, fn string) bool {
p := v.pool(name)
if p == nil {
return false
}
p.mu.Lock()
script := p.script
p.mu.Unlock()
if script == "" {
return false
}
L := golua.NewState()
defer L.Close()
L.OpenLibs()
setupGlobals(L)
if err := L.DoString(script); err != nil {
return false
}
if L.Type(-1) != golua.LUA_TTABLE {
return false
}
L.GetField(-1, fn)
defer L.Pop(1)
return !L.IsNil(-1) && L.IsFunction(-1)
}
// ---- static inspection (compile-once at load time) ----
func inspectScript(name, code string) (*staticInfo, error) {
L := golua.NewState()
L.OpenLibs()
setupGlobals(L)
if err := L.DoString(code); err != nil {
L.Close()
return nil, fmt.Errorf("compile adapter: %w", err)
}
if L.Type(-1) != golua.LUA_TTABLE {
L.Close()
return nil, fmt.Errorf("adapter script must return a table")
}
// The registration key is the default identity; a script MAY declare its own
// name, in which case the declared one wins. That case matters for plugins:
// a plugin registers under a key it controls, but the adapter it installs
// may declare the name sources actually reference — so the bridge reports
// the effective name back instead of guessing.
info := &staticInfo{name: name, headers: map[string]string{}}
if n := readString(L, -1, "name"); n != "" {
info.name = n
}
info.registeredAs = name
info.version = readString(L, -1, "version")
info.endpoint = readString(L, -1, "endpoint")
info.endpointStream = readString(L, -1, "endpoint_stream")
info.ingress = readString(L, -1, "ingress")
// hidden:适配器可声明 adapter.hidden = true 退出用户可见列表(仍可被源引用)。
L.GetField(-1, "hidden")
info.hidden = L.ToBoolean(-1)
L.Pop(1)
if getFieldBatch(L, "headers") {
luaTableToMap(L, -1, info.headers)
L.Pop(1)
} else {
L.Pop(1)
}
L.Close()
return info, nil
}
// getFieldBatch fetches field of the table at the top of the stack and returns
// true when the fetched value (now at the top) is a table.
func getFieldBatch(L *golua.State, field string) bool {
L.GetField(-1, field)
return L.Type(-1) == golua.LUA_TTABLE
}
func readString(L *golua.State, tableIdx int, field string) string {
L.GetField(tableIdx, field)
defer L.Pop(1)
if L.Type(-1) == golua.LUA_TNIL || L.Type(-1) == golua.LUA_TNONE {
return ""
}
return normalizeString(L, -1)
}
func normalizeString(L *golua.State, idx int) string {
if L.Type(idx) == golua.LUA_TNUMBER {
n := L.ToNumber(idx)
return strconv.FormatFloat(n, 'f', -1, 64)
}
return L.ToString(idx)
}
// ---- shared globals installed into every worker state ----
// restorePcall puts the real pcall/xpcall back in place. golua's OpenLibs
// replaces them with unsafe_ variants (it captures errors via its own panic
// handler); the built-in adapters rely on pcall for soft error handling.
func restorePcall(L *golua.State) {
for _, name := range []string{"pcall", "xpcall"} {
L.GetGlobal("unsafe_" + name)
if !L.IsNil(-1) {
L.SetGlobal(name)
} else {
L.Pop(1)
}
}
}
// hookGuardName is the Lua global that wraps every plugin hook call.
const hookGuardName = "__llmsproxy_call_hook"
// hookGuardSrc defines the wrapper. It exists because of a hard constraint in
// the binding: golua's callEx calls L.StackTrace() on ANY pcall error, and
// StackTrace() calls lua_getinfo, which SIGSEGVs in this LuaJIT build once the
// stack is deep enough (a request_end payload with the AUTO chain trace does
// it). Go cannot recover from a C-level signal, so a single Lua mistake inside
// one plugin killed the whole gateway — the production outage this was written
// for. Repeated crashes proved the boundary is not theoretical.
//
// pcall INSIDE Lua catches the error before golua ever sees a non-zero
// pcall status, so the C stack-trace path is never entered. The failure comes
// back as an ordinary (nil, message) pair, which the Go side records in
// hook_errors and moves on from — the documented contract that "a broken plugin
// must not affect request forwarding" finally holds for script errors too, not
// just for Go panics.
//
// returns: (result, errorMessage) — both nil/"" on success.
const hookGuardSrc = `
function ` + hookGuardName + `(fn, payload)
local ok, res = pcall(fn, payload)
if not ok then
return nil, tostring(res)
end
return res, nil
end`
func registerHookGuard(L *golua.State) {
if err := L.DoString(hookGuardSrc); err != nil {
// Nothing useful to do here beyond leaving the global absent: invoke()
// checks for it and falls back to a direct call, which still works for
// correct plugins (only their ERRORS stop being survivable).
L.Pop(1)
}
}
func setupGlobals(L *golua.State) {
restorePcall(L)
buildJSONTable(L)
registerHookGuard(L)
registerFn(L, "hmac_sha256_hex", func(L *golua.State) int {
key := L.ToString(1)
data := L.ToString(2)
m := hmac.New(sha256.New, []byte(key))
m.Write([]byte(data))
L.PushString(hex.EncodeToString(m.Sum(nil)))
return 1
})
registerFn(L, "sha256_hex", func(L *golua.State) int {
h := sha256.Sum256([]byte(L.ToString(1)))
L.PushString(hex.EncodeToString(h[:]))
return 1
})
registerFn(L, "base64_encode", func(L *golua.State) int {
L.PushString(base64.StdEncoding.EncodeToString([]byte(L.ToString(1))))
return 1
})
registerFn(L, "tohex", func(L *golua.State) int {
L.PushString(hex.EncodeToString([]byte(L.ToString(1))))
return 1
})
registerFn(L, "log", func(L *golua.State) int {
fmt.Printf("[adapter/%s] %s\n", L.ToString(1), L.ToString(2))
return 0
})
}
func buildJSONTable(L *golua.State) {
L.NewTable()
for _, name := range []string{"encode", "decode"} {
var fn golua.LuaGoFunction
switch name {
case "encode":
fn = func(L *golua.State) int {
b, err := jsonEncode(luaValueToGo(L, 1))
if err != nil {
L.PushString("null")
return 1
}
L.PushString(string(b))
return 1
}
case "decode":
fn = func(L *golua.State) int {
v, err := jsonDecode(L.ToString(1))
if err != nil {
L.PushNil()
return 1
}
pushGoValue(L, v)
return 1
}
}
L.PushGoClosure(fn)
L.SetField(-2, name)
}
L.SetGlobal("json")
}
func registerFn(L *golua.State, name string, fn golua.LuaGoFunction) {
L.PushGoClosure(fn)
L.SetGlobal(name)
}
// ---- value conversions ----
func luaValueToGo(L *golua.State, idx int) interface{} {
switch L.Type(idx) {
case golua.LUA_TSTRING:
return L.ToString(idx)
case golua.LUA_TNUMBER:
return L.ToNumber(idx)
case golua.LUA_TBOOLEAN:
return L.ToBoolean(idx)
case golua.LUA_TTABLE:
return luaTableToGoValue(L, idx)
default:
return nil
}
}
// luaTableToGo value converts a Lua table to either []interface{} (array 1..n)
// or map[string]interface{}.
func luaTableToGoValue(L *golua.State, idx int) interface{} {
abs := idx
if idx < 0 {
abs = L.GetTop() + idx + 1
}
m := map[string]interface{}{}
allInt, n, maxK := true, 0, 0
L.PushNil()
for L.Next(abs) != 0 {
kt := L.Type(-2)
if kt == golua.LUA_TNUMBER {
k := int(L.ToNumber(-2))
if k > maxK {
maxK = k
}
} else {
allInt = false
}
m[keyToString(L, -2)] = luaValueToGo(L, -1)
L.Pop(1)
n++
}
if allInt && maxK > 0 && n == maxK {
arr := make([]interface{}, maxK)
for i := 1; i <= maxK; i++ {
arr[i-1] = m[strconv.Itoa(i)]
}
return arr
}
return m
}
func luaTableToMap(L *golua.State, idx int, dst map[string]string) {
abs := idx
if idx < 0 {
abs = L.GetTop() + idx + 1
}
L.PushNil()
for L.Next(abs) != 0 {
dst[keyToString(L, -2)] = normalizeString(L, -1)
L.Pop(1)
}
}
// keyToString reads the stack value at idx as a string without modifying the
// original (golua's ToString mutates numeric stack values in place, which
// would break the next lua_next call), then pops the temporary copy.
func keyToString(L *golua.State, idx int) string {
L.PushValue(idx)
defer L.Pop(1)
return normalizeString(L, -1)
}
func pushGoValue(L *golua.State, v interface{}) {
switch x := v.(type) {
case nil:
L.PushNil()
case string:
L.PushString(x)
case bool:
L.PushBoolean(x)
case int:
L.PushNumber(float64(x))
case int64:
L.PushNumber(float64(x))
case float64:
L.PushNumber(x)
case []interface{}:
L.NewTable()
for i, it := range x {
pushGoValue(L, it)
L.RawSeti(-2, i+1)
}
case map[string]interface{}:
L.NewTable()
for k, it := range x {
pushGoValue(L, it)
L.SetField(-2, k)
}
default:
if b, err := json.Marshal(x); err == nil {
var iv interface{}
if json.Unmarshal(b, &iv) == nil {
pushGoValue(L, iv)
return
}
}
L.PushNil()
}
}
func jsonEncode(v interface{}) ([]byte, error) { return json.Marshal(v) }
// luaToJSON converts the Lua value at idx into a Go value via json.encode, then
// unmarshals it into out. It is the bridge used by the plugin manifest/UI
// reader: the plugin returns a plain Lua table, and Go wants a typed struct.
//
// It goes through JSON rather than walking the Lua stack directly because the
// adapter/plugin boundary already speaks JSON everywhere else (transform_request
// gets a JSON string, hooks get a JSON string), so this keeps one representation
// instead of two.
func luaToJSON(L *golua.State, idx int, out interface{}) error {
if L.GetTop() < 1 {
return fmt.Errorf("empty stack")
}
abs := idx
if abs < 0 {
abs = L.GetTop() + 1 + abs
}
if abs < 1 || abs > L.GetTop() {
return fmt.Errorf("index %d out of range (top=%d)", idx, L.GetTop())
}
// Absolute indices throughout: this binding aborts the process (SIGABRT)
// on a bad index rather than panicking, so the stack is captured before any
// push instead of being addressed relative to a shifting top.
//
// json.encode is pushed onto the stack and the value is pushed AFTER it, so
// Call(1, 1) invokes it (Call takes no function index — it calls whatever
// sits below the nargs values).
L.GetGlobal("json")
if L.IsNil(-1) {
L.SetTop(0)
return fmt.Errorf("json global missing")
}
L.GetField(-1, "encode")
if L.Type(-1) != golua.LUA_TFUNCTION {
L.SetTop(0)
return fmt.Errorf("json.encode missing")
}
L.PushValue(abs)
if err := L.Call(1, 1); err != nil {
L.SetTop(0)
return err
}
if L.GetTop() < 1 || L.Type(-1) != golua.LUA_TSTRING {
L.SetTop(0)
return fmt.Errorf("json.encode did not return a string")
}
s := L.ToString(-1)
L.SetTop(0)
return json.Unmarshal([]byte(s), out)
}
func jsonDecode(s string) (interface{}, error) {
var v interface{}
if err := json.Unmarshal([]byte(s), &v); err != nil {
return nil, err
}
return v, nil
}
// AdapterRevision returns a monotonically increasing revision for the named
// adapter's CURRENT source. Every LoadAdapterSource / RemoveAdapter bumps it.
//
// Purpose: callers that cache a decision derived from an adapter's declared
// surface (e.g. "does this adapter declare serialize_chunk?") store the
// revision alongside the cached value and treat a mismatch as invalid. That
// removes the need for those callers to be told when to invalidate — which
// they previously were not, so a plugin rewriting an adapter at runtime could
// leave a stale decision in place forever.
//
// The revision is per-adapter rather than global so an unrelated adapter's
// reload does not discard a still-valid cache entry.
func (v *VM) AdapterRevision(name string) uint64 {
p := v.pool(name)
if p == nil {
return 0
}
p.mu.Lock()
defer p.mu.Unlock()
return p.revision
}
// InstallPluginAdapter registers an adapter ON BEHALF OF A PLUGIN, with the
// one extra rule plugins (and only plugins) are held to: the name must not
// shadow a built-in adapter.
//
// Why plugins are stricter than operators: an operator dropping openai.lua in
// adapter_dir is overriding their own deployment (documented, intended). A
// third-party plugin registering an adapter named "openai" would silently
// swap out the kernel's shipped protocol translation for code the operator
// never reviewed — tampering with the shipped artifact. R6 requires that to
// fail loudly with the conflict named, not succeed quietly.
//
// This is the entry point Phase 3's plugin adapter-install bridge calls.
func (v *VM) InstallPluginAdapter(name, code string) error {
if err := ValidateAdapterName(name, v.bundledNames()); err != nil {
return err
}
return v.LoadAdapterSource(name, code)
}
// AdapterVersion returns the version string an adapter declares, or "" when the
// adapter is not loaded.
//
// It exists so an operator (or a plugin's own UI) can SEE that a runtime
// rewrite happened: after a token rotation the version bumps, and /api/status
// reports it. Without that, a rotation is invisible from outside the process.
func (v *VM) AdapterVersion(name string) string {
p := v.pool(name)
if p == nil {
return ""
}
p.mu.Lock()
defer p.mu.Unlock()
return p.static.version
}
// AdapterIngress returns the client-facing endpoint path an adapter CLAIMS
// (its static.ingress), or "" when it claims none. Exported so operator-facing
// surfaces (and tests) can see which adapter, if any, has taken over parsing of
// an inbound endpoint — the same declaration AdapterForIngress consults.
func (v *VM) AdapterIngress(name string) string {
p := v.pool(name)
if p == nil {
return ""
}
p.mu.Lock()
defer p.mu.Unlock()
return p.static.ingress
}
// EffectiveAdapterName resolves the name an adapter is actually registered
// under, given the key it was loaded with.
//
// A script may declare its own `name`, and the declared one wins at load time.
// Callers that keep per-adapter state (the plugin ownership table, the
// serializer cache) must key it on the EFFECTIVE name, or the state attaches to
// a key nothing will ever look up again.
func (v *VM) EffectiveAdapterName(registeredAs string) string {
if p := v.pool(registeredAs); p != nil {
p.mu.Lock()
if p.static.name != "" {
n := p.static.name
p.mu.Unlock()
return n
}
p.mu.Unlock()
return registeredAs
}
// The script declared a name different from the key it was loaded under, so
// the pool is filed under the DECLARED name. Find it that way, or every
// caller would conclude "not loaded" and re-register a duplicate.
v.mu.RLock()
defer v.mu.RUnlock()
for _, p := range v.pools {
if p.static.registeredAs == registeredAs {
p.mu.Lock()
n := p.static.name
p.mu.Unlock()
if n != "" {
return n
}
return registeredAs
}
}
return registeredAs
}