refactor(lua): plugin cleanup + async hive.sh (#326)

## Summary

Architectural cleanup of `internal/hive/plugins/lua/` plus the bundled
async form of `hive.sh.*`. Bundled because the cleanup includes
extracting an async-handle registry that's only meaningful once a second
async module exists, and the async-sh work was the natural pair.

## Cleanup

- **Test harness consolidation** — replaces five per-module harnesses
(`newJSONHarness`/`newKVHarness`/`newTickerHarness`/`newShHarness` + a
bespoke captureModule for sh) with one `newLuaHarness` + unified
`captureModule` (`helpers_test.go`).
- **Lua↔Go conversion helpers** — promotes `goToLua`/`luaToGo` from
`JSONModule` into `lua_convert.go` so future modules can use them.
`arrayMarker` is now an explicit parameter; callers without an
array-tagging concept pass `nil`. Direct unit tests live in
`lua_convert_test.go`.
- **Async-handle registry extraction** — lifts the duplicated
scaffolding (registry sub-table, handle map, root-context fan-out,
metatable, cancel) shared by ticker and sh into `asyncRegistry`
(`async_registry.go`). Both modules now embed one and use
`Go`/`allocate`/`loadFunction`/`release`/`stop` for per-handle work.
`asyncHandle` is the single shared type.
- **Field-reader precondition** — collapses the four `lua*Field`
preludes (RawGetString, reject LTFunction, type-assert) into
`luaCommandField`.
- **Error convention** — documents when to use `ArgError` vs
`RaiseError` vs the two-return form in `modules.go`; standardises the
`hive.<module>.<fn>:` prefix on KV and Commands errors.
- **Logging convention** — documents per-module log levels in
`modules.go`; `KVModule` gains a `Logger` and now logs Warn on store
errors so operators see them even when the plugin `pcall`s the call.
ShModule's sync/async start/finish/cancel logging is deduped through
`execWithLogging` + `log{Start,Finish,PoolCancelled}`.
- **Dead code + godoc** — drops unused `PluginName` fields on
Ticker/ShModule (only LogModule actually uses its name); backfills godoc
on `Plugin`'s exported methods and the bare module `Register` methods.

## Async hive.sh

- `hive.sh.{run,output,exec}` accept a trailing function callback that
switches the call to async mode: returns a handle immediately, runs the
subprocess on a goroutine bound to a per-handle context, fires the
callback on the dispatcher when finished.
- `handle:cancel()` kills the in-flight subprocess and suppresses the
pending callback. Plugin shutdown cancels every in-flight async call.
- Async output uses a `(stdout, err)` callback shape because it cannot
raise from the post-return callback; sync output still raises on
non-zero exit.
- Documented in `docs/configuration/plugins.md`; integration coverage in
`test/integration/lua_plugin_sh_test.go`.

## Test plan

- [x] `go test -count=1 -race ./internal/hive/plugins/lua/...` passes
- [x] `golangci-lint run ./internal/hive/plugins/lua/...` is clean
- [x] `go build -tags integration ./test/integration/...` compiles
- [ ] CI green

Author: hay-kot

docs/configuration/plugins.md

@@ -271,23 +271,54 @@ Run shell commands from Lua. All three functions execute via `sh -c` and share t
 !!! danger "Trust boundary"
     `hive.sh` runs commands with the full privileges of the Hive process. Only enable Lua plugins from sources you trust — installing one is equivalent to running its shell commands as your user.
 
-| Function | Purpose |
-| -------- | ------- |
-| `hive.sh.run(cmd)` | Run `cmd`. Returns the exit code. Never raises. |
-| `hive.sh.output(cmd)` | Run `cmd`, return stdout with one trailing newline stripped. Raises on non-zero exit or executor error. |
-| `hive.sh.exec(cmd[, opts])` | Run `cmd` with optional `{cwd, timeout}`. Returns `{stdout, stderr, code, err}`. Never raises. |
+Every `hive.sh.*` call is async: the call returns a handle immediately and the supplied callback fires on the dispatcher when the subprocess finishes. The callback is **required** — calling without one raises a Lua error.
+
+| Function | Callback signature | Notes |
+| -------- | ------------------ | ----- |
+| `hive.sh.run(cmd, fn)` | `fn(code)` | Receives the exit code. |
+| `hive.sh.output(cmd, fn)` | `fn(stdout, err)` | On success: `stdout` is trimmed of one trailing newline and `err` is nil. On non-zero exit or executor error: `stdout` is nil and `err` is a string with the exit code and stderr. |
+| `hive.sh.exec(cmd[, opts], fn)` | `fn(result)` | `result` is `{stdout, stderr, code, err}`. |
 
 `opts.timeout` is in seconds (number). `opts.cwd` overrides the working directory for that call only; an empty string inherits Hive's working directory.
 
+Each call returns a handle with a `:cancel()` method that kills the in-flight subprocess and suppresses the pending callback.
+
 ```lua
 return function(hive)
-  local code = hive.sh.run("git fetch origin")          -- exit code only
-  local head = hive.sh.output("git rev-parse HEAD")     -- stdout, raises on non-zero
-  local r = hive.sh.exec("ls -la", { cwd = "/tmp", timeout = 5 })
-  -- r.stdout, r.stderr, r.code, r.err
+  hive.sh.run("git fetch origin", function(code)
+    hive.log.info("fetch exited with " .. code)
+  end)
+
+  hive.sh.output("git rev-parse HEAD", function(head, err)
+    if err ~= nil then
+      hive.log.warn("rev-parse failed: " .. err)
+      return
+    end
+    hive.log.info("HEAD is " .. head)
+  end)
+
+  hive.sh.exec("ls -la", { cwd = "/tmp", timeout = 5 }, function(r)
+    hive.log.info("listed " .. r.stdout)
+  end)
+
+  local handle = hive.sh.run("sleep 600", function(_) end)
+  hive.ticker.after("5s", function() handle:cancel() end)
 end
 ```
 
+!!! note "Callbacks run on the dispatcher"
+    Every callback runs on the same Lua dispatcher as ticker fires and
+    the plugin entrypoint. A slow callback serialises later Lua work in
+    that plugin — keep callbacks short or hand long work off via another
+    `hive.sh.*` call.
+
+!!! warning "Shutdown cancels in-flight calls"
+    Plugin reload or `hive` shutdown cancels every in-flight call: the
+    per-handle context is cancelled, the subprocess is killed, and the
+    pinned callback is released for GC. Pending callbacks may still fire
+    once with a non-zero `code` and a cancellation `err`, but never
+    after the runtime has fully closed.
+
 !!! note "Shared shell pool"
     `hive.sh.*` calls draw from the same `plugins.shell_workers` budget as plugin status refreshes, so a slow shell command from Lua can delay other plugins. The default per-call timeout is 30s; tune it via `plugins.lua.shell_timeout`. Plugin shutdown cancels every in-flight command.
 

internal/hive/plugins/lua/async_registry.go

@@ -0,0 +1,294 @@
+package lua
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	glua "github.com/yuin/gopher-lua"
+)
+
+// asyncShutdownGracePeriod bounds how long shutdown waits for in-flight
+// goroutines to finish naturally before force-cancelling. Long enough
+// for fast plugin entrypoint chains (a few subprocess calls) to drain
+// at app exit; short enough that a stuck call can't stall shutdown
+// indefinitely. After the grace period elapses, rootCtx is cancelled
+// and any remaining work aborts via the per-handle ctx.
+const asyncShutdownGracePeriod = time.Second
+
+// asyncRegistry holds the Lua-side and Go-side bookkeeping shared by
+// every async host module (hive.ticker, hive.sh, ...). Each module owns
+// one registry, initialises it during Register, and tears it down via
+// shutdown.
+//
+// The registry handles:
+//
+//   - A per-module sub-table in the Lua registry that pins LFunctions
+//     so callbacks survive Lua GC while a goroutine still references
+//     them.
+//   - A handle map keyed by id, storing the per-handle context.CancelFunc
+//     so any goroutine can cancel a specific running operation.
+//   - A root context whose cancellation fans out to every per-handle
+//     context, used for module shutdown.
+//   - A WaitGroup that lets shutdown block until in-flight goroutines
+//     finish.
+//   - The shared __index metatable installed on every handle userdata.
+//
+// Module code stays small: allocate to start a new operation, loadFunction
+// to invoke its callback on the dispatcher, release after a successful
+// dispatch, and stop or asyncHandle.Cancel for any-goroutine cancellation.
+type asyncRegistry struct {
+	cfg asyncRegistryConfig
+
+	// registryKey namespaces the per-module sub-table in the Lua
+	// registry. The instance pointer is appended so multiple modules of
+	// the same kind on one LState don't stomp each other.
+	registryKey string
+
+	// nextID is also the per-handle registry sub-key, stringified.
+	nextID atomic.Uint64
+
+	mu      sync.Mutex
+	handles map[uint64]context.CancelFunc
+
+	rootCtx    context.Context
+	rootCancel context.CancelFunc
+
+	wg sync.WaitGroup
+
+	closeOnce sync.Once
+}
+
+// asyncRegistryConfig configures the namespacing and Lua-facing names
+// for an asyncRegistry. KeyPrefix should end in a "." for readability.
+type asyncRegistryConfig struct {
+	// KeyPrefix prefixes the per-module Lua registry sub-table key.
+	// Example: "hive.ticker."
+	KeyPrefix string
+	// MetatableName names the Lua metatable installed on every handle
+	// userdata. Example: "hive.ticker.handle"
+	MetatableName string
+}
+
+// asyncHandle is the Go-side state behind every async userdata returned
+// to Lua. Cancel is idempotent, safe from any goroutine, and routes
+// through the registry to delete the map entry, cancel the per-handle
+// context, and release the registry pin.
+type asyncHandle struct {
+	id       uint64
+	once     sync.Once
+	registry *asyncRegistry
+	runtime  *Runtime
+}
+
+// init prepares the registry for use. modulePtr is fmt-formatted with
+// %p to make registryKey unique per module instance (so two modules of
+// the same kind on one LState don't collide). Must run on the
+// dispatcher.
+func (r *asyncRegistry) init(state *glua.LState, modulePtr any, cfg asyncRegistryConfig) error {
+	r.cfg = cfg
+	r.handles = map[uint64]context.CancelFunc{}
+	r.rootCtx, r.rootCancel = context.WithCancel(context.Background())
+	r.registryKey = fmt.Sprintf("%s%p", cfg.KeyPrefix, modulePtr)
+
+	registry, ok := state.Get(glua.RegistryIndex).(*glua.LTable)
+	if !ok {
+		return fmt.Errorf("%slua registry is not a table", cfg.KeyPrefix)
+	}
+	state.SetField(registry, r.registryKey, state.NewTable())
+
+	mt := state.NewTypeMetatable(cfg.MetatableName)
+	state.SetField(mt, "__index", state.SetFuncs(state.NewTable(), map[string]glua.LGFunction{
+		"cancel": r.luaCancel,
+	}))
+	return nil
+}
+
+// shutdown drains in-flight goroutines before force-cancelling. First
+// gives the chain up to asyncShutdownGracePeriod to finish naturally so
+// fast callback chains spawned during plugin Init complete cleanly;
+// only then cancels rootCtx and waits for stragglers. Clears the handle
+// map. Idempotent.
+func (r *asyncRegistry) shutdown() {
+	r.closeOnce.Do(func() {
+		// Phase 1: graceful drain. Wait up to asyncShutdownGracePeriod
+		// for the chain to complete naturally. spawnAsync's goroutines
+		// hold the wg until their dispatch (and any callback-spawned
+		// work) finishes, so wg.Wait covers the full callback chain.
+		drained := make(chan struct{})
+		go func() {
+			r.wg.Wait()
+			close(drained)
+		}()
+
+		select {
+		case <-drained:
+		case <-time.After(asyncShutdownGracePeriod):
+			// Phase 2: force-cancel and wait for stragglers.
+			if r.rootCancel != nil {
+				r.rootCancel()
+			}
+			<-drained
+		}
+
+		// Cancel rootCtx unconditionally so any goroutine that races
+		// past the drain (none should, but defensively) sees a closed
+		// context if it inspects ctx after spawnAsync returns.
+		if r.rootCancel != nil {
+			r.rootCancel()
+		}
+
+		r.mu.Lock()
+		r.handles = map[uint64]context.CancelFunc{}
+		r.mu.Unlock()
+	})
+}
+
+// Go starts fn in a goroutine tracked by the registry's WaitGroup so
+// shutdown blocks until fn completes. Mirrors sync.WaitGroup.Go;
+// capitalised because lowercase `go` is a reserved keyword.
+func (r *asyncRegistry) Go(fn func()) { r.wg.Go(fn) }
+
+// allocate pins fn under a fresh handle id, derives a per-handle context
+// from rootCtx, and records the cancel func in the handle map. Returns
+// the id and ctx for the goroutine to bind to. Must run on the
+// dispatcher.
+func (r *asyncRegistry) allocate(state *glua.LState, fn *glua.LFunction) (uint64, context.Context) {
+	id := r.nextID.Add(1)
+	ctx, cancel := context.WithCancel(r.rootCtx)
+
+	r.storeFunction(state, id, fn)
+
+	r.mu.Lock()
+	r.handles[id] = cancel
+	r.mu.Unlock()
+
+	return id, ctx
+}
+
+// loadFunction returns the pinned function for id, or nil if already
+// released. Must run on the dispatcher.
+func (r *asyncRegistry) loadFunction(state *glua.LState, id uint64) *glua.LFunction {
+	table := r.registryTable(state)
+	if table == nil {
+		return nil
+	}
+	fn, ok := state.GetField(table, strconv.FormatUint(id, 10)).(*glua.LFunction)
+	if !ok {
+		return nil
+	}
+	return fn
+}
+
+// release drops the handle map entry and registry pin for id. Used after
+// a dispatch path has consumed the callback. Safe even if id is unknown.
+// Must run on the dispatcher.
+func (r *asyncRegistry) release(state *glua.LState, id uint64) {
+	r.mu.Lock()
+	delete(r.handles, id)
+	r.mu.Unlock()
+
+	r.releaseFunction(state, id)
+}
+
+// stop cancels the per-handle context, removes the map entry, and
+// schedules the registry pin release on the dispatcher. Safe from any
+// goroutine; a missing id is treated as already released.
+func (r *asyncRegistry) stop(rt *Runtime, id uint64) {
+	r.mu.Lock()
+	cancel, ok := r.handles[id]
+	if ok {
+		delete(r.handles, id)
+	}
+	r.mu.Unlock()
+	if !ok {
+		return
+	}
+	cancel()
+
+	rt.Submit(func(state *glua.LState) {
+		r.releaseFunction(state, id)
+	})
+}
+
+// handleUserData wraps value in *LUserData with the registry's metatable.
+// Must run on the dispatcher.
+func (r *asyncRegistry) handleUserData(state *glua.LState, value any) *glua.LUserData {
+	ud := state.NewUserData()
+	ud.Value = value
+	state.SetMetatable(ud, state.GetTypeMetatable(r.cfg.MetatableName))
+	return ud
+}
+
+// newHandle returns a fresh asyncHandle wired to this registry and the
+// supplied runtime. Pair with allocate to get the id.
+func (r *asyncRegistry) newHandle(id uint64, rt *Runtime) *asyncHandle {
+	return &asyncHandle{id: id, registry: r, runtime: rt}
+}
+
+// luaCancel is the __index method bound to handle userdata. Type-asserts
+// the userdata as *asyncHandle (which both ticker and sh use) and routes
+// to Cancel.
+func (r *asyncRegistry) luaCancel(state *glua.LState) int {
+	ud := state.CheckUserData(1)
+	h, ok := ud.Value.(*asyncHandle)
+	if !ok {
+		state.ArgError(1, "expected "+r.cfg.MetatableName+" handle")
+		return 0
+	}
+	h.Cancel()
+	return 0
+}
+
+// storeFunction pins fn so Lua cannot GC it while the goroutine holds a
+// reference. Must run on the dispatcher.
+func (r *asyncRegistry) storeFunction(state *glua.LState, id uint64, fn *glua.LFunction) {
+	table := r.registryTable(state)
+	if table == nil {
+		return
+	}
+	state.SetField(table, strconv.FormatUint(id, 10), fn)
+}
+
+// releaseFunction drops the registry pin so Lua can GC the LFunction.
+// Must run on the dispatcher.
+func (r *asyncRegistry) releaseFunction(state *glua.LState, id uint64) {
+	table := r.registryTable(state)
+	if table == nil {
+		return
+	}
+	state.SetField(table, strconv.FormatUint(id, 10), glua.LNil)
+}
+
+// registryTable returns this registry's pinning sub-table, or nil if
+// init has not run.
+func (r *asyncRegistry) registryTable(state *glua.LState) *glua.LTable {
+	registry, ok := state.Get(glua.RegistryIndex).(*glua.LTable)
+	if !ok {
+		return nil
+	}
+	table, ok := state.GetField(registry, r.registryKey).(*glua.LTable)
+	if !ok {
+		return nil
+	}
+	return table
+}
+
+// Cancel stops the per-handle goroutine, removes the map entry, and
+// schedules the registry pin release on the dispatcher. Idempotent and
+// safe from any goroutine.
+func (h *asyncHandle) Cancel() {
+	h.once.Do(func() {
+		h.registry.stop(h.runtime, h.id)
+	})
+}
+
+// poison consumes the once so a later Cancel is a true no-op. Used by
+// post-dispatch cleanup paths that already released via the dispatcher
+// and want to suppress redundant Cancel work.
+func (h *asyncHandle) poison() {
+	h.once.Do(func() {})
+}