`
+// block routes through the dispatcher instead of the local registry.
+func TestWalker_PluginBlockDispatched(t *testing.T) {
+ reg := NewRegistry()
+ disp := &fakeDispatcher{resp: template.HTML("from plugin
")}
+ w := New(reg).WithPluginDispatcher(disp)
+
+ tree := BlockTree{
+ {Type: "plugin/seo/sitemap-link", Attributes: map[string]any{"slug": "home"}},
+ }
+ res := w.Walk(tree, Context{"postId": 42})
+
+ if len(res.Errors) != 0 {
+ t.Fatalf("unexpected errors: %v", res.Errors)
+ }
+ if !strings.Contains(string(res.HTML), "from plugin") {
+ t.Errorf("html: got %q", res.HTML)
+ }
+ if len(disp.calls) != 1 {
+ t.Fatalf("dispatch calls: got %d want 1", len(disp.calls))
+ }
+ got := disp.calls[0]
+ if got.BlockType != "plugin/seo/sitemap-link" {
+ t.Errorf("block type: got %q", got.BlockType)
+ }
+ if got.Attributes["slug"] != "home" {
+ t.Errorf("attrs: got %v", got.Attributes)
+ }
+ if got.Context["postId"] != 42 {
+ t.Errorf("context: got %v", got.Context)
+ }
+}
+
+// TestWalker_PluginBlockInnerRendered verifies the dispatcher receives
+// already-rendered children HTML.
+func TestWalker_PluginBlockInnerRendered(t *testing.T) {
+ reg := NewRegistry()
+ reg.MustRegister("core/paragraph", BlockSpec{
+ Render: func(b Block, inner template.HTML, ctx Context) (template.HTML, error) {
+ return template.HTML("" + attrString(b.Attributes, "text", "") + "
"), nil
+ },
+ })
+ disp := &fakeDispatcher{resp: template.HTML("")}
+ w := New(reg).WithPluginDispatcher(disp)
+
+ tree := BlockTree{
+ {
+ Type: "plugin/blog/featured",
+ Attributes: map[string]any{},
+ InnerBlocks: []Block{
+ {Type: "core/paragraph", Attributes: map[string]any{"text": "child"}},
+ },
+ },
+ }
+ w.Walk(tree, nil)
+ if len(disp.calls) != 1 {
+ t.Fatalf("calls: %d", len(disp.calls))
+ }
+ if !strings.Contains(disp.calls[0].Inner, "child
") {
+ t.Errorf("inner: got %q", disp.calls[0].Inner)
+ }
+}
+
+// TestWalker_PluginBlockErrorDegrades shows that a dispatcher error
+// degrades to the render-error placeholder without taking the whole
+// page down.
+func TestWalker_PluginBlockErrorDegrades(t *testing.T) {
+ reg := NewRegistry()
+ disp := &fakeDispatcher{err: errors.New("plugin blew up")}
+ w := New(reg).WithPluginDispatcher(disp)
+
+ tree := BlockTree{
+ {Type: "plugin/foo/bar"},
+ }
+ res := w.Walk(tree, nil)
+ if len(res.Errors) != 1 {
+ t.Fatalf("errors: got %d want 1", len(res.Errors))
+ }
+ if !strings.Contains(string(res.HTML), "gn-block-error") {
+ t.Errorf("html: got %q", res.HTML)
+ }
+}
+
+// TestWalker_PluginBlockUnknownFallbackWhenNoDispatcher confirms a
+// plugin block falls through to the standard "unknown" placeholder
+// when no dispatcher is wired.
+func TestWalker_PluginBlockUnknownFallbackWhenNoDispatcher(t *testing.T) {
+ reg := NewRegistry()
+ w := New(reg) // no dispatcher
+
+ tree := BlockTree{{Type: "plugin/foo/bar"}}
+ res := w.Walk(tree, nil)
+ if len(res.Errors) != 1 {
+ t.Fatalf("errors: got %d want 1", len(res.Errors))
+ }
+ if !errors.Is(res.Errors[0].Err, ErrUnknownBlockType) {
+ t.Errorf("err: %v", res.Errors[0].Err)
+ }
+}
+
+// TestParsePluginBlockType handles edge cases.
+func TestParsePluginBlockType(t *testing.T) {
+ cases := []struct {
+ in string
+ slug, handler string
+ ok bool
+ }{
+ {"plugin/seo/sitemap", "seo", "sitemap", true},
+ {"plugin/foo/bar/baz", "foo", "bar/baz", true},
+ {"core/paragraph", "", "", false},
+ {"plugin/seo", "", "", false},
+ {"plugin//bar", "", "", false},
+ {"plugin/seo/", "", "", false},
+ {"", "", "", false},
+ }
+ for _, tc := range cases {
+ s, h, ok := parsePluginBlockType(tc.in)
+ if ok != tc.ok || s != tc.slug || h != tc.handler {
+ t.Errorf("%q: got (%q,%q,%v) want (%q,%q,%v)",
+ tc.in, s, h, ok, tc.slug, tc.handler, tc.ok)
+ }
+ }
+}
+
+// TestHookBusDispatcher_RoundTrip wires the real hook bus to a plugin
+// stub that returns marshalled HTML; the dispatcher must unwrap it
+// and return template.HTML.
+func TestHookBusDispatcher_RoundTrip(t *testing.T) {
+ bus := hooks.NewBus()
+ // Plugin "myplug" subscribes to its sitemap block.
+ bus.RegisterFilter("block.render:myplug/card", 10,
+ func(ctx context.Context, value any, args ...any) (any, error) {
+ // Read the incoming request to assert wiring.
+ raw, ok := value.(json.RawMessage)
+ if !ok {
+ t.Errorf("plugin: value type %T", value)
+ }
+ var req PluginBlockRequest
+ if err := json.Unmarshal(raw, &req); err != nil {
+ t.Errorf("plugin: unmarshal: %v", err)
+ }
+ if req.BlockType != "plugin/myplug/card" {
+ t.Errorf("block type: %q", req.BlockType)
+ }
+ // Return HTML as a json-encoded string.
+ out, _ := json.Marshal("hello")
+ return json.RawMessage(out), nil
+ })
+
+ disp := NewHookBusDispatcher(context.Background(), bus)
+ w := New(NewRegistry()).WithPluginDispatcher(disp)
+ tree := BlockTree{{Type: "plugin/myplug/card", Attributes: map[string]any{}}}
+ res := w.Walk(tree, nil)
+ if len(res.Errors) != 0 {
+ t.Fatalf("errs: %v", res.Errors)
+ }
+ if !strings.Contains(string(res.HTML), "hello") {
+ t.Errorf("html: %q", res.HTML)
+ }
+}
+
+// TestHookBusDispatcher_BusError surfaces the bus error.
+func TestHookBusDispatcher_BusError(t *testing.T) {
+ bus := hooks.NewBus()
+ want := errors.New("plugin trap")
+ bus.RegisterFilter("block.render:p/h", 10,
+ func(ctx context.Context, value any, args ...any) (any, error) {
+ return value, want
+ })
+ disp := NewHookBusDispatcher(context.Background(), bus)
+ _, err := disp.Dispatch("p", "h", PluginBlockRequest{})
+ if !errors.Is(err, want) {
+ t.Errorf("err: %v", err)
+ }
+}
diff --git a/packages/go/blocks/render/plugin_dispatcher.go b/packages/go/blocks/render/plugin_dispatcher.go
new file mode 100644
index 00000000..8509a554
--- /dev/null
+++ b/packages/go/blocks/render/plugin_dispatcher.go
@@ -0,0 +1,124 @@
+// plugin_dispatcher.go wires the walker's plugin-block path to the
+// host hook bus (issue #222). The walker calls Dispatch with a
+// (slug, handler, request) tuple; this dispatcher translates that into
+// an ApplyFilters call on the bus under the canonical hook key
+//
+// block.render:{slug}/{handler}
+//
+// The plugin's WASM module subscribes to that key, decodes the JSON
+// payload (a PluginBlockRequest), and returns the rendered HTML
+// string. The dispatcher unmarshals the bus return value into a
+// template.HTML and hands it to the walker, which splices it into the
+// output stream.
+//
+// The bus dependency is injected via a tiny interface so this package
+// doesn't take a hard import on packages/go/hooks — the cycle would
+// be: hooks → schemas → blocks (eventually) → hooks. The Dispatcher
+// interface keeps the render package's import graph small.
+
+package render
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "html/template"
+)
+
+// HookFilterBus is the subset of *hooks.Bus this dispatcher needs.
+// Declared here (rather than imported) so the render package stays
+// free of a compile-time dependency on packages/go/hooks. The
+// production wiring passes *hooks.Bus, which satisfies this interface
+// without any adapter.
+type HookFilterBus interface {
+ ApplyFilters(ctx context.Context, name string, value any, args ...any) (any, error)
+}
+
+// HookBusDispatcher implements PluginBlockDispatcher by issuing a
+// filter call on the host hook bus. Construct via NewHookBusDispatcher.
+//
+// The Context the bus call uses is captured at construction time —
+// callers wanting per-request cancellation should construct one
+// dispatcher per request, or pass a longer-lived context (e.g. the
+// HTTP handler's). The walker has no Context surface of its own;
+// adding one would touch every renderer signature.
+type HookBusDispatcher struct {
+ bus HookFilterBus
+ ctx context.Context
+}
+
+// NewHookBusDispatcher constructs a dispatcher rooted at the supplied
+// bus and context. ctx must be non-nil; pass context.Background() for
+// the long-lived dispatcher used by SSR rendering.
+func NewHookBusDispatcher(ctx context.Context, bus HookFilterBus) *HookBusDispatcher {
+ if bus == nil {
+ panic("render.NewHookBusDispatcher: bus is required")
+ }
+ if ctx == nil {
+ ctx = context.Background()
+ }
+ return &HookBusDispatcher{bus: bus, ctx: ctx}
+}
+
+// Dispatch fires the bus filter and unwraps the result into HTML.
+//
+// Hook name format: "block.render:/". Plugins
+// subscribe under this exact key; the colon separator is chosen so
+// the slug+handler segment can hold "/" without ambiguity (the
+// walker splits the block type on "/" inside the plugin/ namespace,
+// but here we collapse them again for a single hook key).
+//
+// Bus return contract: the plugin handler returns a value the bus
+// threads as the new "value". The walker accepts any of these
+// equivalent shapes:
+//
+// - template.HTML (the canonical case)
+// - string
+// - []byte
+// - json.RawMessage
+//
+// Any other type is rejected with a wrapped error so a misbehaving
+// handler can't silently inject a Go-marshalled struct as page
+// content.
+func (d *HookBusDispatcher) Dispatch(slug, handler string, req PluginBlockRequest) (template.HTML, error) {
+ hookName := "block.render:" + slug + "/" + handler
+
+ payload, err := json.Marshal(req)
+ if err != nil {
+ return "", fmt.Errorf("render: marshal plugin-block request: %w", err)
+ }
+
+ out, err := d.bus.ApplyFilters(d.ctx, hookName, json.RawMessage(payload))
+ if err != nil {
+ return "", fmt.Errorf("render: plugin block %q/%q: %w", slug, handler, err)
+ }
+
+ switch v := out.(type) {
+ case template.HTML:
+ return v, nil
+ case string:
+ return template.HTML(v), nil
+ case []byte:
+ return template.HTML(v), nil
+ case json.RawMessage:
+ // Unwrap a JSON-string payload: plugins commonly return
+ // json.Marshal("...") rather than the raw bytes.
+ // If the value is a JSON string, decode it; otherwise treat
+ // the raw bytes as HTML.
+ if len(v) > 0 && v[0] == '"' {
+ var s string
+ if jerr := json.Unmarshal(v, &s); jerr == nil {
+ return template.HTML(s), nil
+ }
+ }
+ return template.HTML(v), nil
+ case nil:
+ return "", errors.New("render: plugin block returned nil")
+ default:
+ return "", fmt.Errorf("render: plugin block returned unsupported type %T", out)
+ }
+}
+
+// Compile-time check.
+var _ PluginBlockDispatcher = (*HookBusDispatcher)(nil)
diff --git a/packages/go/blocks/render/walker.go b/packages/go/blocks/render/walker.go
index 9a49b42e..08281778 100644
--- a/packages/go/blocks/render/walker.go
+++ b/packages/go/blocks/render/walker.go
@@ -147,6 +147,48 @@ func (e WalkError) Error() string {
// Unwrap exposes the underlying error for errors.Is / errors.As.
func (e WalkError) Unwrap() error { return e.Err }
+// PluginBlockDispatcher is the seam through which the walker resolves
+// a `plugin//` block to its plugin-supplied HTML. The
+// production wiring satisfies it by calling into the host hook bus
+// (issue #222): the walker fires the ApplyFilters call
+// "block.render:{slug}/{handler}", the plugin's WASM handler returns
+// the rendered HTML, and the result is spliced into the output stream
+// in place of the block's normal renderer output.
+//
+// Dispatch is called once per plugin block encountered during a walk.
+// slug is the plugin owning the block; handler is the per-plugin block
+// name. The req payload carries the block's attributes, the inner
+// HTML already rendered for children, and the consumed context — JSON
+// shape documented in PluginBlockRequest.
+//
+// Returning an error degrades the block to a render-error placeholder
+// the same way a regular renderer error does. The returned bytes are
+// trusted as already-safe HTML; the dispatcher contract is that the
+// plugin sandbox produces only host-allowed markup (SafeHTML policies
+// live in packages/go/safehtml and are applied INSIDE the WASM host
+// shim, not by the walker).
+type PluginBlockDispatcher interface {
+ Dispatch(slug, handler string, req PluginBlockRequest) (template.HTML, error)
+}
+
+// PluginBlockRequest is the wire payload sent to a plugin's
+// block-render handler. Mirrors the editor-side BlockRenderProps so
+// plugin authors writing both halves see one shape.
+type PluginBlockRequest struct {
+ // BlockType is the full namespaced type ("plugin/seo/sitemap-link").
+ BlockType string `json:"blockType"`
+ // Attributes is the block's persisted attribute bag.
+ Attributes map[string]any `json:"attributes,omitempty"`
+ // Inner is the already-rendered HTML for the block's children.
+ // Plugins compose this with their own markup; the walker has
+ // already escaped it so the plugin must NOT re-escape.
+ Inner string `json:"inner,omitempty"`
+ // Context is the consumed-context map (filtered by the spec's
+ // UsesContext list). Empty when the block didn't opt into any
+ // context keys.
+ Context map[string]any `json:"context,omitempty"`
+}
+
// Walker renders a BlockTree against a Registry.
//
// Walker is stateless besides its Registry pointer — Walk may be
@@ -155,7 +197,8 @@ func (e WalkError) Unwrap() error { return e.Err }
// mutating it while a walk is in flight, but doing so is not a data
// race in the Go memory-model sense.
type Walker struct {
- registry *Registry
+ registry *Registry
+ pluginDispatcher PluginBlockDispatcher
}
// New constructs a Walker bound to the given Registry. The registry
@@ -169,6 +212,15 @@ func New(reg *Registry) *Walker {
return &Walker{registry: reg}
}
+// WithPluginDispatcher attaches a PluginBlockDispatcher so the walker
+// can resolve `plugin//` block types. Without one, plugin
+// blocks fall through to the ErrUnknownBlockType placeholder — the
+// same behaviour callers see for any unregistered type.
+func (w *Walker) WithPluginDispatcher(d PluginBlockDispatcher) *Walker {
+ w.pluginDispatcher = d
+ return w
+}
+
// Walk renders the given tree with the supplied root context.
//
// The walk is depth-first: each block's InnerBlocks are rendered
@@ -210,6 +262,16 @@ func (w *Walker) Walk(tree BlockTree, ctx Context) WalkResult {
// renderer. This mirrors the TS canvas's filterConsumedContext /
// resolveProvidedContext flow.
func (w *Walker) walkBlock(block Block, inherited Context, path string) (template.HTML, []WalkError) {
+ // Plugin-block fast path. Block types of the form
+ // `plugin//` are dispatched to the plugin's WASM
+ // handler via the hook bus rather than through the local registry.
+ // We still recurse into InnerBlocks first so the plugin handler
+ // receives already-rendered children, matching the semantics of
+ // normal block renderers.
+ if slug, handler, ok := parsePluginBlockType(block.Type); ok && w.pluginDispatcher != nil {
+ return w.dispatchPluginBlock(block, inherited, path, slug, handler)
+ }
+
spec, ok := w.registry.Get(block.Type)
if !ok {
err := WalkError{
@@ -253,6 +315,71 @@ func (w *Walker) walkBlock(block Block, inherited Context, path string) (templat
return out, errs
}
+// parsePluginBlockType cracks "plugin//" into its two
+// parts. The split returns ok=false for any non-plugin block type so
+// the walker's fast-path bailout is a single string-compare in the
+// common case.
+//
+// The handler part may itself contain "/"; we split on the first two
+// segments only so a plugin block named "plugin/seo/listing/card" is
+// dispatched as slug="seo", handler="listing/card".
+func parsePluginBlockType(blockType string) (slug, handler string, ok bool) {
+ const prefix = "plugin/"
+ if !strings.HasPrefix(blockType, prefix) {
+ return "", "", false
+ }
+ rest := blockType[len(prefix):]
+ slash := strings.IndexByte(rest, '/')
+ if slash <= 0 || slash == len(rest)-1 {
+ return "", "", false
+ }
+ return rest[:slash], rest[slash+1:], true
+}
+
+// dispatchPluginBlock walks InnerBlocks depth-first (so the plugin
+// handler receives already-rendered children), assembles the
+// PluginBlockRequest, and routes it through the configured
+// PluginBlockDispatcher. The returned HTML replaces the block's
+// position in the output stream; errors degrade to the same
+// render-error placeholder a regular renderer error produces, so the
+// page stays alive when one plugin block misbehaves.
+func (w *Walker) dispatchPluginBlock(block Block, inherited Context, path, slug, handler string) (template.HTML, []WalkError) {
+ // Render inner blocks first so the plugin receives them already
+ // composed. We do NOT pass plugin-provided context here because
+ // the manifest's ProvidesContext mechanism doesn't apply to plugin
+ // blocks (they own their entire render); children inherit the
+ // upstream context unchanged.
+ var innerHTML template.HTML
+ var errs []WalkError
+ if len(block.InnerBlocks) > 0 {
+ var inner strings.Builder
+ for i, child := range block.InnerBlocks {
+ childPath := fmt.Sprintf("%s/innerBlocks/%d", path, i)
+ out, childErrs := w.walkBlock(child, inherited, childPath)
+ inner.WriteString(string(out))
+ errs = append(errs, childErrs...)
+ }
+ innerHTML = template.HTML(inner.String())
+ }
+
+ req := PluginBlockRequest{
+ BlockType: block.Type,
+ Attributes: block.Attributes,
+ Inner: string(innerHTML),
+ Context: inherited,
+ }
+ html, err := w.pluginDispatcher.Dispatch(slug, handler, req)
+ if err != nil {
+ errs = append(errs, WalkError{
+ Path: path,
+ BlockType: block.Type,
+ Err: err,
+ })
+ return renderErrorHTML(block.Type, err), errs
+ }
+ return html, errs
+}
+
// mergeProvidedContext layers the block's ProvidesContext values on
// top of the inherited map, allocating a fresh map only when the
// block actually contributes new values. Otherwise the inherited map
diff --git a/packages/go/hooks/batch_test.go b/packages/go/hooks/batch_test.go
new file mode 100644
index 00000000..9a32eccc
--- /dev/null
+++ b/packages/go/hooks/batch_test.go
@@ -0,0 +1,242 @@
+package hooks
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "testing"
+)
+
+// TestApplyBatch_NoHandlers returns the input slice untouched and no
+// error — symmetric with ApplyFilters on a name nothing subscribes to.
+func TestApplyBatch_NoHandlers(t *testing.T) {
+ bus, _ := newTestBus(t)
+ in := []any{"a", "b", "c"}
+ out, err := bus.ApplyBatch(context.Background(), "noone.subscribes", in)
+ if err != nil {
+ t.Fatalf("ApplyBatch: %v", err)
+ }
+ if len(out) != 3 || out[0] != "a" || out[1] != "b" || out[2] != "c" {
+ t.Errorf("out: got %v want [a b c]", out)
+ }
+}
+
+// TestApplyBatch_BatchAwareHandler routes the whole slice through a
+// BatchFilterHandler in one call.
+func TestApplyBatch_BatchAwareHandler(t *testing.T) {
+ bus, _ := newTestBus(t)
+ var callCount int
+ bus.RegisterBatchFilter("titles", 10, func(ctx context.Context, items []any, args ...any) ([]any, error) {
+ callCount++
+ out := make([]any, len(items))
+ for i, v := range items {
+ out[i] = strings.ToUpper(v.(string))
+ }
+ return out, nil
+ })
+
+ out, err := bus.ApplyBatch(context.Background(), "titles", []any{"hello", "world", "foo"})
+ if err != nil {
+ t.Fatalf("ApplyBatch: %v", err)
+ }
+ if callCount != 1 {
+ t.Errorf("batch handler invocation count: got %d want 1", callCount)
+ }
+ if out[0] != "HELLO" || out[1] != "WORLD" || out[2] != "FOO" {
+ t.Errorf("out: got %v", out)
+ }
+}
+
+// TestApplyBatch_LegacyFilterHandler loops a plain FilterHandler over
+// each item so existing subscribers keep working inside a batched chain.
+func TestApplyBatch_LegacyFilterHandler(t *testing.T) {
+ bus, _ := newTestBus(t)
+ var perItem int
+ bus.RegisterFilter("titles", 10, func(ctx context.Context, value any, args ...any) (any, error) {
+ perItem++
+ return value.(string) + "!", nil
+ })
+
+ out, err := bus.ApplyBatch(context.Background(), "titles", []any{"a", "b", "c"})
+ if err != nil {
+ t.Fatalf("ApplyBatch: %v", err)
+ }
+ if perItem != 3 {
+ t.Errorf("legacy filter calls: got %d want 3", perItem)
+ }
+ if out[0] != "a!" || out[1] != "b!" || out[2] != "c!" {
+ t.Errorf("out: got %v", out)
+ }
+}
+
+// TestApplyBatch_MixedChain interleaves a batch-aware handler with a
+// legacy per-item one. Priorities are respected.
+func TestApplyBatch_MixedChain(t *testing.T) {
+ bus, _ := newTestBus(t)
+ var batchCalls, legacyCalls int
+
+ bus.RegisterFilter("mix", 20, func(ctx context.Context, value any, args ...any) (any, error) {
+ legacyCalls++
+ return value.(string) + ".legacy", nil
+ })
+ bus.RegisterBatchFilter("mix", 10, func(ctx context.Context, items []any, args ...any) ([]any, error) {
+ batchCalls++
+ out := make([]any, len(items))
+ for i, v := range items {
+ out[i] = v.(string) + ".batch"
+ }
+ return out, nil
+ })
+
+ out, err := bus.ApplyBatch(context.Background(), "mix", []any{"x", "y"})
+ if err != nil {
+ t.Fatalf("ApplyBatch: %v", err)
+ }
+ if batchCalls != 1 {
+ t.Errorf("batch calls: got %d want 1", batchCalls)
+ }
+ if legacyCalls != 2 {
+ t.Errorf("legacy calls: got %d want 2", legacyCalls)
+ }
+ if out[0] != "x.batch.legacy" || out[1] != "y.batch.legacy" {
+ t.Errorf("out: got %v", out)
+ }
+}
+
+// TestApplyBatch_LengthMismatchIgnored rejects a slice that changed
+// length: the previous accepted value carries forward.
+func TestApplyBatch_LengthMismatchIgnored(t *testing.T) {
+ bus, _ := newTestBus(t)
+ bus.RegisterBatchFilter("buggy", 10, func(ctx context.Context, items []any, args ...any) ([]any, error) {
+ // Drop the last item — a contract violation.
+ return items[:len(items)-1], nil
+ })
+
+ in := []any{"a", "b", "c"}
+ out, err := bus.ApplyBatch(context.Background(), "buggy", in)
+ if err != nil {
+ t.Fatalf("ApplyBatch: %v", err)
+ }
+ if len(out) != 3 {
+ t.Errorf("output length: got %d want 3 (mismatch should be ignored)", len(out))
+ }
+}
+
+// TestApplyBatch_ShortCircuit stops the chain with the value-so-far.
+func TestApplyBatch_ShortCircuit(t *testing.T) {
+ bus, _ := newTestBus(t)
+ bus.RegisterBatchFilter("sc", 10, func(ctx context.Context, items []any, args ...any) ([]any, error) {
+ out := make([]any, len(items))
+ for i, v := range items {
+ out[i] = "stopped-" + v.(string)
+ }
+ return out, ErrShortCircuit
+ })
+ bus.RegisterBatchFilter("sc", 20, func(ctx context.Context, items []any, args ...any) ([]any, error) {
+ t.Errorf("downstream handler ran after short-circuit")
+ return items, nil
+ })
+
+ out, err := bus.ApplyBatch(context.Background(), "sc", []any{"a", "b"})
+ if err != nil {
+ t.Fatalf("ApplyBatch: %v", err)
+ }
+ if out[0] != "stopped-a" || out[1] != "stopped-b" {
+ t.Errorf("out: got %v", out)
+ }
+}
+
+// TestApplyBatch_HandlerError stops the chain with the last accepted
+// slice and surfaces the error.
+func TestApplyBatch_HandlerError(t *testing.T) {
+ bus, _ := newTestBus(t)
+ bus.RegisterBatchFilter("err", 10, func(ctx context.Context, items []any, args ...any) ([]any, error) {
+ out := make([]any, len(items))
+ for i, v := range items {
+ out[i] = "first-" + v.(string)
+ }
+ return out, nil
+ })
+ want := errors.New("blew up")
+ bus.RegisterBatchFilter("err", 20, func(ctx context.Context, items []any, args ...any) ([]any, error) {
+ return items, want
+ })
+
+ out, err := bus.ApplyBatch(context.Background(), "err", []any{"a"})
+ if !errors.Is(err, want) {
+ t.Errorf("err: got %v want %v", err, want)
+ }
+ // last-accepted value is the first handler's output.
+ if out[0] != "first-a" {
+ t.Errorf("out: got %v want [first-a]", out)
+ }
+}
+
+// BenchmarkApplyBatch_BatchAware vs BenchmarkApplyFilters_PerItem
+// quantify the hot-path improvement issue #263 targets. The batch case
+// dispatches the chain once with the whole slice; the per-item case
+// calls ApplyFilters N times.
+
+func BenchmarkApplyFilters_PerItem(b *testing.B) {
+ bus := NewBus()
+ bus.RegisterFilter("bench", 10, func(ctx context.Context, value any, args ...any) (any, error) {
+ return value.(int) + 1, nil
+ })
+ const N = 100
+ items := make([]int, N)
+ for i := range items {
+ items[i] = i
+ }
+ ctx := context.Background()
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ for _, x := range items {
+ _, _ = bus.ApplyFilters(ctx, "bench", x)
+ }
+ }
+}
+
+func BenchmarkApplyBatch_BatchAware(b *testing.B) {
+ bus := NewBus()
+ bus.RegisterBatchFilter("bench", 10, func(ctx context.Context, in []any, args ...any) ([]any, error) {
+ out := make([]any, len(in))
+ for i, v := range in {
+ out[i] = v.(int) + 1
+ }
+ return out, nil
+ })
+ const N = 100
+ items := make([]any, N)
+ for i := range items {
+ items[i] = i
+ }
+ ctx := context.Background()
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ _, _ = bus.ApplyBatch(ctx, "bench", items)
+ }
+}
+
+func BenchmarkApplyBatch_LegacyHandler(b *testing.B) {
+ bus := NewBus()
+ bus.RegisterFilter("bench", 10, func(ctx context.Context, value any, args ...any) (any, error) {
+ return value.(int) + 1, nil
+ })
+ const N = 100
+ items := make([]any, N)
+ for i := range items {
+ items[i] = i
+ }
+ ctx := context.Background()
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ _, _ = bus.ApplyBatch(ctx, "bench", items)
+ }
+}
+
+// Self-test: errors.New + fmt.Errorf identity preserved.
+var _ = fmt.Errorf
diff --git a/packages/go/hooks/bus.go b/packages/go/hooks/bus.go
index 7e1702d0..884213b3 100644
--- a/packages/go/hooks/bus.go
+++ b/packages/go/hooks/bus.go
@@ -3,6 +3,7 @@ package hooks
import (
"context"
"errors"
+ "fmt"
"log/slog"
"sort"
"sync"
@@ -97,6 +98,19 @@ type Bus struct {
// torn reads on concurrent in-flight Apply/Do calls — the same
// pattern used for logger and metrics.
schemas atomic.Pointer[SchemaEnforcer]
+
+ // batchAdapters indexes registrations that opted into the batch
+ // filter path via RegisterBatchFilter. The map is keyed by hook
+ // name then by the registration's token so invokeBatch can spot a
+ // batch-aware handler in the middle of a chain that also contains
+ // legacy per-item filters.
+ //
+ // The mutex is taken on the hot path of ApplyBatch (one RLock per
+ // handler invocation), but only by hooks that have *any* batch
+ // registration — the nil-map fast path inside batchAdapterFor
+ // short-circuits without touching the lock.
+ batchAdaptersMu sync.RWMutex
+ batchAdapters map[string]map[uint64]*batchFilterAdapter
}
// chainSlot holds a single hook's handler chain plus the mutex that
@@ -808,6 +822,272 @@ func (b *Bus) ApplyFilters(ctx context.Context, name string, value any, args ...
return current, nil
}
+// BatchFilterHandler is the signature opted-into by plugins that prefer
+// to receive a whole []any slice in one call rather than be invoked N
+// times by ApplyFilters. It mirrors FilterHandler but pluralises the
+// `value` parameter to a slice — the handler returns the transformed
+// slice (same length, mapped element-by-element) or any error.
+//
+// The slice length is preserved on the returned value: ApplyBatch
+// validates this contract and falls back to the input slice if a
+// misbehaving handler resizes it (we log loudly so the bug surfaces
+// without dropping items from the chain).
+//
+// Plugins opt in via the manifest flag `flags.apply_filters_batch`
+// (parsed by packages/go/plugins/manifest). The hook bus carries no
+// manifest awareness itself — it just exposes RegisterBatchFilter for
+// the manifest-driven wiring layer to call.
+type BatchFilterHandler func(ctx context.Context, items []any, args ...any) ([]any, error)
+// ApplyBatch is the hot-path equivalent of ApplyFilters for callers
+// that want to thread a whole slice through a filter chain in a single
+// dispatch. It is the issue #263 optimisation: instead of N separate
+// ApplyFilters invocations for N items (each paying the per-call
+// validate + metrics + per-handler overhead), ApplyBatch dispatches
+// once with the whole slice.
+//
+// Two flavours of handler participate in the chain:
+//
+// - Batch-aware handlers (registered via RegisterBatchFilter)
+// receive items as []any and return the transformed []any. Plugins
+// opt in via the manifest flag `flags.apply_filters_batch`; the
+// wiring layer that reads the manifest is what calls
+// RegisterBatchFilter on this bus.
+//
+// - Regular filter handlers (registered via RegisterFilter) are
+// called once per item — the bus loops the legacy handler over the
+// slice. This keeps backward compatibility: existing handlers
+// continue to work unchanged inside a batched chain.
+//
+// Order of dispatch is the same priority-sorted chain ApplyFilters
+// uses; the batch and non-batch handlers interleave by priority. A
+// batch-aware handler that mutates the slice's length is rejected
+// (the input slice is preserved and a warning logged) so downstream
+// handlers always see a slice whose i-th item is the transformed
+// version of the original i-th item.
+//
+// Short-circuit / error semantics mirror ApplyFilters: returning
+// ErrShortCircuit from any handler stops the chain successfully with
+// the value-so-far; any other error stops the chain with the
+// last-accepted slice.
+func (b *Bus) ApplyBatch(ctx context.Context, name string, items []any, args ...any) ([]any, error) {
+ start := time.Now()
+ sink := b.sink()
+ sink.Counter(metricDispatchTotal, map[string]string{
+ labelKind: kindFilter,
+ labelHook: name,
+ labelBatch: "true",
+ })
+ defer func() {
+ sink.Histogram(metricDispatchDuration, time.Since(start).Seconds(),
+ map[string]string{labelKind: kindFilter, labelHook: name, labelBatch: "true"})
+ }()
+
+ // Validate every item once before entering the chain. A single bad
+ // item rejects the whole batch — that matches the contract of
+ // ApplyFilters (one bad value, one returned error) extended to the
+ // pluralised case.
+ if enf := b.schemaEnforcer(); enf != nil {
+ for i, v := range items {
+ if err := enf.Validate(name, v); err != nil {
+ sink.Counter(metricSchemaRejected, map[string]string{
+ labelKind: kindFilter,
+ labelHook: name,
+ labelBatch: "true",
+ })
+ return items, fmt.Errorf("hooks: ApplyBatch %q: item %d: %w", name, i, err)
+ }
+ }
+ }
+
+ slot, ok := b.filters.Load(name)
+ if !ok {
+ return items, nil
+ }
+ snapshot := slot.(*chainSlot).chain.Load()
+ if len(*snapshot) == 0 {
+ return items, nil
+ }
+
+ // Defensive copy: handlers operate on the slice the bus owns, and we
+ // hand the final value back to the caller. A handler mutating in place
+ // is fine; what we want to avoid is the *caller* seeing a half-mutated
+ // slice if the chain errors midway. The copy is one allocation per
+ // ApplyBatch — cheap relative to the N-call alternative this method
+ // exists to avoid.
+ current := make([]any, len(items))
+ copy(current, items)
+
+ for i, reg := range *snapshot {
+ if !reg.active.Load() {
+ continue
+ }
+ next, err := b.invokeBatch(ctx, name, i, reg, current, args)
+ if err != nil {
+ if errors.Is(err, ErrShortCircuit) {
+ sink.Counter(metricShortCircuit, map[string]string{labelHook: name})
+ return next, nil
+ }
+ return current, err
+ }
+ if len(next) != len(current) {
+ // Length-mismatch is a contract violation. Log and keep the
+ // previous slice — silently dropping or padding items would
+ // confuse downstream handlers and the caller.
+ b.log().Error("hooks: batch filter returned mismatched slice length; ignoring",
+ slog.String("hook", name),
+ slog.Int("expected", len(current)),
+ slog.Int("got", len(next)),
+ slog.Uint64("token", reg.token),
+ )
+ continue
+ }
+ current = next
+ }
+ return current, nil
+}
+
+// RegisterBatchFilter registers a batch-aware filter handler. The
+// manifest layer calls this when a plugin sets `flags.apply_filters_batch`
+// in its manifest; everyday code paths should keep using RegisterFilter.
+//
+// The returned unsubscribe closure behaves like the one RegisterFilter
+// returns: idempotent, safe to call from concurrent goroutines.
+func (b *Bus) RegisterBatchFilter(name string, priority int, handler BatchFilterHandler) func() {
+ if handler == nil {
+ return noopUnsub
+ }
+ // Wrap the batch handler as a regular FilterHandler so the registration
+ // machinery and chain semantics need no changes. invokeBatch knows
+ // how to spot the wrapper and call the batch path; legacy ApplyFilters
+ // callers see the handler as a per-item filter that applies the batch
+ // over a singleton slice.
+ wrapper := &batchFilterAdapter{handler: handler}
+ filterFn := func(ctx context.Context, value any, args ...any) (any, error) {
+ out, err := handler(ctx, []any{value}, args...)
+ if err != nil {
+ return value, err
+ }
+ if len(out) != 1 {
+ return value, fmt.Errorf("hooks: batch filter %q returned %d items for singleton input", name, len(out))
+ }
+ return out[0], nil
+ }
+ off, _ := b.register(name, kindFilterCall, false, nil, filterFn, RegisterOptions{Priority: priority})
+ // Tag the latest registration so invokeBatch can route the slice
+ // straight to the batch entry-point. The lookup is done by token
+ // inside invokeBatch.
+ b.batchAdaptersMu.Lock()
+ if b.batchAdapters == nil {
+ b.batchAdapters = make(map[string]map[uint64]*batchFilterAdapter)
+ }
+ if b.batchAdapters[name] == nil {
+ b.batchAdapters[name] = make(map[uint64]*batchFilterAdapter)
+ }
+ // The token assigned to this registration is the last one issued. We
+ // snapshot regSeq AFTER register returns: register's call to b.regSeq.Add
+ // reserved exactly one slot, so the current value of regSeq is this
+ // handler's token.
+ b.batchAdapters[name][b.regSeq.Load()] = wrapper
+ b.batchAdaptersMu.Unlock()
+
+ originalOff := off
+ return func() {
+ originalOff()
+ b.batchAdaptersMu.Lock()
+ delete(b.batchAdapters[name], b.regSeq.Load())
+ b.batchAdaptersMu.Unlock()
+ }
+}
+
+// batchFilterAdapter pairs a RegisterBatchFilter call with its
+// original handler so invokeBatch can route the slice directly into
+// the BatchFilterHandler without re-routing through the per-item
+// wrapper.
+type batchFilterAdapter struct {
+ handler BatchFilterHandler
+}
+
+// invokeBatch runs one handler against the running slice. If the
+// handler was registered via RegisterBatchFilter it dispatches the
+// batch path in one call; otherwise it loops the legacy FilterHandler
+// over each item.
+func (b *Bus) invokeBatch(ctx context.Context, name string, idx int, reg registration, items []any, args []any) (result []any, err error) {
+ start := time.Now()
+ sink := b.sink()
+ labels := map[string]string{labelKind: kindFilter, labelHook: name, labelBatch: "true"}
+ defer func() {
+ sink.Histogram(metricHandlerDuration, time.Since(start).Seconds(), labels)
+ }()
+
+ defer func() {
+ if r := recover(); r != nil {
+ pe := &panicError{hook: name, handler: idx, value: r}
+ sink.Counter(metricHandlerPanic, labels)
+ b.log().ErrorContext(ctx, "hook batch handler panicked",
+ slog.String("hook", name),
+ slog.String("kind", kindFilter),
+ slog.Any("recovered", r),
+ )
+ result = items
+ err = pe
+ }
+ }()
+
+ // Batch-aware path: route the whole slice to the BatchFilterHandler
+ // in one call.
+ if adapter := b.batchAdapterFor(name, reg.token); adapter != nil {
+ out, hErr := adapter.handler(ctx, items, args...)
+ if hErr != nil && !errors.Is(hErr, ErrShortCircuit) {
+ sink.Counter(metricHandlerError, labels)
+ }
+ return out, hErr
+ }
+
+ // Legacy path: loop the per-item filter over the slice. Yes, this
+ // is N calls — but the caller already paid the cost of ApplyBatch's
+ // per-call setup once for the whole batch (validation, metrics
+ // timing, snapshot loading), so the per-item overhead is just the
+ // handler invocation itself.
+ out := make([]any, len(items))
+ for i, v := range items {
+ next, hErr := reg.filter(ctx, v, args...)
+ if hErr != nil {
+ if errors.Is(hErr, ErrShortCircuit) {
+ // Short-circuit at item i applies to the whole batch: the
+ // item the short-circuit returned replaces the i-th slot;
+ // the rest of the items pass through unchanged (caller
+ // sees the latest accepted values).
+ out[i] = next
+ for j := i + 1; j < len(items); j++ {
+ out[j] = items[j]
+ }
+ return out, ErrShortCircuit
+ }
+ sink.Counter(metricHandlerError, labels)
+ return items, hErr
+ }
+ out[i] = next
+ }
+ return out, nil
+}
+
+// batchAdapterFor returns the registered adapter for (name, token) or
+// nil if the registration was not made via RegisterBatchFilter. The
+// fast path (no batch-aware handler ever registered) avoids the lock
+// entirely.
+func (b *Bus) batchAdapterFor(name string, token uint64) *batchFilterAdapter {
+ b.batchAdaptersMu.RLock()
+ defer b.batchAdaptersMu.RUnlock()
+ if b.batchAdapters == nil {
+ return nil
+ }
+ m := b.batchAdapters[name]
+ if m == nil {
+ return nil
+ }
+ return m[token]
+}
+
// dispatchAsync launches the handler in its own goroutine. The goroutine
// is tracked in asyncWG so tests (via Wait) can synchronize without
// time.Sleep. Errors are logged; nothing is returned to Do's caller.
diff --git a/packages/go/hooks/metrics.go b/packages/go/hooks/metrics.go
index 078bb4ba..4d1e55a8 100644
--- a/packages/go/hooks/metrics.go
+++ b/packages/go/hooks/metrics.go
@@ -64,6 +64,11 @@ const (
labelKind = "kind"
labelHook = "hook"
labelAsync = "async"
+ // labelBatch tags dispatch / handler metrics emitted via ApplyBatch
+ // (the issue #263 hot-path). Always "true" when present; absent for
+ // the per-item ApplyFilters path so a Grafana dashboard can split
+ // the two without inferring from a missing dimension.
+ labelBatch = "batch"
kindAction = "action"
kindFilter = "filter"
diff --git a/packages/go/plugins/lifecycle/manager.go b/packages/go/plugins/lifecycle/manager.go
index 72fa7e6b..8d0ec6d4 100644
--- a/packages/go/plugins/lifecycle/manager.go
+++ b/packages/go/plugins/lifecycle/manager.go
@@ -146,6 +146,13 @@ type Manager struct {
now func() time.Time
dependGate *depends.Gate
breaker *Breaker
+
+ // Versioned-update fields (issue #63). nil/zero unless the Manager
+ // was constructed with EnableVersionedUpdates. See update.go.
+ versionLog VersionLog
+ drainTracker *drainTracker
+ retainFor time.Duration
+ drainTimeout time.Duration
}
// ManagerOption configures a Manager at construction time. Functional
diff --git a/packages/go/plugins/lifecycle/update.go b/packages/go/plugins/lifecycle/update.go
new file mode 100644
index 00000000..4d6cfc70
--- /dev/null
+++ b/packages/go/plugins/lifecycle/update.go
@@ -0,0 +1,535 @@
+// update.go implements the versioned-update path documented in
+// issue #63. Where Install is the first-time admission of a bundle
+// and Activate flips it on, Update is the operator-driven swap from
+// one already-installed version to a fresh bundle: the new version is
+// staged side-by-side, the old version drains its in-flight requests,
+// the active pointer atomically flips, and the previous version is
+// retained for 24h so a rollback is one cheap call away.
+//
+// The state machine is unchanged — Update operates one level above it
+// by maintaining a per-slug version log distinct from the plugins row
+// the lifecycle Manager already owns. The plugins row continues to
+// reflect the *active* version's state; the version log tracks every
+// version we've shipped (active, retained, retired) with its drain
+// status.
+//
+// Versions in the log
+//
+// - active — currently serving traffic. Exactly one per slug.
+// - retained — previously-active, kept warm for rollback up to
+// retainFor (24h by default). Drained: no new
+// requests reach it but a rollback re-promotes it
+// without re-loading the WASM.
+// - retiring — actively draining in-flight requests immediately
+// after a swap. Once the counter reaches zero we
+// flip to retained.
+// - retired — fully drained and cleared from runtime memory.
+// Eligible for GC by the retention cron.
+
+package lifecycle
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "sort"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/Singleton-Solution/GoNext/packages/go/audit"
+ "github.com/Singleton-Solution/GoNext/packages/go/plugins/manifest"
+)
+
+// defaultRetentionFor is the window during which a previous version
+// stays warm for rollback after a successful swap. Operators can
+// override via WithRetention.
+const defaultRetentionFor = 24 * time.Hour
+
+// defaultDrainTimeout is how long a swap waits for the old version's
+// in-flight counter to hit zero before forcing the cut-over. A drain
+// that exceeds this is logged but does not block the swap — the
+// alternative (waiting forever on a wedged handler) is strictly
+// worse.
+const defaultDrainTimeout = 30 * time.Second
+
+// ErrNoRollback is returned by Rollback when no retained version is
+// available to swap back to (none ever recorded, or the retention
+// window expired and the cron purged it).
+var ErrNoRollback = errors.New("lifecycle: no retained version available for rollback")
+
+// VersionState captures the lifecycle of one row in the version log.
+type VersionState string
+
+const (
+ // VersionActive is the version currently serving requests.
+ VersionActive VersionState = "active"
+ // VersionRetiring is the previous active version still draining
+ // in-flight calls right after a swap.
+ VersionRetiring VersionState = "retiring"
+ // VersionRetained is a fully-drained version warm in memory for
+ // rollback during the retention window.
+ VersionRetained VersionState = "retained"
+ // VersionRetired is a fully-drained and unloaded version,
+ // awaiting cron-driven cleanup from the version log.
+ VersionRetired VersionState = "retired"
+)
+
+// VersionRow is one entry in the version log. Stored per slug; the
+// active row's Version matches the plugins row's Version column.
+type VersionRow struct {
+ Slug string
+ Version string
+ ABIVersion int
+ State VersionState
+ InstalledAt time.Time
+ ActivatedAt time.Time
+ RetiredAt time.Time // when State transitioned to Retiring
+ RetentionEnd time.Time // when this row becomes eligible for GC
+}
+
+// VersionLog is the storage seam for the version log. The memory and
+// postgres implementations live in lifecycle_versions_storage.go.
+// The methods are narrow on purpose — Update / Rollback / cleanup
+// touch every row through this interface so wiring a new backend is
+// "implement six methods".
+type VersionLog interface {
+ // AppendActive inserts a brand-new row in VersionActive state.
+ // If the slug already has an active row, that row is moved to
+ // VersionRetiring atomically with the insert.
+ AppendActive(ctx context.Context, row VersionRow) (previous *VersionRow, err error)
+ // MarkRetained transitions slug+version from Retiring to
+ // Retained with RetentionEnd populated. Idempotent.
+ MarkRetained(ctx context.Context, slug, version string, retentionEnd time.Time) error
+ // PromoteToActive swaps the active row to the named version.
+ // Used by Rollback. Returns ErrNoRollback if no Retained row
+ // for slug+version exists.
+ PromoteToActive(ctx context.Context, slug, version string) error
+ // MarkRetired transitions a row to Retired (unloaded).
+ MarkRetired(ctx context.Context, slug, version string) error
+ // ListRetained returns every Retained row for a slug, newest first.
+ ListRetained(ctx context.Context, slug string) ([]VersionRow, error)
+ // PurgeExpired deletes Retired rows whose RetentionEnd has
+ // passed. Returns the count. Called by the cleanup cron.
+ PurgeExpired(ctx context.Context, now time.Time) (int, error)
+}
+
+// drainTracker counts in-flight requests for a (slug, version) pair.
+// AttachDrainTracker hooks this counter into the request-routing
+// layer: every handler invocation calls Begin/End before/after
+// dispatch so Update can wait for End to drain to zero before
+// flipping the active pointer.
+type drainTracker struct {
+ mu sync.Mutex
+ counters map[drainKey]*atomic.Int64
+}
+
+type drainKey struct {
+ slug string
+ version string
+}
+
+func newDrainTracker() *drainTracker {
+ return &drainTracker{counters: make(map[drainKey]*atomic.Int64)}
+}
+
+// Begin increments the in-flight counter for (slug, version). Returns
+// a closure the caller MUST defer to decrement.
+func (d *drainTracker) Begin(slug, version string) func() {
+ key := drainKey{slug: slug, version: version}
+ d.mu.Lock()
+ c, ok := d.counters[key]
+ if !ok {
+ c = &atomic.Int64{}
+ d.counters[key] = c
+ }
+ d.mu.Unlock()
+ c.Add(1)
+ return func() { c.Add(-1) }
+}
+
+// Snapshot returns the current in-flight count for (slug, version).
+// 0 (or missing key) means fully drained.
+func (d *drainTracker) Snapshot(slug, version string) int64 {
+ d.mu.Lock()
+ c, ok := d.counters[drainKey{slug: slug, version: version}]
+ d.mu.Unlock()
+ if !ok {
+ return 0
+ }
+ return c.Load()
+}
+
+// Drop removes a (slug, version) entry from the tracker once retired.
+func (d *drainTracker) Drop(slug, version string) {
+ d.mu.Lock()
+ delete(d.counters, drainKey{slug: slug, version: version})
+ d.mu.Unlock()
+}
+
+// UpdateOption configures the version-management surface on a
+// Manager. Each option adjusts behaviour without changing the
+// Manager's constructor signature.
+type UpdateOption func(*Manager)
+
+// WithVersionLog injects the version-tracking store. Without it,
+// Update / Rollback / Cleanup return an "unsupported" error so a
+// Manager constructed by an older caller keeps the legacy behaviour
+// (no versioned updates).
+func WithVersionLog(vl VersionLog) UpdateOption {
+ return func(m *Manager) {
+ m.versionLog = vl
+ }
+}
+
+// WithRetention overrides the default 24h rollback window.
+func WithRetention(d time.Duration) UpdateOption {
+ return func(m *Manager) {
+ if d > 0 {
+ m.retainFor = d
+ }
+ }
+}
+
+// WithDrainTimeout overrides the default 30s drain ceiling.
+func WithDrainTimeout(d time.Duration) UpdateOption {
+ return func(m *Manager) {
+ if d > 0 {
+ m.drainTimeout = d
+ }
+ }
+}
+
+// EnableVersionedUpdates is the entry point used by the constructor's
+// new options. Returns a list of base ManagerOption-compatible
+// closures so callers can intermix versioned-update options with the
+// existing ManagerOption set without juggling two function types.
+func EnableVersionedUpdates(opts ...UpdateOption) ManagerOption {
+ return func(m *Manager) {
+ if m.retainFor == 0 {
+ m.retainFor = defaultRetentionFor
+ }
+ if m.drainTimeout == 0 {
+ m.drainTimeout = defaultDrainTimeout
+ }
+ if m.drainTracker == nil {
+ m.drainTracker = newDrainTracker()
+ }
+ for _, o := range opts {
+ o(m)
+ }
+ }
+}
+
+// Update installs a new version of an already-installed plugin
+// side-by-side, drains the previous version, and atomically swaps the
+// active pointer to the new one. The previous version is retained
+// for the rollback window.
+//
+// Semantics:
+//
+// 1. The slug must be Active (Update is a "roll out a new version"
+// gesture, not a re-install). Use Install for first-time
+// admission.
+// 2. The new bundle's manifest must declare the same slug.
+// 3. The new bundle's version string must be different from the
+// active version's. Re-installing the same version is a no-op
+// return — operators expect that to be safe.
+// 4. The new version is loaded into the runtime BEFORE any swap;
+// a failed load parks nothing and leaves the previous version
+// serving.
+// 5. The version log records the new row as Active and flips the
+// previous to Retiring.
+// 6. We wait up to drainTimeout for the old version's in-flight
+// counter to drain; on timeout we proceed (logging a warning)
+// because a wedged handler should not block the entire swap.
+// 7. After drain, the previous row is moved to Retained with a
+// RetentionEnd of now+retainFor.
+// 8. Periodic cleanup (RunRetentionCleanup) unloads expired
+// Retained rows and purges them from the log.
+//
+// Returns the new version string on success.
+func (m *Manager) Update(ctx context.Context, bundle io.Reader) (string, error) {
+ if m.versionLog == nil {
+ return "", errors.New("lifecycle: Update requires WithVersionLog (see EnableVersionedUpdates)")
+ }
+ if bundle == nil {
+ return "", errors.New("lifecycle: Update: bundle reader is required")
+ }
+
+ // Reuse readManifestFromBundle to pull the slug + version + ABI.
+ parsed, err := readManifestFromBundle(bundle)
+ if err != nil {
+ return "", fmt.Errorf("lifecycle: Update: %w", err)
+ }
+ if declaresAPIVersion(parsed.Raw) {
+ if _, vErr := manifest.Validate(parsed.Raw); vErr != nil {
+ return "", fmt.Errorf("lifecycle: Update: %w", vErr)
+ }
+ }
+ if !slugRegex.MatchString(parsed.Slug) {
+ return "", fmt.Errorf("lifecycle: Update: invalid slug %q", parsed.Slug)
+ }
+ if parsed.Version == "" {
+ return "", errors.New("lifecycle: Update: manifest version is required")
+ }
+
+ current, err := m.storage.Get(ctx, parsed.Slug)
+ if err != nil {
+ return "", fmt.Errorf("lifecycle: Update: %w", err)
+ }
+ if current.State != StateActive {
+ return "", fmt.Errorf("lifecycle: Update %q: plugin must be Active (got %q)", parsed.Slug, current.State)
+ }
+ if current.Version == parsed.Version {
+ // Idempotent no-op — operator re-uploaded the same bundle.
+ m.audit(ctx, parsed.Slug, "plugin.update.noop", audit.SeverityInfo, map[string]any{
+ "version": parsed.Version,
+ })
+ return parsed.Version, nil
+ }
+
+ // Stage the new version. Runtime.Load is called with a synthetic
+ // Plugin whose Version is the NEW one — the runtime is expected
+ // to key its instance map on (slug, version) so the previous
+ // version's module stays alive in parallel.
+ staged := current
+ staged.Version = parsed.Version
+ staged.ABIVersion = parsed.ABIVersion
+ staged.Manifest = parsed.Raw
+ if err := m.runtime.Load(ctx, staged); err != nil {
+ // Old version is untouched; surface and return.
+ return "", fmt.Errorf("lifecycle: Update %q: stage new version: %w", parsed.Slug, err)
+ }
+
+ // Commit the version log: record the new row as active and flip
+ // the previous one to Retiring. This is the single source of
+ // truth for the swap; storage.UpdateState on the plugins row is
+ // the user-visible mirror we apply next.
+ now := m.now().UTC()
+ newRow := VersionRow{
+ Slug: parsed.Slug,
+ Version: parsed.Version,
+ ABIVersion: parsed.ABIVersion,
+ State: VersionActive,
+ InstalledAt: now,
+ ActivatedAt: now,
+ }
+ previous, err := m.versionLog.AppendActive(ctx, newRow)
+ if err != nil {
+ // Roll back the runtime load: we own the new module but the
+ // log refuses our claim.
+ if uErr := m.runtime.Unload(ctx, parsed.Slug); uErr != nil {
+ m.logger.Warn("lifecycle: Update: failed to unload staged new version after log error",
+ slog.String("slug", parsed.Slug),
+ slog.String("version", parsed.Version),
+ slog.String("err", uErr.Error()),
+ )
+ }
+ return "", fmt.Errorf("lifecycle: Update %q: append version: %w", parsed.Slug, err)
+ }
+
+ // Mirror the new version on the plugins row. Active → Active is
+ // not a state transition the regular CAS handles, so we go via
+ // the dedicated UpdateActiveVersion helper (a Storage extension —
+ // see versions_storage.go) when available; otherwise fall back to
+ // rewriting through a deactivate/activate cycle is too disruptive,
+ // so we just keep the plugins row in sync via Storage.
+ if err := m.applyActiveVersion(ctx, parsed.Slug, current.Version, parsed.Version, parsed.Raw, parsed.ABIVersion); err != nil {
+ m.logger.Error("lifecycle: Update: failed to write plugins row",
+ slog.String("slug", parsed.Slug),
+ slog.String("err", err.Error()),
+ )
+ // We don't roll back here — the version log is authoritative;
+ // the plugins row will reconverge on next List/Get when
+ // callers see the version mismatch. The operator gets a
+ // warning, not a hard failure.
+ }
+
+ // Drain the previous version. We poll the in-flight counter
+ // because the alternative (a channel handed to every handler)
+ // requires wiring on every dispatcher; the poll is cheap and
+ // gives a deterministic deadline.
+ drained := true
+ if previous != nil && m.drainTracker != nil {
+ drained = m.waitDrain(ctx, parsed.Slug, previous.Version)
+ }
+
+ // Move the previous version to Retained.
+ retentionEnd := m.now().UTC().Add(m.retainFor)
+ if previous != nil {
+ if err := m.versionLog.MarkRetained(ctx, parsed.Slug, previous.Version, retentionEnd); err != nil {
+ m.logger.Warn("lifecycle: Update: failed to mark previous version retained",
+ slog.String("slug", parsed.Slug),
+ slog.String("version", previous.Version),
+ slog.String("err", err.Error()),
+ )
+ }
+ }
+
+ m.audit(ctx, parsed.Slug, "plugin.updated", audit.SeverityInfo, map[string]any{
+ "from_version": current.Version,
+ "to_version": parsed.Version,
+ "abi_version": parsed.ABIVersion,
+ "drained_clean": drained,
+ "retention_end": retentionEnd.Format(time.RFC3339),
+ })
+ return parsed.Version, nil
+}
+
+// Rollback re-promotes the most recent Retained version (or the
+// caller-specified version) back to active, after draining the
+// currently-active row. The drained-then-retained cycle runs again so
+// repeated rollbacks remain reversible.
+//
+// Returns ErrNoRollback if no Retained row is available.
+func (m *Manager) Rollback(ctx context.Context, slug, toVersion string) (string, error) {
+ if m.versionLog == nil {
+ return "", errors.New("lifecycle: Rollback requires WithVersionLog")
+ }
+ current, err := m.storage.Get(ctx, slug)
+ if err != nil {
+ return "", err
+ }
+ retained, err := m.versionLog.ListRetained(ctx, slug)
+ if err != nil {
+ return "", fmt.Errorf("lifecycle: Rollback %q: %w", slug, err)
+ }
+ if len(retained) == 0 {
+ return "", ErrNoRollback
+ }
+ if toVersion == "" {
+ toVersion = retained[0].Version
+ }
+ // Validate the requested version is in the retained set.
+ found := false
+ var target VersionRow
+ for _, r := range retained {
+ if r.Version == toVersion {
+ target = r
+ found = true
+ break
+ }
+ }
+ if !found {
+ return "", fmt.Errorf("%w: version %q", ErrNoRollback, toVersion)
+ }
+ if target.Version == current.Version {
+ return current.Version, nil // no-op
+ }
+
+ if err := m.versionLog.PromoteToActive(ctx, slug, toVersion); err != nil {
+ return "", fmt.Errorf("lifecycle: Rollback %q: promote: %w", slug, err)
+ }
+ if err := m.applyActiveVersion(ctx, slug, current.Version, toVersion, target.toManifestBytes(), target.ABIVersion); err != nil {
+ m.logger.Warn("lifecycle: Rollback: failed to write plugins row",
+ slog.String("slug", slug), slog.String("err", err.Error()))
+ }
+ if m.drainTracker != nil {
+ m.waitDrain(ctx, slug, current.Version)
+ }
+ retentionEnd := m.now().UTC().Add(m.retainFor)
+ if err := m.versionLog.MarkRetained(ctx, slug, current.Version, retentionEnd); err != nil {
+ m.logger.Warn("lifecycle: Rollback: failed to retain old active",
+ slog.String("slug", slug), slog.String("err", err.Error()))
+ }
+ m.audit(ctx, slug, "plugin.rollback", audit.SeverityWarning, map[string]any{
+ "from_version": current.Version,
+ "to_version": toVersion,
+ })
+ return toVersion, nil
+}
+
+// RunRetentionCleanup is the cron entrypoint: walks the version log,
+// unloads versions whose RetentionEnd has passed, and purges the
+// rows. Idempotent and safe to call from a leader-only scheduler.
+//
+// Returns the count of versions cleaned up.
+func (m *Manager) RunRetentionCleanup(ctx context.Context) (int, error) {
+ if m.versionLog == nil {
+ return 0, errors.New("lifecycle: RunRetentionCleanup requires WithVersionLog")
+ }
+ purged, err := m.versionLog.PurgeExpired(ctx, m.now().UTC())
+ if err != nil {
+ return 0, err
+ }
+ return purged, nil
+}
+
+// DrainTracker exposes the in-flight counter so the request-routing
+// layer can wire Begin/End around every dispatch. Returns nil when
+// versioned updates aren't enabled, which signals to the dispatcher
+// that no version tracking is required.
+func (m *Manager) DrainTracker() *drainTracker {
+ return m.drainTracker
+}
+
+// waitDrain polls the drain counter until it reaches zero or the
+// timeout fires. Returns true when drained clean, false on timeout.
+func (m *Manager) waitDrain(ctx context.Context, slug, version string) bool {
+ if m.drainTracker == nil {
+ return true
+ }
+ deadline := time.Now().Add(m.drainTimeout)
+ for time.Now().Before(deadline) {
+ if m.drainTracker.Snapshot(slug, version) == 0 {
+ return true
+ }
+ select {
+ case <-ctx.Done():
+ return false
+ case <-time.After(50 * time.Millisecond):
+ }
+ }
+ m.logger.Warn("lifecycle: Update: drain timeout — proceeding with swap",
+ slog.String("slug", slug),
+ slog.String("version", version),
+ slog.Int64("in_flight", m.drainTracker.Snapshot(slug, version)),
+ )
+ return false
+}
+
+// applyActiveVersion writes the new version onto the plugins row.
+// Active → Active isn't an UpdateState transition, so we touch the
+// row directly through Storage.Insert/Delete is wrong — we want a
+// single update. The current Storage interface doesn't expose an
+// arbitrary update, so we model this as a no-op write through the
+// versions extension (see VersionedStorage).
+func (m *Manager) applyActiveVersion(ctx context.Context, slug, fromVersion, toVersion string, manifestBytes []byte, abiVersion int) error {
+ if vs, ok := m.storage.(VersionedStorage); ok {
+ return vs.UpdateActiveVersion(ctx, slug, toVersion, manifestBytes, abiVersion)
+ }
+ // Fallback: log and continue. The version log carries the
+ // authoritative version; the plugins row will look stale on Get
+ // until a backend supports the update.
+ return nil
+}
+
+// toManifestBytes is a placeholder — the retained row doesn't carry
+// the manifest bytes today (the runtime keeps them in its instance
+// map). When Rollback fires we hand an empty manifest to
+// applyActiveVersion; the storage layer keeps the current row's
+// manifest untouched, which is correct because the retained version
+// is the one that originally produced it.
+func (r VersionRow) toManifestBytes() []byte {
+ return nil
+}
+
+// VersionedStorage is the optional extension implemented by Storage
+// backends that can write the active version onto an Active row
+// without going through the CAS path. The memory + postgres backends
+// implement it; older callers can opt out by sticking to the basic
+// Storage interface (Update is a no-op against the plugins row in
+// that case).
+type VersionedStorage interface {
+ UpdateActiveVersion(ctx context.Context, slug, version string, manifestBytes []byte, abiVersion int) error
+}
+
+// sortByInstalledAtDesc orders rows newest first.
+func sortByInstalledAtDesc(rows []VersionRow) {
+ sort.Slice(rows, func(i, j int) bool {
+ return rows[i].InstalledAt.After(rows[j].InstalledAt)
+ })
+}
diff --git a/packages/go/plugins/lifecycle/update_test.go b/packages/go/plugins/lifecycle/update_test.go
new file mode 100644
index 00000000..1f3f34ae
--- /dev/null
+++ b/packages/go/plugins/lifecycle/update_test.go
@@ -0,0 +1,292 @@
+package lifecycle
+
+import (
+ "context"
+ "errors"
+ "io"
+ "testing"
+ "time"
+)
+
+// makeBundle builds a minimal .gnplugin ZIP with a manifest carrying
+// the requested slug/version/abi. Wraps the existing buildBundle
+// helper in manager_test.go so update-specific tests can build many
+// bundles concisely.
+func makeBundle(t *testing.T, slug, version string, abi int) io.Reader {
+ t.Helper()
+ manifest := `{"slug":"` + slug + `","version":"` + version + `","abi_version":` + itoaUT(abi) + `}`
+ return buildBundle(t, manifest)
+}
+
+func itoaUT(n int) string {
+ if n == 0 {
+ return "0"
+ }
+ var digits []byte
+ for n > 0 {
+ digits = append([]byte{byte('0' + n%10)}, digits...)
+ n /= 10
+ }
+ return string(digits)
+}
+
+// TestUpdate_AppendsActiveAndRetainsOld verifies the happy path:
+// install + activate v1.0.0, then Update to v1.1.0. The version log
+// has a retained 1.0.0 row at the end.
+func TestUpdate_AppendsActiveAndRetainsOld(t *testing.T) {
+ vl := NewMemoryVersionLog()
+ rt := &recordingRuntime{}
+ m, store, _ := newManagerForTest(t,
+ WithRuntime(rt),
+ EnableVersionedUpdates(WithVersionLog(vl), WithRetention(2*time.Hour)),
+ )
+
+ ctx := context.Background()
+ if _, err := m.Install(ctx, makeBundle(t, "test-plugin", "1.0.0", 1)); err != nil {
+ t.Fatalf("Install: %v", err)
+ }
+ if err := m.Activate(ctx, "test-plugin"); err != nil {
+ t.Fatalf("Activate: %v", err)
+ }
+ // Seed the version log with the initial active row (Install
+ // doesn't touch the version log — Update is the first interaction).
+ _, err := vl.AppendActive(ctx, VersionRow{
+ Slug: "test-plugin", Version: "1.0.0", ABIVersion: 1,
+ InstalledAt: time.Now().UTC(),
+ })
+ if err != nil {
+ t.Fatalf("seed: %v", err)
+ }
+
+ v, err := m.Update(ctx, makeBundle(t, "test-plugin", "1.1.0", 1))
+ if err != nil {
+ t.Fatalf("Update: %v", err)
+ }
+ if v != "1.1.0" {
+ t.Errorf("returned version: %q", v)
+ }
+
+ // plugins row should reflect 1.1.0.
+ got, err := store.Get(ctx, "test-plugin")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got.Version != "1.1.0" {
+ t.Errorf("plugins.Version: got %q want 1.1.0", got.Version)
+ }
+
+ // 1.0.0 should now be retained.
+ retained, _ := vl.ListRetained(ctx, "test-plugin")
+ if len(retained) != 1 || retained[0].Version != "1.0.0" {
+ t.Errorf("retained: %v", retained)
+ }
+}
+
+// TestUpdate_RejectsSameVersion treats same-version as a no-op.
+func TestUpdate_RejectsSameVersion(t *testing.T) {
+ vl := NewMemoryVersionLog()
+ rt := &recordingRuntime{}
+ m, _, _ := newManagerForTest(t,
+ WithRuntime(rt),
+ EnableVersionedUpdates(WithVersionLog(vl)),
+ )
+ ctx := context.Background()
+ if _, err := m.Install(ctx, makeBundle(t, "plug-x", "2.0.0", 1)); err != nil {
+ t.Fatal(err)
+ }
+ if err := m.Activate(ctx, "plug-x"); err != nil {
+ t.Fatal(err)
+ }
+ _, _ = vl.AppendActive(ctx, VersionRow{Slug: "plug-x", Version: "2.0.0", ABIVersion: 1, InstalledAt: time.Now()})
+
+ v, err := m.Update(ctx, makeBundle(t, "plug-x", "2.0.0", 1))
+ if err != nil {
+ t.Fatalf("Update same version: %v", err)
+ }
+ if v != "2.0.0" {
+ t.Errorf("version: %q", v)
+ }
+ // No new row in the log.
+ rows := vl.List("plug-x")
+ if len(rows) != 1 {
+ t.Errorf("rows: %d (no-op should not append)", len(rows))
+ }
+}
+
+// TestUpdate_LoadFailureLeavesOldVersionRunning verifies that a
+// failed Runtime.Load on the new version doesn't disturb the active
+// plugin row.
+func TestUpdate_LoadFailureLeavesOldVersionRunning(t *testing.T) {
+ vl := NewMemoryVersionLog()
+ rt := &recordingRuntime{}
+ m, store, _ := newManagerForTest(t,
+ WithRuntime(rt),
+ EnableVersionedUpdates(WithVersionLog(vl)),
+ )
+ ctx := context.Background()
+ if _, err := m.Install(ctx, makeBundle(t, "plug-a", "1.0.0", 1)); err != nil {
+ t.Fatal(err)
+ }
+ if err := m.Activate(ctx, "plug-a"); err != nil {
+ t.Fatal(err)
+ }
+ _, _ = vl.AppendActive(ctx, VersionRow{Slug: "plug-a", Version: "1.0.0", ABIVersion: 1, InstalledAt: time.Now()})
+
+ rt.loadErr = errors.New("wasm explode")
+ _, err := m.Update(ctx, makeBundle(t, "plug-a", "1.1.0", 1))
+ if err == nil {
+ t.Fatal("expected error")
+ }
+ got, _ := store.Get(ctx, "plug-a")
+ if got.Version != "1.0.0" {
+ t.Errorf("plugin row mutated: %q", got.Version)
+ }
+}
+
+// TestRollback_RestoresRetainedVersion installs 1.0.0, updates to 1.1.0,
+// then rolls back. After rollback the plugins row reports 1.0.0.
+func TestRollback_RestoresRetainedVersion(t *testing.T) {
+ vl := NewMemoryVersionLog()
+ m, store, _ := newManagerForTest(t,
+ WithRuntime(&recordingRuntime{}),
+ EnableVersionedUpdates(WithVersionLog(vl), WithRetention(2*time.Hour)),
+ )
+ ctx := context.Background()
+ if _, err := m.Install(ctx, makeBundle(t, "plug-rb", "1.0.0", 1)); err != nil {
+ t.Fatal(err)
+ }
+ if err := m.Activate(ctx, "plug-rb"); err != nil {
+ t.Fatal(err)
+ }
+ _, _ = vl.AppendActive(ctx, VersionRow{Slug: "plug-rb", Version: "1.0.0", ABIVersion: 1, InstalledAt: time.Now()})
+
+ if _, err := m.Update(ctx, makeBundle(t, "plug-rb", "1.1.0", 1)); err != nil {
+ t.Fatalf("Update: %v", err)
+ }
+
+ v, err := m.Rollback(ctx, "plug-rb", "")
+ if err != nil {
+ t.Fatalf("Rollback: %v", err)
+ }
+ if v != "1.0.0" {
+ t.Errorf("rolled back to %q", v)
+ }
+ got, _ := store.Get(ctx, "plug-rb")
+ if got.Version != "1.0.0" {
+ t.Errorf("plugins.Version: %q", got.Version)
+ }
+}
+
+// TestRollback_NoRetainedReturnsErr returns ErrNoRollback when no
+// previous version is retained.
+func TestRollback_NoRetainedReturnsErr(t *testing.T) {
+ vl := NewMemoryVersionLog()
+ m, _, _ := newManagerForTest(t,
+ WithRuntime(&recordingRuntime{}),
+ EnableVersionedUpdates(WithVersionLog(vl)),
+ )
+ ctx := context.Background()
+ if _, err := m.Install(ctx, makeBundle(t, "fresh", "1.0.0", 1)); err != nil {
+ t.Fatal(err)
+ }
+ if err := m.Activate(ctx, "fresh"); err != nil {
+ t.Fatal(err)
+ }
+ _, err := m.Rollback(ctx, "fresh", "")
+ if !errors.Is(err, ErrNoRollback) {
+ t.Errorf("err: %v", err)
+ }
+}
+
+// TestRetentionCleanup_PurgesExpired verifies the cron-style helper.
+func TestRetentionCleanup_PurgesExpired(t *testing.T) {
+ vl := NewMemoryVersionLog()
+ now := fixedTime
+ m, _, _ := newManagerForTest(t,
+ WithRuntime(&recordingRuntime{}),
+ WithNowFunc(func() time.Time { return now }),
+ EnableVersionedUpdates(WithVersionLog(vl), WithRetention(time.Minute)),
+ )
+ ctx := context.Background()
+ if _, err := m.Install(ctx, makeBundle(t, "plug-rt", "1.0.0", 1)); err != nil {
+ t.Fatal(err)
+ }
+ if err := m.Activate(ctx, "plug-rt"); err != nil {
+ t.Fatal(err)
+ }
+ _, _ = vl.AppendActive(ctx, VersionRow{Slug: "plug-rt", Version: "1.0.0", ABIVersion: 1, InstalledAt: now})
+ if _, err := m.Update(ctx, makeBundle(t, "plug-rt", "1.1.0", 1)); err != nil {
+ t.Fatal(err)
+ }
+ // Advance the clock past the retention window.
+ now = now.Add(2 * time.Minute)
+ purged, err := m.RunRetentionCleanup(ctx)
+ if err != nil {
+ t.Fatalf("Cleanup: %v", err)
+ }
+ if purged != 1 {
+ t.Errorf("purged: %d", purged)
+ }
+}
+
+// TestDrainTracker_BeginEnd verifies the in-flight counter.
+func TestDrainTracker_BeginEnd(t *testing.T) {
+ d := newDrainTracker()
+ end1 := d.Begin("s", "v")
+ end2 := d.Begin("s", "v")
+ if got := d.Snapshot("s", "v"); got != 2 {
+ t.Errorf("snapshot: %d", got)
+ }
+ end1()
+ if got := d.Snapshot("s", "v"); got != 1 {
+ t.Errorf("snapshot after one End: %d", got)
+ }
+ end2()
+ if got := d.Snapshot("s", "v"); got != 0 {
+ t.Errorf("snapshot drained: %d", got)
+ }
+}
+
+// TestUpdate_DrainObserved verifies Update waits for in-flight calls
+// against the old version before marking it retained.
+func TestUpdate_DrainObserved(t *testing.T) {
+ vl := NewMemoryVersionLog()
+ m, _, _ := newManagerForTest(t,
+ WithRuntime(&recordingRuntime{}),
+ EnableVersionedUpdates(
+ WithVersionLog(vl),
+ WithDrainTimeout(200*time.Millisecond),
+ ),
+ )
+ ctx := context.Background()
+ if _, err := m.Install(ctx, makeBundle(t, "plug-dr", "1.0.0", 1)); err != nil {
+ t.Fatal(err)
+ }
+ if err := m.Activate(ctx, "plug-dr"); err != nil {
+ t.Fatal(err)
+ }
+ _, _ = vl.AppendActive(ctx, VersionRow{Slug: "plug-dr", Version: "1.0.0", ABIVersion: 1, InstalledAt: time.Now()})
+
+ // Start an in-flight request that finishes shortly.
+ dt := m.DrainTracker()
+ end := dt.Begin("plug-dr", "1.0.0")
+ go func() {
+ time.Sleep(50 * time.Millisecond)
+ end()
+ }()
+
+ start := time.Now()
+ if _, err := m.Update(ctx, makeBundle(t, "plug-dr", "1.1.0", 1)); err != nil {
+ t.Fatalf("Update: %v", err)
+ }
+ elapsed := time.Since(start)
+ if elapsed < 40*time.Millisecond {
+ t.Errorf("Update returned before drain: %v", elapsed)
+ }
+ if elapsed > 200*time.Millisecond {
+ t.Errorf("Update exceeded drain timeout: %v", elapsed)
+ }
+ if dt.Snapshot("plug-dr", "1.0.0") != 0 {
+ t.Errorf("drain not zero")
+ }
+}
diff --git a/packages/go/plugins/lifecycle/versions_memory.go b/packages/go/plugins/lifecycle/versions_memory.go
new file mode 100644
index 00000000..972d505b
--- /dev/null
+++ b/packages/go/plugins/lifecycle/versions_memory.go
@@ -0,0 +1,206 @@
+package lifecycle
+
+import (
+ "context"
+ "fmt"
+ "sort"
+ "sync"
+ "time"
+)
+
+// MemoryVersionLog is the in-process VersionLog used by tests and
+// dev setups. It is paired with MemoryStorage; together they exercise
+// the same paths the production postgres backend does without a live
+// database.
+//
+// Concurrency: every mutating method takes a single Mutex. Reads also
+// take it because the slice mutations involve multiple steps that
+// must appear atomic to a concurrent List call. Contention is fine at
+// test scale.
+type MemoryVersionLog struct {
+ mu sync.Mutex
+ rows map[string][]VersionRow // slug -> ordered (by InstalledAt) rows
+}
+
+// NewMemoryVersionLog returns an empty in-memory version log.
+func NewMemoryVersionLog() *MemoryVersionLog {
+ return &MemoryVersionLog{rows: make(map[string][]VersionRow)}
+}
+
+// AppendActive inserts row as Active. Any existing Active row for the
+// slug is moved to Retiring atomically with the insert so a concurrent
+// reader sees exactly one Active per slug.
+func (m *MemoryVersionLog) AppendActive(_ context.Context, row VersionRow) (*VersionRow, error) {
+ if row.Slug == "" || row.Version == "" {
+ return nil, fmt.Errorf("lifecycle/memory-versions: AppendActive: slug and version are required")
+ }
+ row.State = VersionActive
+ m.mu.Lock()
+ defer m.mu.Unlock()
+
+ rows := m.rows[row.Slug]
+ var previous *VersionRow
+ for i := range rows {
+ if rows[i].State == VersionActive {
+ rows[i].State = VersionRetiring
+ rows[i].RetiredAt = row.InstalledAt
+ cp := rows[i]
+ previous = &cp
+ }
+ if rows[i].Version == row.Version {
+ return nil, fmt.Errorf("lifecycle/memory-versions: version %q already exists for %q", row.Version, row.Slug)
+ }
+ }
+ rows = append(rows, row)
+ m.rows[row.Slug] = rows
+ return previous, nil
+}
+
+// MarkRetained transitions slug+version from Retiring (or Active,
+// for symmetry with rollback flows) to Retained.
+func (m *MemoryVersionLog) MarkRetained(_ context.Context, slug, version string, retentionEnd time.Time) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ rows := m.rows[slug]
+ for i := range rows {
+ if rows[i].Version == version {
+ rows[i].State = VersionRetained
+ rows[i].RetentionEnd = retentionEnd
+ return nil
+ }
+ }
+ return fmt.Errorf("lifecycle/memory-versions: MarkRetained %q/%q: not found", slug, version)
+}
+
+// PromoteToActive swaps the existing active row to Retiring and
+// flips the named retained row to Active.
+func (m *MemoryVersionLog) PromoteToActive(_ context.Context, slug, version string) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ rows := m.rows[slug]
+ if len(rows) == 0 {
+ return fmt.Errorf("%w: slug %q", ErrNoRollback, slug)
+ }
+ foundTarget := false
+ for i := range rows {
+ if rows[i].Version == version && rows[i].State == VersionRetained {
+ foundTarget = true
+ break
+ }
+ }
+ if !foundTarget {
+ return fmt.Errorf("%w: version %q", ErrNoRollback, version)
+ }
+ now := time.Now().UTC()
+ for i := range rows {
+ switch {
+ case rows[i].Version == version:
+ rows[i].State = VersionActive
+ rows[i].ActivatedAt = now
+ case rows[i].State == VersionActive:
+ rows[i].State = VersionRetiring
+ rows[i].RetiredAt = now
+ }
+ }
+ m.rows[slug] = rows
+ return nil
+}
+
+// MarkRetired moves a row to Retired (fully unloaded). Used by the
+// cleanup cron after the runtime drops the WASM module.
+func (m *MemoryVersionLog) MarkRetired(_ context.Context, slug, version string) error {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ rows := m.rows[slug]
+ for i := range rows {
+ if rows[i].Version == version {
+ rows[i].State = VersionRetired
+ return nil
+ }
+ }
+ return fmt.Errorf("lifecycle/memory-versions: MarkRetired %q/%q: not found", slug, version)
+}
+
+// ListRetained returns Retained rows sorted newest first.
+func (m *MemoryVersionLog) ListRetained(_ context.Context, slug string) ([]VersionRow, error) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ var out []VersionRow
+ for _, r := range m.rows[slug] {
+ if r.State == VersionRetained {
+ out = append(out, r)
+ }
+ }
+ sort.Slice(out, func(i, j int) bool {
+ return out[i].InstalledAt.After(out[j].InstalledAt)
+ })
+ return out, nil
+}
+
+// PurgeExpired drops Retained rows whose RetentionEnd has passed,
+// then any Retired rows. Returns the count purged.
+func (m *MemoryVersionLog) PurgeExpired(_ context.Context, now time.Time) (int, error) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ purged := 0
+ for slug, rows := range m.rows {
+ kept := rows[:0]
+ for _, r := range rows {
+ if r.State == VersionRetained && !r.RetentionEnd.IsZero() && now.After(r.RetentionEnd) {
+ purged++
+ continue
+ }
+ if r.State == VersionRetired {
+ purged++
+ continue
+ }
+ kept = append(kept, r)
+ }
+ if len(kept) == 0 {
+ delete(m.rows, slug)
+ } else {
+ m.rows[slug] = kept
+ }
+ }
+ return purged, nil
+}
+
+// List exposes the full slice for tests / debug. Not part of the
+// VersionLog interface so callers can't depend on it.
+func (m *MemoryVersionLog) List(slug string) []VersionRow {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ out := make([]VersionRow, len(m.rows[slug]))
+ copy(out, m.rows[slug])
+ return out
+}
+
+// UpdateActiveVersion is the VersionedStorage extension on
+// MemoryStorage. It rewrites the version/manifest/ABI on an Active
+// row in place — used by Update when the storage backend supports
+// active-to-active writes.
+func (s *MemoryStorage) UpdateActiveVersion(_ context.Context, slug, version string, manifestBytes []byte, abiVersion int) error {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ p, ok := s.rows[slug]
+ if !ok {
+ return fmt.Errorf("%w: %q", ErrNotFound, slug)
+ }
+ p.Version = version
+ if abiVersion > 0 {
+ p.ABIVersion = abiVersion
+ }
+ if len(manifestBytes) > 0 {
+ cp := make([]byte, len(manifestBytes))
+ copy(cp, manifestBytes)
+ p.Manifest = cp
+ }
+ p.UpdatedAt = s.now().UTC()
+ p.RowVersion++
+ s.rows[slug] = p
+ return nil
+}
+
+// Compile-time check.
+var _ VersionLog = (*MemoryVersionLog)(nil)
+var _ VersionedStorage = (*MemoryStorage)(nil)
diff --git a/packages/go/plugins/lifecycle/versions_postgres.go b/packages/go/plugins/lifecycle/versions_postgres.go
new file mode 100644
index 00000000..5d693321
--- /dev/null
+++ b/packages/go/plugins/lifecycle/versions_postgres.go
@@ -0,0 +1,333 @@
+package lifecycle
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "time"
+
+ "github.com/jackc/pgx/v5"
+ "github.com/jackc/pgx/v5/pgconn"
+)
+
+// PostgresVersionLog persists the version log against the
+// plugin_version_log table (migration 000039). The methods mirror the
+// VersionLog interface 1:1; transactions span multi-row mutations
+// (AppendActive flips the previous active row + inserts the new one)
+// so a concurrent caller cannot observe two active rows for the same
+// slug.
+type PostgresVersionLog struct {
+ db PgxQuerier
+}
+
+// NewPostgresVersionLog wraps a pgx pool. Reuses PgxQuerier from the
+// plugins-row store so test code can inject the same fake.
+func NewPostgresVersionLog(db PgxQuerier) *PostgresVersionLog {
+ if db == nil {
+ panic("lifecycle.NewPostgresVersionLog: db is required")
+ }
+ return &PostgresVersionLog{db: db}
+}
+
+// AppendActive flips any current active row to retiring and inserts
+// the new row as active. The two writes share a transaction so the
+// "single active row per slug" invariant is preserved across
+// concurrent callers.
+//
+// Because we share the PgxQuerier interface with PostgresStorage,
+// we accept either a pool (which auto-creates a connection-scoped
+// transaction inside Exec) or an explicit tx the caller threaded
+// through. In the pool case we open our own BEGIN/COMMIT pair.
+func (s *PostgresVersionLog) AppendActive(ctx context.Context, row VersionRow) (*VersionRow, error) {
+ if row.Slug == "" || row.Version == "" {
+ return nil, errors.New("lifecycle/postgres-versions: AppendActive: slug and version required")
+ }
+
+ // Read the current active row (if any) for the slug. We do this
+ // inside an explicit transaction so the flip-then-insert pair is
+ // atomic.
+ pool, ok := s.db.(beginTxer)
+ if !ok {
+ // Tests pass a pre-existing transaction directly — bypass the
+ // pool dance and execute the two statements in-place.
+ return s.appendActiveWithin(ctx, s.db, row)
+ }
+ tx, err := pool.Begin(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("lifecycle/postgres-versions: begin: %w", err)
+ }
+ defer func() { _ = tx.Rollback(ctx) }()
+ prev, err := s.appendActiveWithin(ctx, txQuerier{tx: tx}, row)
+ if err != nil {
+ return nil, err
+ }
+ if err := tx.Commit(ctx); err != nil {
+ return nil, fmt.Errorf("lifecycle/postgres-versions: commit: %w", err)
+ }
+ return prev, nil
+}
+
+// appendActiveWithin runs the flip-then-insert pair on the supplied
+// querier (pool or tx). Exported only inside the package.
+func (s *PostgresVersionLog) appendActiveWithin(ctx context.Context, q PgxQuerier, row VersionRow) (*VersionRow, error) {
+ // Lookup the current active row first so we can return it to the
+ // Manager (which uses it to know what to drain).
+ var prev VersionRow
+ hasPrev := false
+ {
+ r := q.QueryRow(ctx, `
+ SELECT slug, version, abi_version, installed_at,
+ COALESCE(activated_at, 'epoch'::TIMESTAMPTZ)
+ FROM plugin_version_log
+ WHERE slug = $1 AND state = 'active'`, row.Slug)
+ var activatedAt time.Time
+ err := r.Scan(&prev.Slug, &prev.Version, &prev.ABIVersion, &prev.InstalledAt, &activatedAt)
+ switch {
+ case errors.Is(err, pgx.ErrNoRows):
+ // No prior active row — that's fine, first time we record
+ // this slug. (The lifecycle Manager only calls AppendActive
+ // from Update, which requires the slug to be Active in the
+ // plugins table, but the bootstrap flow could also use it
+ // to record an initial version. We don't enforce the
+ // "must have a previous" rule here.)
+ case err != nil:
+ return nil, fmt.Errorf("lookup previous active: %w", err)
+ default:
+ prev.State = VersionRetiring
+ if !isEpoch(activatedAt) {
+ prev.ActivatedAt = activatedAt
+ }
+ hasPrev = true
+ }
+ }
+
+ if hasPrev {
+ _, err := q.Exec(ctx, `
+ UPDATE plugin_version_log
+ SET state = 'retiring',
+ retired_at = $1
+ WHERE slug = $2 AND version = $3 AND state = 'active'`,
+ row.InstalledAt, row.Slug, prev.Version)
+ if err != nil {
+ return nil, fmt.Errorf("flip previous active: %w", err)
+ }
+ }
+
+ if row.ActivatedAt.IsZero() {
+ row.ActivatedAt = row.InstalledAt
+ }
+ _, err := q.Exec(ctx, `
+ INSERT INTO plugin_version_log (slug, version, abi_version, state, installed_at, activated_at)
+ VALUES ($1, $2, $3, 'active', $4, $5)`,
+ row.Slug, row.Version, row.ABIVersion, row.InstalledAt, row.ActivatedAt)
+ if err != nil {
+ return nil, fmt.Errorf("insert new active: %w", err)
+ }
+
+ if hasPrev {
+ return &prev, nil
+ }
+ return nil, nil
+}
+
+// MarkRetained transitions a row to retained with the given retention_end.
+func (s *PostgresVersionLog) MarkRetained(ctx context.Context, slug, version string, retentionEnd time.Time) error {
+ tag, err := s.db.Exec(ctx, `
+ UPDATE plugin_version_log
+ SET state = 'retained',
+ retention_end = $1
+ WHERE slug = $2 AND version = $3`,
+ retentionEnd, slug, version)
+ if err != nil {
+ return fmt.Errorf("lifecycle/postgres-versions: MarkRetained: %w", err)
+ }
+ if tag.RowsAffected() == 0 {
+ return fmt.Errorf("lifecycle/postgres-versions: MarkRetained %q/%q: not found", slug, version)
+ }
+ return nil
+}
+
+// PromoteToActive swaps a retained row to active and flips the
+// previous active row to retiring in the same transaction.
+func (s *PostgresVersionLog) PromoteToActive(ctx context.Context, slug, version string) error {
+ pool, ok := s.db.(beginTxer)
+ if !ok {
+ return s.promoteWithin(ctx, s.db, slug, version)
+ }
+ tx, err := pool.Begin(ctx)
+ if err != nil {
+ return fmt.Errorf("lifecycle/postgres-versions: PromoteToActive begin: %w", err)
+ }
+ defer func() { _ = tx.Rollback(ctx) }()
+ if err := s.promoteWithin(ctx, txQuerier{tx: tx}, slug, version); err != nil {
+ return err
+ }
+ return tx.Commit(ctx)
+}
+
+func (s *PostgresVersionLog) promoteWithin(ctx context.Context, q PgxQuerier, slug, version string) error {
+ // The target row must exist and be retained.
+ var existingState string
+ if err := q.QueryRow(ctx,
+ `SELECT state FROM plugin_version_log WHERE slug = $1 AND version = $2`,
+ slug, version,
+ ).Scan(&existingState); err != nil {
+ if errors.Is(err, pgx.ErrNoRows) {
+ return fmt.Errorf("%w: %q/%q", ErrNoRollback, slug, version)
+ }
+ return fmt.Errorf("promote lookup: %w", err)
+ }
+ if existingState != string(VersionRetained) {
+ return fmt.Errorf("%w: %q/%q not retained (state=%s)", ErrNoRollback, slug, version, existingState)
+ }
+
+ now := time.Now().UTC()
+ // Flip current active → retiring.
+ _, err := q.Exec(ctx, `
+ UPDATE plugin_version_log
+ SET state = 'retiring', retired_at = $1
+ WHERE slug = $2 AND state = 'active'`, now, slug)
+ if err != nil {
+ return fmt.Errorf("promote flip active: %w", err)
+ }
+ // Promote target → active.
+ _, err = q.Exec(ctx, `
+ UPDATE plugin_version_log
+ SET state = 'active', activated_at = $1, retention_end = NULL
+ WHERE slug = $2 AND version = $3`, now, slug, version)
+ if err != nil {
+ return fmt.Errorf("promote target: %w", err)
+ }
+ return nil
+}
+
+// MarkRetired moves a row to retired (fully unloaded).
+func (s *PostgresVersionLog) MarkRetired(ctx context.Context, slug, version string) error {
+ tag, err := s.db.Exec(ctx,
+ `UPDATE plugin_version_log SET state = 'retired' WHERE slug = $1 AND version = $2`,
+ slug, version)
+ if err != nil {
+ return fmt.Errorf("lifecycle/postgres-versions: MarkRetired: %w", err)
+ }
+ if tag.RowsAffected() == 0 {
+ return fmt.Errorf("lifecycle/postgres-versions: MarkRetired %q/%q: not found", slug, version)
+ }
+ return nil
+}
+
+// ListRetained returns retained rows for slug, newest first.
+func (s *PostgresVersionLog) ListRetained(ctx context.Context, slug string) ([]VersionRow, error) {
+ rows, err := s.db.Query(ctx, `
+ SELECT slug, version, abi_version, installed_at,
+ COALESCE(activated_at, 'epoch'::TIMESTAMPTZ),
+ COALESCE(retired_at, 'epoch'::TIMESTAMPTZ),
+ COALESCE(retention_end, 'epoch'::TIMESTAMPTZ)
+ FROM plugin_version_log
+ WHERE slug = $1 AND state = 'retained'
+ ORDER BY installed_at DESC`, slug)
+ if err != nil {
+ return nil, fmt.Errorf("lifecycle/postgres-versions: ListRetained: %w", err)
+ }
+ defer rows.Close()
+ var out []VersionRow
+ for rows.Next() {
+ var r VersionRow
+ var actAt, retAt, retEnd time.Time
+ if err := rows.Scan(&r.Slug, &r.Version, &r.ABIVersion,
+ &r.InstalledAt, &actAt, &retAt, &retEnd); err != nil {
+ return nil, fmt.Errorf("lifecycle/postgres-versions: ListRetained scan: %w", err)
+ }
+ r.State = VersionRetained
+ if !isEpoch(actAt) {
+ r.ActivatedAt = actAt
+ }
+ if !isEpoch(retAt) {
+ r.RetiredAt = retAt
+ }
+ if !isEpoch(retEnd) {
+ r.RetentionEnd = retEnd
+ }
+ out = append(out, r)
+ }
+ return out, rows.Err()
+}
+
+// PurgeExpired drops retained rows whose retention_end has passed
+// and any retired rows. Returns the count purged.
+func (s *PostgresVersionLog) PurgeExpired(ctx context.Context, now time.Time) (int, error) {
+ tag, err := s.db.Exec(ctx, `
+ DELETE FROM plugin_version_log
+ WHERE (state = 'retained' AND retention_end < $1)
+ OR state = 'retired'`, now)
+ if err != nil {
+ return 0, fmt.Errorf("lifecycle/postgres-versions: PurgeExpired: %w", err)
+ }
+ return int(tag.RowsAffected()), nil
+}
+
+// UpdateActiveVersion satisfies VersionedStorage on PostgresStorage:
+// rewrites the version/manifest/abi on an Active plugins row without
+// going through the state CAS. Reused by Update + Rollback.
+func (s *PostgresStorage) UpdateActiveVersion(ctx context.Context, slug, version string, manifestBytes []byte, abiVersion int) error {
+ if len(manifestBytes) == 0 {
+ // Rollback path — keep the manifest column untouched.
+ tag, err := s.db.Exec(ctx, `
+ UPDATE plugins
+ SET version = $1,
+ abi_version = COALESCE(NULLIF($2, 0), abi_version),
+ row_version = row_version + 1,
+ updated_at = $3
+ WHERE slug = $4 AND state = 'active'`, version, abiVersion, s.now().UTC(), slug)
+ if err != nil {
+ return fmt.Errorf("lifecycle/postgres: UpdateActiveVersion (no manifest): %w", err)
+ }
+ if tag.RowsAffected() == 0 {
+ return fmt.Errorf("lifecycle/postgres: UpdateActiveVersion: row not active for %q", slug)
+ }
+ return nil
+ }
+ tag, err := s.db.Exec(ctx, `
+ UPDATE plugins
+ SET version = $1,
+ abi_version = COALESCE(NULLIF($2, 0), abi_version),
+ manifest = $3::JSONB,
+ row_version = row_version + 1,
+ updated_at = $4
+ WHERE slug = $5 AND state = 'active'`, version, abiVersion, string(manifestBytes), s.now().UTC(), slug)
+ if err != nil {
+ return fmt.Errorf("lifecycle/postgres: UpdateActiveVersion: %w", err)
+ }
+ if tag.RowsAffected() == 0 {
+ return fmt.Errorf("lifecycle/postgres: UpdateActiveVersion: row not active for %q", slug)
+ }
+ return nil
+}
+
+// beginTxer is the subset of *pgxpool.Pool we use to open an explicit
+// transaction. Declared as an interface so the package doesn't need a
+// direct dependency on pgxpool for this one operation; both the pool
+// and the testutil fake satisfy it (or don't, in which case we fall
+// through to the no-transaction branch).
+type beginTxer interface {
+ Begin(ctx context.Context) (pgx.Tx, error)
+}
+
+// txQuerier adapts a pgx.Tx to the PgxQuerier interface so the
+// statements in AppendActive / PromoteToActive can run against it
+// without changing signatures.
+type txQuerier struct {
+ tx pgx.Tx
+}
+
+func (t txQuerier) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row {
+ return t.tx.QueryRow(ctx, sql, args...)
+}
+func (t txQuerier) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) {
+ return t.tx.Query(ctx, sql, args...)
+}
+func (t txQuerier) Exec(ctx context.Context, sql string, args ...any) (pgconn.CommandTag, error) {
+ return t.tx.Exec(ctx, sql, args...)
+}
+
+// Compile-time check.
+var _ VersionLog = (*PostgresVersionLog)(nil)
diff --git a/packages/go/plugins/manifest/manifest.go b/packages/go/plugins/manifest/manifest.go
index 71730bf7..a3d595fd 100644
--- a/packages/go/plugins/manifest/manifest.go
+++ b/packages/go/plugins/manifest/manifest.go
@@ -104,6 +104,13 @@ type Manifest struct {
// supplies (typically a small budget).
Storage *Storage `json:"storage,omitempty"`
+ // Flags is the opt-in capability flag bag. Each flag is a small
+ // boolean toggle that switches the plugin into a non-default
+ // dispatch mode. New flags land here without breaking existing
+ // manifests; unknown flags are accepted (additionalProperties is
+ // false at the top level but Flags is its own object).
+ Flags *Flags `json:"flags,omitempty"`
+
// Raw is the original bytes the manifest was decoded from. Callers
// that need to persist the manifest verbatim (e.g. the lifecycle
// Plugin row) read this instead of re-marshalling — re-marshal
@@ -112,6 +119,19 @@ type Manifest struct {
Raw json.RawMessage `json:"-"`
}
+// Flags is the manifest's flags bag — currently only one entry, set
+// to opt the plugin into the issue #263 ApplyBatch hot path. Future
+// flags will accumulate here without breaking compatibility.
+type Flags struct {
+ // ApplyFiltersBatch, when true, signals that the plugin's filter
+ // handlers want to be invoked with a whole []any slice in one call
+ // rather than once per item. The host wiring layer reads this flag
+ // and dispatches RegisterBatchFilter on the hook bus rather than
+ // RegisterFilter. Default false; legacy plugins keep the per-item
+ // contract.
+ ApplyFiltersBatch bool `json:"apply_filters_batch,omitempty"`
+}
+
// Hooks is the actions/filters split. Both arrays are optional; an
// empty Hooks object is legal (and pointless).
type Hooks struct {
diff --git a/packages/go/plugins/manifest/schema.json b/packages/go/plugins/manifest/schema.json
index ad1a1a87..1a5d7d77 100644
--- a/packages/go/plugins/manifest/schema.json
+++ b/packages/go/plugins/manifest/schema.json
@@ -135,6 +135,17 @@
}
}
}
+ },
+ "flags": {
+ "description": "Opt-in capability flags. Each flag switches the plugin into a non-default dispatch mode. Unknown flags are rejected so a typo doesn't silently flip the plugin into legacy behaviour.",
+ "type": "object",
+ "additionalProperties": false,
+ "properties": {
+ "apply_filters_batch": {
+ "description": "When true, the host invokes the plugin's filter handlers with a whole []any slice in one call (issue #263). The plugin SDK exposes the batched signature; the per-item ApplyFilters path remains the default for legacy plugins.",
+ "type": "boolean"
+ }
+ }
}
}
}