fix: harden plugin failure recovery (#2169)

This commit is contained in:
Carlos Florêncio
2026-08-02 22:25:27 +01:00
committed by GitHub
parent d6602c9fdf
commit 87239d1b29
35 changed files with 1699 additions and 160 deletions
@@ -38,10 +38,11 @@ import (
// Tests override these via Option so retry/backoff loops don't rely on real
// sleeps.
const (
DefaultRequestTimeout = 10 * time.Second
DefaultQueueSize = 100
DefaultRingBufferCap = 100
DefaultRingBufferTTL = 5 * time.Minute
DefaultRequestTimeout = 10 * time.Second
DefaultQueueSize = 100
DefaultRingBufferCap = 100
DefaultRingBufferTTL = 5 * time.Minute
DefaultOverflowLogInterval = time.Minute
)
// DefaultRetryDelays is the backoff schedule between delivery attempts
@@ -115,6 +116,13 @@ func WithNow(fn func() time.Time) Option {
return func(d *Deliverer) { d.nowFn = fn }
}
// WithOverflowLogInterval overrides the per-plugin ring-buffer overflow log
// aggregation window. Production uses one minute; tests inject a shorter
// window or advance an injected clock instead of sleeping.
func WithOverflowLogInterval(interval time.Duration) Option {
return func(d *Deliverer) { d.overflowLogInterval = interval }
}
// withOnProcessed is an unexported Option (only constructible from within
// this package) used by tests to observe exactly when a worker finishes
// handling each queued Delivery, so a test can wait for "all N events have
@@ -136,12 +144,13 @@ type Deliverer struct {
lister PluginLister
log *logger.Logger
requestTimeout time.Duration
retryDelays []time.Duration
queueSize int
ringBufferCap int
ringBufferTTL time.Duration
nowFn func() time.Time
requestTimeout time.Duration
retryDelays []time.Duration
queueSize int
ringBufferCap int
ringBufferTTL time.Duration
overflowLogInterval time.Duration
nowFn func() time.Time
mu sync.Mutex
workers map[string]*pluginWorker
@@ -160,17 +169,18 @@ type Deliverer struct {
// from the current plugin set.
func New(eventBus bus.EventBus, transport Transport, lister PluginLister, log *logger.Logger, opts ...Option) *Deliverer {
d := &Deliverer{
eventBus: eventBus,
transport: transport,
lister: lister,
log: log,
requestTimeout: DefaultRequestTimeout,
retryDelays: DefaultRetryDelays(),
queueSize: DefaultQueueSize,
ringBufferCap: DefaultRingBufferCap,
ringBufferTTL: DefaultRingBufferTTL,
nowFn: time.Now,
workers: make(map[string]*pluginWorker),
eventBus: eventBus,
transport: transport,
lister: lister,
log: log,
requestTimeout: DefaultRequestTimeout,
retryDelays: DefaultRetryDelays(),
queueSize: DefaultQueueSize,
ringBufferCap: DefaultRingBufferCap,
ringBufferTTL: DefaultRingBufferTTL,
overflowLogInterval: DefaultOverflowLogInterval,
nowFn: time.Now,
workers: make(map[string]*pluginWorker),
}
for _, opt := range opts {
opt(d)
@@ -267,11 +277,13 @@ func (d *Deliverer) newWorker(id string) *pluginWorker {
return &pluginWorker{
id: id,
deps: &workerDeps{
transport: d.transport,
requestTimeout: d.requestTimeout,
retryDelays: d.retryDelays,
log: d.log,
onProcessed: d.onProcessed,
transport: d.transport,
requestTimeout: d.requestTimeout,
retryDelays: d.retryDelays,
log: d.log,
nowFn: d.nowFn,
overflowLogInterval: d.overflowLogInterval,
onProcessed: d.onProcessed,
},
queue: make(chan Delivery, d.queueSize),
buffer: newRingBuffer(d.ringBufferCap, d.ringBufferTTL, d.nowFn),
@@ -357,11 +369,13 @@ func (d *Deliverer) makeHandler(w *pluginWorker, pattern string) bus.EventHandle
// delivery, factored out of Deliverer so pluginWorker doesn't hold a back
// reference to it.
type workerDeps struct {
transport Transport
requestTimeout time.Duration
retryDelays []time.Duration
log *logger.Logger
onProcessed func(pluginID string, d Delivery)
transport Transport
requestTimeout time.Duration
retryDelays []time.Duration
log *logger.Logger
nowFn func() time.Time
overflowLogInterval time.Duration
onProcessed func(pluginID string, d Delivery)
}
// pluginWorker owns sequential delivery for one plugin: a bounded queue
@@ -388,9 +402,12 @@ type pluginWorker struct {
deliverCtx context.Context
cancelDeliver context.CancelFunc
mu sync.Mutex
subs []bus.Subscription
status string
mu sync.Mutex
subs []bus.Subscription
status string
overflowLogAt time.Time
overflowLogSet bool
overflowDropped int
}
// start launches the worker's run loop. Must be called at most once.
@@ -479,8 +496,7 @@ func (w *pluginWorker) run() {
func (w *pluginWorker) process(d Delivery) {
if w.snapshotStatus() == store.StatusError {
if dropped := w.buffer.Add(d); dropped != "" {
w.deps.log.Warn("plugin event ring buffer overflow, dropped oldest buffered event",
zap.String("plugin_id", w.id), zap.String("dropped_delivery_id", dropped))
w.logOverflow(dropped)
}
} else {
w.deliverWithRetry(w.deliverCtx, d)
@@ -490,6 +506,30 @@ func (w *pluginWorker) process(d Delivery) {
}
}
// logOverflow rate-limits ring-buffer overflow warnings per plugin while
// retaining an aggregate count of suppressed drops. The first drop is logged
// immediately; the first drop after the interval logs every drop accumulated
// since the previous warning.
func (w *pluginWorker) logOverflow(dropped string) {
now := w.deps.nowFn().UTC()
w.overflowDropped++
count := w.overflowDropped
shouldLog := !w.overflowLogSet || !now.Before(w.overflowLogAt.Add(w.deps.overflowLogInterval))
if shouldLog {
w.overflowDropped = 0
w.overflowLogAt = now
w.overflowLogSet = true
}
if !shouldLog {
return
}
w.deps.log.Warn("plugin event ring buffer overflow, dropped oldest buffered event",
zap.String("plugin_id", w.id),
zap.String("latest_dropped_delivery_id", dropped),
zap.Int("dropped_count", count))
}
// deliverWithRetry attempts d up to 1+len(retryDelays) times, sleeping
// (interruptibly) between attempts per the configured backoff schedule.
// Logs and gives up (event is lost from the live-delivery path; a
@@ -12,6 +12,9 @@ import (
"github.com/kandev/kandev/internal/events/bus"
"github.com/kandev/kandev/internal/plugins/store"
"github.com/kandev/kandev/pkg/pluginsdk"
"go.uber.org/zap"
"go.uber.org/zap/zapcore"
"go.uber.org/zap/zaptest/observer"
)
// fakeTransport is a controllable Transport for tests: DeliverEvent calls
@@ -351,6 +354,124 @@ func TestDeliverer_RingBufferOverflowDropsOldest(t *testing.T) {
requireTimeout(t, arrivedCh, 150*time.Millisecond, "a 3rd replayed event")
}
func TestDeliverer_RingBufferOverflowLogsAggregatedDrops(t *testing.T) {
clock := time.Date(2026, time.August, 2, 12, 0, 0, 0, time.UTC)
var clockMu sync.Mutex
now := func() time.Time {
clockMu.Lock()
defer clockMu.Unlock()
return clock
}
core, observed := observer.New(zapcore.WarnLevel)
log, err := logger.NewFromZap(zap.New(core))
if err != nil {
t.Fatalf("NewFromZap(): %v", err)
}
eventBus := bus.NewMemoryEventBus(logger.Default())
lister := &fakeLister{}
lister.set(PluginRecord{ID: "plug1", EventSubjects: []string{"task.*"}, Status: store.StatusError})
processedCh := make(chan struct{}, 10)
d := New(eventBus, &fakeTransport{}, lister, log,
WithRetryDelays(nil),
WithRingBuffer(1, 5*time.Minute),
WithNow(now),
WithOverflowLogInterval(time.Minute),
withOnProcessed(func(string, Delivery) { processedCh <- struct{}{} }))
t.Cleanup(d.Stop)
d.Refresh()
ctx := context.Background()
for _, subject := range []string{"task.created", "task.updated", "task.deleted"} {
if err := eventBus.Publish(ctx, subject, bus.NewEvent(subject, "test", map[string]interface{}{})); err != nil {
t.Fatalf("Publish(%s): %v", subject, err)
}
}
for i := 0; i < 3; i++ {
requireNoTimeout(t, processedCh, 2*time.Second, fmt.Sprintf("event %d buffered", i+1))
}
entries := observed.All()
if len(entries) != 1 {
t.Fatalf("overflow warnings after two drops = %d, want 1", len(entries))
}
if got := entries[0].ContextMap()["dropped_count"]; got != int64(1) {
t.Fatalf("first dropped_count = %v, want 1", got)
}
if got := entries[0].ContextMap()["latest_dropped_delivery_id"]; got == nil || got == "" {
t.Fatalf("first latest_dropped_delivery_id = %v, want a non-empty id", got)
}
if _, ok := entries[0].ContextMap()["dropped_delivery_id"]; ok {
t.Fatal("first overflow warning still uses the ambiguous dropped_delivery_id field")
}
clockMu.Lock()
clock = clock.Add(time.Minute)
clockMu.Unlock()
if err := eventBus.Publish(ctx, "task.archived", bus.NewEvent("task.archived", "test", map[string]interface{}{})); err != nil {
t.Fatalf("Publish(task.archived): %v", err)
}
requireNoTimeout(t, processedCh, 2*time.Second, "post-interval event buffered")
entries = observed.All()
if len(entries) != 2 {
t.Fatalf("overflow warnings after interval = %d, want 2", len(entries))
}
if got := entries[1].ContextMap()["dropped_count"]; got != int64(2) {
t.Fatalf("aggregated dropped_count = %v, want 2", got)
}
if got := entries[1].ContextMap()["latest_dropped_delivery_id"]; got == nil || got == "" {
t.Fatalf("aggregated latest_dropped_delivery_id = %v, want a non-empty id", got)
}
if _, ok := entries[1].ContextMap()["dropped_delivery_id"]; ok {
t.Fatal("aggregated overflow warning still uses the ambiguous dropped_delivery_id field")
}
}
func TestDeliverer_RingBufferOverflowLimiterIsPerPlugin(t *testing.T) {
core, observed := observer.New(zapcore.WarnLevel)
log, err := logger.NewFromZap(zap.New(core))
if err != nil {
t.Fatalf("NewFromZap(): %v", err)
}
eventBus := bus.NewMemoryEventBus(logger.Default())
lister := &fakeLister{}
lister.set(
PluginRecord{ID: "plug1", EventSubjects: []string{"task.*"}, Status: store.StatusError},
PluginRecord{ID: "plug2", EventSubjects: []string{"task.*"}, Status: store.StatusError},
)
processedCh := make(chan struct{}, 10)
d := New(eventBus, &fakeTransport{}, lister, log,
WithRetryDelays(nil),
WithRingBuffer(1, 5*time.Minute),
WithOverflowLogInterval(time.Minute),
withOnProcessed(func(string, Delivery) { processedCh <- struct{}{} }))
t.Cleanup(d.Stop)
d.Refresh()
for _, subject := range []string{"task.created", "task.updated"} {
if err := eventBus.Publish(context.Background(), subject, bus.NewEvent(subject, "test", map[string]interface{}{})); err != nil {
t.Fatalf("Publish(%s): %v", subject, err)
}
}
for i := 0; i < 4; i++ {
requireNoTimeout(t, processedCh, 2*time.Second, fmt.Sprintf("plugin event %d buffered", i+1))
}
entries := observed.All()
if len(entries) != 2 {
t.Fatalf("overflow warnings for two plugins = %d, want 2", len(entries))
}
seen := map[string]bool{}
for _, entry := range entries {
seen[entry.ContextMap()["plugin_id"].(string)] = true
}
if !seen["plug1"] || !seen["plug2"] {
t.Fatalf("overflow warning plugin ids = %v, want both plugins", seen)
}
}
func TestDeliverer_RefreshStopsDeliveryWhenPluginRemoved(t *testing.T) {
arrivedCh := make(chan struct{}, 5)
transport := &fakeTransport{}
@@ -0,0 +1,73 @@
package plugins
import (
"os"
"regexp"
"strings"
"sync"
"unicode/utf8"
)
// maxPluginErrorBytes bounds the diagnostic persisted into a plugin record.
// Failure messages can originate in subprocess/transport errors, so keeping
// them bounded prevents an unexpectedly verbose error from bloating the
// record or the settings API response.
const maxPluginErrorBytes = 2048
var (
pluginBearerTokenPattern = regexp.MustCompile(`(?i)\bBearer\s+[A-Za-z0-9._~+/=-]+`)
pluginLabeledSecretPattern = regexp.MustCompile(`(?i)\b((?:[A-Za-z0-9]+[_-])*(?:password|passwd|passphrase|secret|token|api[_ -]?(?:key|token|secret)|access[_ -]?token|refresh[_ -]?token|pat))\b(\s*[:=]\s*)(?:"[^"]*"|'[^']*'|[^\s,;}]+)`)
pluginPATPattern = regexp.MustCompile(`\b(?:github_pat_[A-Za-z0-9_]+|gh[pousr]_[A-Za-z0-9]+|glpat-[A-Za-z0-9_-]+)\b`)
pluginHomePathPattern = regexp.MustCompile(`(/Users/[^/\s]+|/home/[^/\s]+|/root)(/|$)`)
pluginWindowsHomePattern = regexp.MustCompile(`(?i)([A-Za-z]:[/\\]Users[/\\][^/\\\s]+)([/\\]|$)`)
pluginURLCredentialPattern = regexp.MustCompile(`(?i)(https?|ftp)://[^:@/\s]+:[^@/\s]+@`)
pluginHomePatternOnce sync.Once
cachedPluginHomePattern *regexp.Regexp
)
// normalizePluginError turns a runtime failure into a compact, single-line
// diagnostic suitable for persistence and display. Runtime errors can include
// arbitrary go-plugin subprocess output, so credential-like values and home
// paths are removed before the bounded message is stored.
func normalizePluginError(err error) string {
if err == nil {
return ""
}
message := strings.Join(strings.Fields(err.Error()), " ")
message = strings.ToValidUTF8(message, "�")
message = redactPluginError(message)
if len([]byte(message)) <= maxPluginErrorBytes {
return message
}
return truncatePluginError(message)
}
func truncatePluginError(message string) string {
const marker = "…"
content := []byte(message)
cut := maxPluginErrorBytes - len([]byte(marker))
for cut > 0 && cut < len(content) && !utf8.RuneStart(content[cut]) {
cut--
}
return string(content[:cut]) + marker
}
func redactPluginError(message string) string {
pluginHomePatternOnce.Do(func() {
if home, err := os.UserHomeDir(); err == nil && home != "" && home != "/" {
cachedPluginHomePattern = regexp.MustCompile(regexp.QuoteMeta(home) + `([/\\]|$)`)
}
})
if cachedPluginHomePattern != nil {
message = cachedPluginHomePattern.ReplaceAllString(message, "~$1")
}
message = pluginHomePathPattern.ReplaceAllString(message, "~$2")
message = pluginWindowsHomePattern.ReplaceAllString(message, "~$2")
message = pluginURLCredentialPattern.ReplaceAllString(message, "${1}://[REDACTED]@")
message = pluginBearerTokenPattern.ReplaceAllString(message, "Bearer [REDACTED]")
message = pluginLabeledSecretPattern.ReplaceAllString(message, "${1}${2}[REDACTED]")
message = pluginPATPattern.ReplaceAllString(message, "[REDACTED]")
return message
}
@@ -0,0 +1,113 @@
package plugins
import (
"errors"
"os"
"strings"
"testing"
)
func TestNormalizePluginErrorCollapsesWhitespace(t *testing.T) {
got := normalizePluginError(errors.New(" handshake\nfailed\twhile starting "))
if got != "handshake failed while starting" {
t.Fatalf("normalizePluginError() = %q, want %q", got, "handshake failed while starting")
}
}
func TestNormalizePluginErrorBoundsUTF8Message(t *testing.T) {
got := normalizePluginError(errors.New(strings.Repeat("é", maxPluginErrorBytes)))
if len([]byte(got)) > maxPluginErrorBytes {
t.Fatalf("normalizePluginError() returned %d bytes, want <= %d", len([]byte(got)), maxPluginErrorBytes)
}
if !strings.HasSuffix(got, "…") {
t.Fatalf("normalizePluginError() = %q, want a truncation marker", got)
}
}
func TestNormalizePluginErrorNilIsEmpty(t *testing.T) {
if got := normalizePluginError(nil); got != "" {
t.Fatalf("normalizePluginError(nil) = %q, want empty", got)
}
}
func TestNormalizePluginErrorRedactsCredentialsAndHomePath(t *testing.T) {
home, err := os.UserHomeDir()
if err != nil {
t.Fatalf("os.UserHomeDir() error: %v", err)
}
const (
classicPAT = "ghp_abcdefghijklmnopqrstuvwxyz1234567890AB"
bearer = "eyJhbGciOiJIUzI1NiJ9.super-secret-signature"
password = "hunter2-rocks"
token = "plugin-token-value"
secret = "s3cret-value"
apiKey = "key-value-123"
apiToken = "api-token-value"
clientSecret = "client-secret-value"
urlPassword = "url-password-value"
)
cases := []struct {
name string
input string
secret []string
want string
}{
{
name: "personal access token",
input: "plugin stdout included PAT " + classicPAT,
secret: []string{classicPAT},
want: "[REDACTED]",
},
{
name: "bearer authorization",
input: "connect failed: Authorization: Bearer " + bearer,
secret: []string{bearer},
want: "Bearer [REDACTED]",
},
{
name: "labeled secrets",
input: "password=" + password + " token: " + token + " secret='" + secret + "' api_key=\"" + apiKey + "\" api_token=" + apiToken,
secret: []string{password, token, secret, apiKey, apiToken},
want: "[REDACTED]",
},
{
name: "compound labeled secret",
input: "plugin failed with client_secret=" + clientSecret,
secret: []string{clientSecret},
want: "client_secret=[REDACTED]",
},
{
name: "URL credentials",
input: "dial failed: https://admin:" + urlPassword + "@internal.example/api",
secret: []string{urlPassword},
want: "https://[REDACTED]@internal.example/api",
},
{
name: "home path",
input: home + "/.kandev/plugins/acme/server: handshake failed",
secret: []string{home + "/"},
want: "~/.kandev/plugins/acme/server",
},
{
name: "windows home path",
input: `plugin failed at C:\Users\alice\.kandev\plugins\acme\server: handshake failed`,
secret: []string{`C:\Users\alice\`},
want: `~\.kandev\plugins\acme\server`,
},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := normalizePluginError(errors.New(tc.input))
for _, value := range tc.secret {
if strings.Contains(got, value) {
t.Fatalf("normalizePluginError() leaked %q in %q", value, got)
}
}
if !strings.Contains(got, tc.want) {
t.Fatalf("normalizePluginError() = %q, want it to contain %q", got, tc.want)
}
})
}
}
+45 -14
View File
@@ -4,6 +4,7 @@ import (
"fmt"
"sort"
"sync"
"time"
"github.com/kandev/kandev/internal/plugins/store"
)
@@ -15,7 +16,8 @@ import (
//
// Get and List return copies of the stored *store.Record so callers cannot
// mutate registry state by holding onto a returned pointer; all writes go
// through Add / Remove / SetStatus / SetRestartCount.
// through Add / Remove / SetStatus / SetRuntimeState / SetAutoUpdate /
// SetRestartCount.
type Registry struct {
mu sync.RWMutex
byID map[string]*store.Record
@@ -37,7 +39,7 @@ func (r *Registry) Load(s store.Store) error {
byID := make(map[string]*store.Record, len(records))
for _, rec := range records {
byID[rec.ID] = rec
byID[rec.ID] = cloneRecord(rec)
}
r.mu.Lock()
@@ -54,8 +56,7 @@ func (r *Registry) Get(id string) (*store.Record, bool) {
if !ok {
return nil, false
}
clone := *rec
return &clone, true
return cloneRecord(rec), true
}
// List returns a copy of every registered record, sorted by id for
@@ -65,8 +66,7 @@ func (r *Registry) List() []*store.Record {
defer r.mu.RUnlock()
out := make([]*store.Record, 0, len(r.byID))
for _, rec := range r.byID {
clone := *rec
out = append(out, &clone)
out = append(out, cloneRecord(rec))
}
sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
return out
@@ -76,8 +76,7 @@ func (r *Registry) List() []*store.Record {
func (r *Registry) Add(record *store.Record) {
r.mu.Lock()
defer r.mu.Unlock()
clone := *record
r.byID[record.ID] = &clone
r.byID[record.ID] = cloneRecord(record)
}
// Remove deletes the record for id, if present. A no-op if id is unknown.
@@ -99,8 +98,7 @@ func (r *Registry) SetStatus(id string, status Status) (*store.Record, bool) {
return nil, false
}
rec.Status = status
clone := *rec
return &clone, true
return cloneRecord(rec), true
}
// SetAutoUpdate updates the in-memory per-plugin auto-update override for id
@@ -115,8 +113,7 @@ func (r *Registry) SetAutoUpdate(id string, v *bool) (*store.Record, bool) {
return nil, false
}
rec.AutoUpdate = v
clone := *rec
return &clone, true
return cloneRecord(rec), true
}
// SetRestartCount updates the in-memory restart count for id and returns a
@@ -131,6 +128,40 @@ func (r *Registry) SetRestartCount(id string, count int) (*store.Record, bool) {
return nil, false
}
rec.RestartCount = count
clone := *rec
return &clone, true
return cloneRecord(rec), true
}
// SetRuntimeState updates the lifecycle status and persisted failure
// diagnostic for id and returns a copy of the resulting record. The caller is
// responsible for FSM validation and persistence; this method only performs
// the in-memory mutation under the registry lock.
func (r *Registry) SetRuntimeState(id string, status Status, lastError string, lastErrorAt *time.Time) (*store.Record, bool) {
r.mu.Lock()
defer r.mu.Unlock()
rec, ok := r.byID[id]
if !ok {
return nil, false
}
rec.Status = status
rec.LastError = lastError
if lastErrorAt == nil {
rec.LastErrorAt = nil
} else {
at := *lastErrorAt
rec.LastErrorAt = &at
}
return cloneRecord(rec), true
}
func cloneRecord(rec *store.Record) *store.Record {
clone := *rec
if rec.AutoUpdate != nil {
autoUpdate := *rec.AutoUpdate
clone.AutoUpdate = &autoUpdate
}
if rec.LastErrorAt != nil {
at := *rec.LastErrorAt
clone.LastErrorAt = &at
}
return &clone
}
+10 -1
View File
@@ -105,13 +105,19 @@ func TestRegistrySetStatusMissingReturnsNotOK(t *testing.T) {
func TestRegistryGetReturnsIndependentCopy(t *testing.T) {
reg := NewRegistry()
reg.Add(&store.Record{Manifest: *testManifest("kandev-plugin-slack"), Status: store.StatusRegistered})
autoUpdate := true
reg.Add(&store.Record{
Manifest: *testManifest("kandev-plugin-slack"),
Status: store.StatusRegistered,
AutoUpdate: &autoUpdate,
})
rec, ok := reg.Get("kandev-plugin-slack")
if !ok {
t.Fatalf("Get() expected ok = true")
}
rec.Status = store.StatusActive // mutate the returned copy
*rec.AutoUpdate = false // mutate the returned pointer copy
fresh, ok := reg.Get("kandev-plugin-slack")
if !ok {
@@ -120,6 +126,9 @@ func TestRegistryGetReturnsIndependentCopy(t *testing.T) {
if fresh.Status != store.StatusRegistered {
t.Fatalf("mutating a Get() result leaked into the registry: Status = %q, want %q", fresh.Status, store.StatusRegistered)
}
if fresh.AutoUpdate == nil || !*fresh.AutoUpdate {
t.Fatalf("mutating a Get() result leaked into the registry: AutoUpdate = %v, want true", fresh.AutoUpdate)
}
}
func TestRegistrySetRestartCountUpdatesRecord(t *testing.T) {
@@ -88,7 +88,7 @@ func WithStartTimeout(d time.Duration) Option {
type Manager struct {
pluginsDir string
log *logger.Logger
onStatusChange func(id string, healthy bool)
onStatusChange func(id string, healthy bool, reason error)
pingInterval time.Duration
maxConsecutiveFailures int
@@ -124,9 +124,11 @@ type Manager struct {
// store.FSStore persists records under — KANDEV_PLUGIN_DATA_DIR for plugin
// id "x" is pluginsDir/x/data). onStatusChange is invoked from the
// supervision loop's own goroutine whenever a running plugin's health
// transitions (degraded -> unhealthy, or a restart recovers it); it must
// not block. May be nil (e.g. tests that don't care about transitions).
func NewManager(pluginsDir string, onStatusChange func(id string, healthy bool), log *logger.Logger, opts ...Option) *Manager {
// transitions (degraded -> unhealthy, or a restart recovers it); reason is
// the triggering ping/process/restart error for unhealthy transitions and nil
// on recovery. The callback must not block. May be nil (e.g. tests that don't
// care about transitions).
func NewManager(pluginsDir string, onStatusChange func(id string, healthy bool, reason error), log *logger.Logger, opts ...Option) *Manager {
m := &Manager{
pluginsDir: pluginsDir,
log: log,
@@ -236,27 +238,34 @@ func (m *Manager) consumeStopRequested(id string) bool {
// tracked entry for id has already permanently given up (exhausted its
// restart attempts — current==nil, gaveUp==true), it is dropped so a fresh
// Start can respawn cleanly rather than being rejected as "already
// running" against a dead entry. p.stop() on a gave-up process is a fast,
// safe no-op (its supervision loop has already exited), so this is safe to
// do while holding m.mu.
// running" against a dead entry. The exhausted process is stopped after m.mu
// is released: p.stop() waits for the supervision goroutine, and its final
// status callback may call back into Manager and need the same mutex.
func (m *Manager) claimStart(id string) error {
m.mu.Lock()
defer m.mu.Unlock()
if _, inFlight := m.starting[id]; inFlight {
m.mu.Unlock()
return fmt.Errorf("plugins/runtime: plugin %q is already starting", id)
}
var exhausted *process
if p, exists := m.processes[id]; exists {
if !p.isGaveUp() {
m.mu.Unlock()
return fmt.Errorf("plugins/runtime: plugin %q is already running", id)
}
delete(m.processes, id)
p.stop()
exhausted = p
}
if m.starting == nil {
m.starting = make(map[string]struct{})
}
m.starting[id] = struct{}{}
m.mu.Unlock()
if exhausted != nil {
exhausted.stop()
}
return nil
}
@@ -270,7 +270,7 @@ func TestManager_CrashTriggersAutomaticRestart(t *testing.T) {
var mu sync.Mutex
var transitions []bool
onStatusChange := func(_ string, healthy bool) {
onStatusChange := func(_ string, healthy bool, _ error) {
mu.Lock()
transitions = append(transitions, healthy)
mu.Unlock()
@@ -502,6 +502,60 @@ func TestManager_StartAfterExhaustionRespawns(t *testing.T) {
}
}
// TestManager_ConcurrentEnableDuringRestartExhaustionDoesNotDeadlock pins the
// lock ordering between the final supervision callback and a manual Enable.
// The callback deliberately re-enters Manager.RestartCount after a barrier;
// Enable must be able to stop the exhausted process without holding m.mu while
// it waits for that callback to return.
func TestManager_ConcurrentEnableDuringRestartExhaustionDoesNotDeadlock(t *testing.T) {
callbackEntered := make(chan struct{})
releaseCallback := make(chan struct{})
var m *Manager
m = NewManager(t.TempDir(), func(id string, healthy bool, reason error) {
if healthy || reason == nil || reason.Error() != "plugin restart attempts exhausted" {
return
}
close(callbackEntered)
<-releaseCallback
_ = m.RestartCount(id)
}, testLogger(t))
t.Cleanup(m.StopAll)
const id = "deadlock-plugin"
p := newProcess(id, testLogger(t), nil, m.onStatusChange)
p.maxRestartAttempts = 0
p.current = &fakeSpawnedProcess{}
m.mu.Lock()
m.processes[id] = p
m.mu.Unlock()
p.wg.Add(1)
go func() {
defer p.wg.Done()
_ = p.handleFailureAndRestart(errors.New("initial failure"))
}()
<-callbackEntered
startDone := make(chan error, 1)
go func() {
startDone <- m.startProcess(id, func() (spawnedProcess, error) {
return &fakeSpawnedProcess{}, nil
})
}()
<-p.stopCh
close(releaseCallback)
select {
case err := <-startDone:
if err != nil {
t.Fatalf("startProcess() returned unexpected error: %v", err)
}
case <-time.After(time.Second):
t.Fatal("concurrent Enable remained blocked after restart-exhaustion callback was released")
}
m.Stop(id)
}
// TestManager_DefaultStartTimeoutIs30s pins the fix for bounded startup:
// Manager.Start previously spawned via a hcplugin.Client with no configured
// StartTimeout, defaulting to go-plugin's own 1-minute handshake timeout, so
@@ -1,6 +1,7 @@
package runtime
import (
"fmt"
"sync"
"time"
@@ -53,7 +54,7 @@ type process struct {
log *logger.Logger
spawnFn func() (spawnedProcess, error)
onStatusChange func(id string, healthy bool)
onStatusChange func(id string, healthy bool, reason error)
pingInterval time.Duration
maxConsecutiveFails int
@@ -84,7 +85,7 @@ type process struct {
// newProcess builds a process with default tuning. Callers may override
// pingInterval/maxConsecutiveFails/restartBackoff/maxRestartAttempts/
// sleepFn/onTick before calling start.
func newProcess(id string, log *logger.Logger, spawnFn func() (spawnedProcess, error), onStatusChange func(string, bool)) *process {
func newProcess(id string, log *logger.Logger, spawnFn func() (spawnedProcess, error), onStatusChange func(string, bool, error)) *process {
return &process{
id: id,
log: log,
@@ -227,8 +228,9 @@ func (p *process) tick() bool {
}
if cur.Exited() {
reason := fmt.Errorf("plugin process exited unexpectedly")
p.log.Warn("plugin process exited unexpectedly", zap.String("plugin_id", p.id))
return p.handleFailureAndRestart()
return p.handleFailureAndRestart(reason)
}
if err := cur.Ping(); err != nil {
@@ -239,7 +241,7 @@ func (p *process) tick() bool {
if reachedThreshold {
p.log.Warn("plugin health check failed repeatedly",
zap.String("plugin_id", p.id), zap.Error(err))
return p.handleFailureAndRestart()
return p.handleFailureAndRestart(err)
}
return false
}
@@ -249,7 +251,7 @@ func (p *process) tick() bool {
p.failures = 0
p.mu.Unlock()
if hadFailures {
p.notify(true)
p.notify(true, nil)
}
return false
}
@@ -258,8 +260,8 @@ func (p *process) tick() bool {
// left of the old process, and attempts to respawn it with backoff up to
// maxRestartAttempts times. Returns true if every attempt failed (the
// process has given up permanently).
func (p *process) handleFailureAndRestart() bool {
p.notify(false)
func (p *process) handleFailureAndRestart(reason error) bool {
p.notify(false, reason)
p.mu.Lock()
if p.current != nil {
@@ -269,6 +271,7 @@ func (p *process) handleFailureAndRestart() bool {
p.failures = 0
p.mu.Unlock()
var lastErr error
for attempt := 0; attempt < p.maxRestartAttempts; attempt++ {
delay := time.Duration(0)
if attempt < len(p.restartBackoff) {
@@ -280,6 +283,7 @@ func (p *process) handleFailureAndRestart() bool {
proc, err := p.spawnFn()
if err != nil {
lastErr = err
p.log.Warn("plugin restart attempt failed",
zap.String("plugin_id", p.id), zap.Int("attempt", attempt+1), zap.Error(err))
continue
@@ -289,7 +293,7 @@ func (p *process) handleFailureAndRestart() bool {
p.current = proc
p.restarts++
p.mu.Unlock()
p.notify(true)
p.notify(true, nil)
return false
}
@@ -298,12 +302,17 @@ func (p *process) handleFailureAndRestart() bool {
p.mu.Lock()
p.gaveUp = true
p.mu.Unlock()
if lastErr != nil {
p.notify(false, fmt.Errorf("plugin restart attempts exhausted: %w", lastErr))
} else {
p.notify(false, fmt.Errorf("plugin restart attempts exhausted"))
}
return true
}
// notify invokes onStatusChange, if set.
func (p *process) notify(healthy bool) {
func (p *process) notify(healthy bool, reason error) {
if p.onStatusChange != nil {
p.onStatusChange(p.id, healthy)
p.onStatusChange(p.id, healthy, reason)
}
}
@@ -45,17 +45,19 @@ func (f *fakeSpawnedProcess) isKilled() bool {
return f.killed
}
// statusRecorder records every (id, healthy) pair passed to
// onStatusChange, for assertion.
// statusRecorder records every status transition and its failure cause passed
// to onStatusChange, for assertion.
type statusRecorder struct {
mu sync.Mutex
calls []bool
mu sync.Mutex
calls []bool
reasons []error
}
func (r *statusRecorder) record(_ string, healthy bool) {
func (r *statusRecorder) record(_ string, healthy bool, reason error) {
r.mu.Lock()
defer r.mu.Unlock()
r.calls = append(r.calls, healthy)
r.reasons = append(r.reasons, reason)
}
func (r *statusRecorder) snapshot() []bool {
@@ -66,6 +68,14 @@ func (r *statusRecorder) snapshot() []bool {
return out
}
func (r *statusRecorder) snapshotReasons() []error {
r.mu.Lock()
defer r.mu.Unlock()
out := make([]error, len(r.reasons))
copy(out, r.reasons)
return out
}
// noSleep is a sleepFn that never actually blocks: it returns true
// immediately unless stopCh is already closed, in which case it returns
// false. Used by every test in this file so supervision-decision logic
@@ -171,6 +181,9 @@ func TestProcess_Tick_ThresholdReachedRestartsAndNotifiesUnhealthyThenHealthy(t
if want := []bool{false, true}; !equalBoolSlices(rec.snapshot(), want) {
t.Fatalf("onStatusChange calls = %v, want %v", rec.snapshot(), want)
}
if reasons := rec.snapshotReasons(); reasons[0] == nil || reasons[0].Error() != "boom" || reasons[1] != nil {
t.Fatalf("onStatusChange reasons = %v, want [boom nil]", reasons)
}
}
func TestProcess_Tick_ExitedTriggersImmediateRestartRegardlessOfFailureCount(t *testing.T) {
@@ -189,6 +202,9 @@ func TestProcess_Tick_ExitedTriggersImmediateRestartRegardlessOfFailureCount(t *
if want := []bool{false, true}; !equalBoolSlices(rec.snapshot(), want) {
t.Fatalf("onStatusChange calls = %v, want %v (immediate restart on exit)", rec.snapshot(), want)
}
if reasons := rec.snapshotReasons(); reasons[0] == nil || reasons[0].Error() != "plugin process exited unexpectedly" || reasons[1] != nil {
t.Fatalf("onStatusChange reasons = %v, want [process exit nil]", reasons)
}
}
func TestProcess_HandleFailureAndRestart_GivesUpAfterMaxAttempts(t *testing.T) {
@@ -204,14 +220,17 @@ func TestProcess_HandleFailureAndRestart_GivesUpAfterMaxAttempts(t *testing.T) {
p.maxRestartAttempts = 3
p.restartBackoff = []time.Duration{0, 0, 0}
if gaveUp := p.handleFailureAndRestart(); !gaveUp {
if gaveUp := p.handleFailureAndRestart(errors.New("initial failure")); !gaveUp {
t.Fatal("handleFailureAndRestart() = false, want true after exhausting every attempt")
}
if spawnCalls != 3 {
t.Fatalf("spawnFn called %d times, want 3 (maxRestartAttempts)", spawnCalls)
}
if want := []bool{false}; !equalBoolSlices(rec.snapshot(), want) {
t.Fatalf("onStatusChange calls = %v, want %v (only the initial degradation, no false recovery)", rec.snapshot(), want)
if want := []bool{false, false}; !equalBoolSlices(rec.snapshot(), want) {
t.Fatalf("onStatusChange calls = %v, want %v (initial degradation plus final restart failure)", rec.snapshot(), want)
}
if reasons := rec.snapshotReasons(); reasons[0] == nil || reasons[0].Error() != "initial failure" || reasons[1] == nil || reasons[1].Error() != "plugin restart attempts exhausted: spawn failed" {
t.Fatalf("onStatusChange reasons = %v, want [initial cause, exhausted restart cause]", reasons)
}
if _, ok := p.remote(); ok {
t.Fatal("remote() ok = true after giving up, want false")
@@ -231,7 +250,7 @@ func TestProcess_HandleFailureAndRestart_StopDuringBackoffSkipsSpawn(t *testing.
close(p.stopCh) // simulate Stop() racing the backoff wait
p.sleepFn = sleepOrStop
if gaveUp := p.handleFailureAndRestart(); gaveUp {
if gaveUp := p.handleFailureAndRestart(errors.New("initial failure")); gaveUp {
t.Fatal("handleFailureAndRestart() = true, want false: stopping is not the same as giving up")
}
if spawnCalls != 0 {
+51 -16
View File
@@ -665,7 +665,7 @@ func (s *Service) restartForConfigChange(rec *store.Record) error {
ctx, cancel := context.WithTimeout(context.Background(), activateStartTimeout)
defer cancel()
if err := s.runtime.Start(ctx, rec, s.hostForPlugin); err != nil {
if setErr := s.SetStatus(rec.ID, StatusError); setErr != nil {
if setErr := s.setStatusAndDiagnostic(rec.ID, StatusError, err, true); setErr != nil {
// The restart error stays the returned error (it is the primary
// signal), but a failed status write means the registry may show
// StatusActive with no process running — don't lose that.
@@ -1010,15 +1010,17 @@ func (s *Service) Disable(id string) error {
const activateStartTimeout = 30 * time.Second
// activate spawns rec's process (if not already running) and transitions it
// to StatusActive. If the spawn fails, it best-effort transitions the
// record to StatusError (ignoring an invalid-transition failure, e.g. from
// "disabled") and returns the spawn error.
// to StatusActive. If the spawn fails, it records the failure and transitions
// the record to StatusError before returning the spawn error.
func (s *Service) activate(rec *store.Record) error {
if s.runtime != nil && !s.runtime.Running(rec.ID) {
ctx, cancel := context.WithTimeout(context.Background(), activateStartTimeout)
defer cancel()
if err := s.runtime.Start(ctx, rec, s.hostForPlugin); err != nil {
_ = s.SetStatus(rec.ID, StatusError)
if setErr := s.setStatusAndDiagnostic(rec.ID, StatusError, err, true); setErr != nil {
s.log.Warn("plugins: could not persist activation failure",
zap.String("plugin_id", rec.ID), zap.Error(setErr))
}
return fmt.Errorf("plugins: start %q: %w", rec.ID, err)
}
}
@@ -1035,30 +1037,50 @@ func (s *Service) activate(rec *store.Record) error {
// both for the runtime spawn/stop and the status transition, and only want
// a single Refresh for the whole operation.
func (s *Service) SetStatus(id string, status Status) error {
return s.setStatusAndDiagnostic(id, status, nil, false)
}
// setStatusAndDiagnostic applies a lifecycle transition together with its
// persisted runtime diagnostic. allowSame is used only for repeated failure
// reports (for example, a failed retry while the record is already in error);
// public SetStatus keeps same-state transitions invalid.
func (s *Service) setStatusAndDiagnostic(id string, status Status, failure error, allowSame bool) error {
s.mu.Lock()
defer s.mu.Unlock()
rec, ok := s.registry.Get(id)
if !ok {
s.mu.Unlock()
return store.ErrNotFound
}
if !canTransition(rec.Status, status) {
s.mu.Unlock()
if rec.Status == status {
if !allowSame {
return &ErrInvalidTransition{ID: id, From: rec.Status, To: status}
}
} else if !canTransition(rec.Status, status) {
return &ErrInvalidTransition{ID: id, From: rec.Status, To: status}
}
updated, ok := s.registry.SetStatus(id, status)
lastError := rec.LastError
lastErrorAt := rec.LastErrorAt
if status == StatusActive {
lastError = ""
lastErrorAt = nil
}
if failure != nil {
lastError = normalizePluginError(failure)
now := time.Now().UTC()
lastErrorAt = &now
}
updated, ok := s.registry.SetRuntimeState(id, status, lastError, lastErrorAt)
if !ok {
s.mu.Unlock()
return store.ErrNotFound
}
if err := s.store.Save(updated); err != nil {
// Roll back the in-memory change so registry and disk stay in sync.
s.registry.SetStatus(id, rec.Status)
s.mu.Unlock()
s.registry.Add(rec)
return err
}
s.mu.Unlock()
return nil
}
@@ -1068,12 +1090,12 @@ func (s *Service) SetStatus(id string, status Status) error {
// health transitions. healthy=false drives active -> error; healthy=true
// drives error -> active plus a Deliverer.Flush (the buffered-event
// recovery replay). Restart count is persisted best-effort afterward.
func (s *Service) handleStatusChange(id string, healthy bool) {
func (s *Service) handleStatusChange(id string, healthy bool, reason error) {
newStatus := StatusError
if healthy {
newStatus = StatusActive
}
if err := s.SetStatus(id, newStatus); err != nil {
if err := s.setStatusAndDiagnostic(id, newStatus, reason, !healthy); err != nil {
s.log.Warn("plugins: health transition failed",
zap.String("plugin_id", id), zap.Bool("healthy", healthy), zap.Error(err))
} else {
@@ -1093,6 +1115,11 @@ func (s *Service) recordRestartCount(id string) {
if s.runtime == nil {
return
}
// Serialize this metadata write with lifecycle/diagnostic persistence so a
// restart callback cannot save a stale record over a concurrent Enable or
// recovery that just cleared LastError.
s.mu.Lock()
defer s.mu.Unlock()
updated, ok := s.registry.SetRestartCount(id, s.runtime.RestartCount(id))
if !ok {
return
@@ -1122,7 +1149,15 @@ func (s *Service) StartActivePlugins(ctx context.Context) {
if err := s.runtime.Start(ctx, rec, s.hostForPlugin); err != nil {
s.log.Warn("plugins: failed to spawn active plugin at boot",
zap.String("plugin_id", rec.ID), zap.Error(err))
_ = s.SetStatus(rec.ID, StatusError)
if setErr := s.setStatusAndDiagnostic(rec.ID, StatusError, err, true); setErr != nil {
s.log.Warn("plugins: could not persist boot activation failure",
zap.String("plugin_id", rec.ID), zap.Error(setErr))
} else {
// The deliverer was refreshed before boot activation began, so
// reconcile it after an active plugin fails to spawn. Otherwise
// its worker would continue treating the plugin as active.
s.notifyDeliverer()
}
}
}
}
+19 -27
View File
@@ -2,7 +2,6 @@ package plugins
import (
"context"
"errors"
"fmt"
"os"
"path/filepath"
@@ -274,6 +273,12 @@ func (s *Service) scanMissingInstalls() []string {
}
if _, err := os.Stat(rec.InstallPath); err == nil {
continue
} else if !os.IsNotExist(err) {
s.log.Warn("plugins: sync could not inspect install path",
zap.String("plugin_id", rec.ID),
zap.String("path", rec.InstallPath),
zap.Error(err))
continue
}
if s.runtime != nil && s.runtime.Running(rec.ID) {
s.runtime.Stop(rec.ID)
@@ -284,35 +289,22 @@ func (s *Service) scanMissingInstalls() []string {
return missing
}
// markMissing transitions id to StatusError via the normal FSM transition,
// falling back to a direct (transition-check-bypassing) status write if the
// transition is not a legal single-hop edge — e.g. the record is already
// StatusError, which SetStatus rejects as a same-status "transition".
// markMissing transitions id to StatusError and records the missing path as
// the current runtime diagnostic. Repeated scans replace the timestamp and
// diagnostic instead of bypassing the normal persistence path.
func (s *Service) markMissing(id string) {
err := s.SetStatus(id, StatusError)
rec, getErr := s.Get(id)
if getErr != nil {
s.log.Warn("plugins: sync failed to read missing install record",
zap.String("plugin_id", id), zap.Error(getErr))
return
}
err := s.setStatusAndDiagnostic(id, StatusError,
fmt.Errorf("plugin install path %q is missing", rec.InstallPath), true)
if err == nil {
s.notifyDeliverer()
return
}
var invalidErr *ErrInvalidTransition
if !errors.As(err, &invalidErr) {
s.log.Warn("plugins: sync failed to mark missing install as error",
zap.String("plugin_id", id), zap.Error(err))
return
}
if err := s.setStatusUnchecked(id, StatusError); err != nil {
s.log.Warn("plugins: sync failed to force missing install to error",
zap.String("plugin_id", id), zap.Error(err))
}
}
// setStatusUnchecked directly writes status onto id's record in both the
// registry and the store, bypassing SetStatus's FSM transition check. Only
// used by markMissing's fallback path above.
func (s *Service) setStatusUnchecked(id string, status Status) error {
updated, ok := s.registry.SetStatus(id, status)
if !ok {
return store.ErrNotFound
}
return s.store.Save(updated)
s.log.Warn("plugins: sync failed to mark missing install as error",
zap.String("plugin_id", id), zap.Error(err))
}
@@ -6,6 +6,7 @@ import (
"os"
"path/filepath"
goruntime "runtime"
"strings"
"testing"
"time"
)
@@ -217,6 +218,31 @@ func TestServiceSync_MissingInstallPathSetsErrorAndStopsRuntime(t *testing.T) {
if got.Status != StatusError {
t.Fatalf("Status after missing-install detection = %q, want %q", got.Status, StatusError)
}
if got.LastError == "" || got.LastErrorAt == nil || !strings.Contains(got.LastError, "install path") {
t.Fatalf("diagnostic after missing-install detection = (%q, %v), want install path/non-nil", got.LastError, got.LastErrorAt)
}
}
func TestServiceSync_StatErrorDoesNotMarkInstallMissing(t *testing.T) {
svc, _, _ := newTestService(t)
rec := installTestPlugin(t, svc, "kandev-plugin-stat-error")
rec.InstallPath = "\x00"
svc.registry.Add(rec)
result, err := svc.Sync(context.Background())
if err != nil {
t.Fatalf("Sync() unexpected error: %v", err)
}
if len(result.Missing) != 0 {
t.Fatalf("Sync().Missing = %v, want none for a non-NotExist stat error", result.Missing)
}
got, err := svc.Get("kandev-plugin-stat-error")
if err != nil {
t.Fatalf("Get(): %v", err)
}
if got.Status != StatusActive {
t.Fatalf("Status after stat error = %q, want %q", got.Status, StatusActive)
}
}
// TestServiceSync_MissingInstallAlreadyErrorIsIdempotent proves the "direct
+127 -4
View File
@@ -9,6 +9,7 @@ import (
"os"
"path/filepath"
goruntime "runtime"
"strings"
"sync"
"testing"
"time"
@@ -148,6 +149,12 @@ func (r *fakeRuntime) setStartErr(id string, err error) {
r.startErr[id] = err
}
func (r *fakeRuntime) clearStartErr(id string) {
r.mu.Lock()
defer r.mu.Unlock()
delete(r.startErr, id)
}
func (r *fakeRuntime) startCallCount(id string) int {
r.mu.Lock()
defer r.mu.Unlock()
@@ -489,6 +496,81 @@ func TestServiceDisabledCanReEnableAndRespawns(t *testing.T) {
}
}
func TestServiceEnableFailurePersistsDiagnosticAndSuccessfulRetryClearsIt(t *testing.T) {
svc, fsStore, rt := newTestService(t)
installTestPlugin(t, svc, "kandev-plugin-slack")
if err := svc.Disable("kandev-plugin-slack"); err != nil {
t.Fatalf("Disable(): %v", err)
}
rt.setStartErr("kandev-plugin-slack", errors.New("handshake failed: binary exited"))
if err := svc.Enable("kandev-plugin-slack"); err == nil {
t.Fatal("Enable() unexpectedly succeeded with a configured start failure")
}
failed, err := svc.Get("kandev-plugin-slack")
if err != nil {
t.Fatalf("Get() after failed Enable(): %v", err)
}
if failed.Status != StatusError {
t.Fatalf("Status after failed Enable() = %q, want %q", failed.Status, StatusError)
}
if failed.LastError == "" || !strings.Contains(failed.LastError, "handshake failed") {
t.Fatalf("LastError = %q, want the start failure", failed.LastError)
}
if failed.LastErrorAt == nil {
t.Fatal("LastErrorAt is nil after failed Enable()")
}
onDisk, err := fsStore.Get("kandev-plugin-slack")
if err != nil {
t.Fatalf("store.Get() after failed Enable(): %v", err)
}
if onDisk.LastError != failed.LastError || onDisk.LastErrorAt == nil {
t.Fatalf("persisted diagnostic = (%q, %v), want (%q, non-nil)", onDisk.LastError, onDisk.LastErrorAt, failed.LastError)
}
rt.clearStartErr("kandev-plugin-slack")
if err := svc.Enable("kandev-plugin-slack"); err != nil {
t.Fatalf("Enable() retry unexpected error: %v", err)
}
recovered, err := svc.Get("kandev-plugin-slack")
if err != nil {
t.Fatalf("Get() after successful retry: %v", err)
}
if recovered.Status != StatusActive {
t.Fatalf("Status after successful retry = %q, want %q", recovered.Status, StatusActive)
}
if recovered.LastError != "" || recovered.LastErrorAt != nil {
t.Fatalf("diagnostic after successful retry = (%q, %v), want empty/nil", recovered.LastError, recovered.LastErrorAt)
}
}
func TestServiceFailedRetryReplacesDiagnostic(t *testing.T) {
svc, _, rt := newTestService(t)
installTestPlugin(t, svc, "kandev-plugin-slack")
if err := svc.Disable("kandev-plugin-slack"); err != nil {
t.Fatalf("Disable(): %v", err)
}
rt.setStartErr("kandev-plugin-slack", errors.New("first handshake failure"))
if err := svc.Enable("kandev-plugin-slack"); err == nil {
t.Fatal("first Enable() unexpectedly succeeded")
}
first, _ := svc.Get("kandev-plugin-slack")
rt.setStartErr("kandev-plugin-slack", errors.New("second executable failure"))
if err := svc.Enable("kandev-plugin-slack"); err == nil {
t.Fatal("second Enable() unexpectedly succeeded")
}
second, _ := svc.Get("kandev-plugin-slack")
if second.Status != StatusError {
t.Fatalf("Status after failed retry = %q, want %q", second.Status, StatusError)
}
if second.LastError == first.LastError || !strings.Contains(second.LastError, "second executable failure") {
t.Fatalf("LastError after failed retry = %q, want replacement diagnostic", second.LastError)
}
}
// TestServiceEnable_ConcurrentCallsForSameID_OnlyOneActivationNoError proves
// Enable is guarded by a per-plugin lock: two near-simultaneous Enable calls
// for the same disabled plugin id must not both pass the StatusActive
@@ -588,12 +670,15 @@ func TestServiceHandleStatusChangeUnhealthyTransitionsToErrorAndRefreshesDeliver
deliverer := &fakeDeliverer{}
svc.SetDeliverer(deliverer)
svc.handleStatusChange("kandev-plugin-slack", false)
svc.handleStatusChange("kandev-plugin-slack", false, errors.New("ping timeout"))
got, _ := svc.Get("kandev-plugin-slack")
if got.Status != StatusError {
t.Fatalf("Status after unhealthy transition = %q, want %q", got.Status, StatusError)
}
if got.LastError == "" || !strings.Contains(got.LastError, "ping timeout") || got.LastErrorAt == nil {
t.Fatalf("diagnostic after unhealthy transition = (%q, %v), want ping timeout/non-nil", got.LastError, got.LastErrorAt)
}
if deliverer.refreshCount != 1 {
t.Fatalf("Refresh() call count = %d, want 1", deliverer.refreshCount)
}
@@ -605,16 +690,19 @@ func TestServiceHandleStatusChangeUnhealthyTransitionsToErrorAndRefreshesDeliver
func TestServiceHandleStatusChangeHealthyRecoversAndFlushesDeliverer(t *testing.T) {
svc, _, _ := newTestService(t)
installTestPlugin(t, svc, "kandev-plugin-slack")
svc.handleStatusChange("kandev-plugin-slack", false) // degrade first
svc.handleStatusChange("kandev-plugin-slack", false, errors.New("ping timeout")) // degrade first
deliverer := &fakeDeliverer{}
svc.SetDeliverer(deliverer)
svc.handleStatusChange("kandev-plugin-slack", true)
svc.handleStatusChange("kandev-plugin-slack", true, nil)
got, _ := svc.Get("kandev-plugin-slack")
if got.Status != StatusActive {
t.Fatalf("Status after recovery = %q, want %q", got.Status, StatusActive)
}
if got.LastError != "" || got.LastErrorAt != nil {
t.Fatalf("diagnostic after recovery = (%q, %v), want empty/nil", got.LastError, got.LastErrorAt)
}
if len(deliverer.flushedIDs) != 1 || deliverer.flushedIDs[0] != "kandev-plugin-slack" {
t.Fatalf("Flush() calls = %v, want [kandev-plugin-slack]", deliverer.flushedIDs)
}
@@ -627,7 +715,7 @@ func TestServiceHandleStatusChangePersistsRestartCountBestEffort(t *testing.T) {
rt.restartCounts["kandev-plugin-slack"] = 3
rt.mu.Unlock()
svc.handleStatusChange("kandev-plugin-slack", false)
svc.handleStatusChange("kandev-plugin-slack", false, errors.New("ping timeout"))
got, _ := svc.Get("kandev-plugin-slack")
if got.RestartCount != 3 {
@@ -780,3 +868,38 @@ func TestServiceStartActivePluginsSpawnsOnlyActiveManagedNotAlreadyRunning(t *te
t.Fatal("StartActivePlugins() did not spawn the active plugin")
}
}
func TestServiceStartActivePluginsFailurePersistsDiagnosticAndRefreshesDeliverer(t *testing.T) {
svc, dir, fsStore, _ := newTestServiceWithDir(t)
installTestPlugin(t, svc, "kandev-plugin-slack")
reg2 := NewRegistry()
if err := reg2.Load(fsStore); err != nil {
t.Fatalf("Load(): %v", err)
}
svc2 := NewService(fsStore, reg2, nil, testLogger(t))
svc2.SetPluginsDir(dir)
rt2 := newFakeRuntime()
rt2.setStartErr("kandev-plugin-slack", errors.New("boot handshake failed"))
svc2.SetRuntime(rt2)
deliverer := &fakeDeliverer{}
svc2.SetDeliverer(deliverer)
svc2.StartActivePlugins(context.Background())
got, err := svc2.Get("kandev-plugin-slack")
if err != nil {
t.Fatalf("Get() after boot failure: %v", err)
}
if got.Status != StatusError {
t.Fatalf("Status after boot failure = %q, want %q", got.Status, StatusError)
}
if !strings.Contains(got.LastError, "boot handshake failed") || got.LastErrorAt == nil {
t.Fatalf("diagnostic after boot failure = (%q, %v), want boot failure and timestamp", got.LastError, got.LastErrorAt)
}
// bootScan refreshes once at the end of its normal reconciliation, then
// the failed active spawn refreshes again so delivery sees StatusError.
if deliverer.refreshCount != 2 {
t.Fatalf("Refresh() calls after boot failure = %d, want 2", deliverer.refreshCount)
}
}
@@ -216,6 +216,51 @@ func TestFSStore_Save_PersistsUpdatedRecord(t *testing.T) {
}
}
func TestFSStore_Save_PersistsFailureDiagnostics(t *testing.T) {
dir := t.TempDir()
s := NewFSStore(dir)
failedAt := time.Date(2026, time.August, 2, 12, 34, 56, 0, time.UTC)
rec := testRecord("kandev-plugin-slack")
rec.Status = StatusError
rec.LastError = "plugins/runtime: connect to plugin: handshake failed"
rec.LastErrorAt = &failedAt
if err := s.Save(rec); err != nil {
t.Fatalf("Save() unexpected error: %v", err)
}
got, err := s.Get(rec.ID)
if err != nil {
t.Fatalf("Get() unexpected error: %v", err)
}
if got.LastError != rec.LastError {
t.Fatalf("Get().LastError = %q, want %q", got.LastError, rec.LastError)
}
if got.LastErrorAt == nil || !got.LastErrorAt.Equal(failedAt) {
t.Fatalf("Get().LastErrorAt = %v, want %v", got.LastErrorAt, failedAt)
}
}
func TestFSStore_Get_LegacyRecordDefaultsFailureDiagnostics(t *testing.T) {
dir := t.TempDir()
s := NewFSStore(dir)
rec := testRecord("kandev-plugin-slack")
if err := s.Save(rec); err != nil {
t.Fatalf("Save() unexpected error: %v", err)
}
got, err := s.Get(rec.ID)
if err != nil {
t.Fatalf("Get() unexpected error: %v", err)
}
if got.LastError != "" {
t.Fatalf("legacy record LastError = %q, want empty", got.LastError)
}
if got.LastErrorAt != nil {
t.Fatalf("legacy record LastErrorAt = %v, want nil", got.LastErrorAt)
}
}
func TestFSStore_SetConfigThenGetConfig_RoundTrips(t *testing.T) {
dir := t.TempDir()
s := NewFSStore(dir)
+5 -3
View File
@@ -48,9 +48,11 @@ type Record struct {
// and false forces it off — either override wins over the global default.
// Persisted so the choice survives restart and, unlike Status, is carried
// over verbatim when a plugin is upgraded in place (see Service.Install).
AutoUpdate *bool `yaml:"auto_update,omitempty" json:"auto_update,omitempty"`
InstalledAt time.Time `yaml:"installed_at" json:"installed_at"`
RestartCount int `yaml:"restart_count" json:"restart_count"`
AutoUpdate *bool `yaml:"auto_update,omitempty" json:"auto_update,omitempty"`
InstalledAt time.Time `yaml:"installed_at" json:"installed_at"`
RestartCount int `yaml:"restart_count" json:"restart_count"`
LastError string `yaml:"last_error,omitempty" json:"last_error,omitempty"`
LastErrorAt *time.Time `yaml:"last_error_at,omitempty" json:"last_error_at,omitempty"`
}
// Store is the persistence interface for plugin installations and their
+3 -3
View File
@@ -18,8 +18,8 @@ import (
// docs/specs/plugins/spec.md ("State machine"):
//
// registered -> active -> disabled -> uninstalled
// | |
// +-> error -+
// registered|active|disabled --failure--> error
// recovery --> active; Disable --> disabled
//
// Status is a type alias (not a distinct type) for store's Record.Status
// field, and the constants below are direct aliases of the string values
@@ -60,7 +60,7 @@ var allowedTransitions = map[Status][]Status{
StatusRegistered: {StatusActive, StatusError},
StatusActive: {StatusDisabled, StatusError},
StatusError: {StatusActive, StatusDisabled},
StatusDisabled: {StatusActive},
StatusDisabled: {StatusActive, StatusError},
}
// canTransition reports whether to is a legal single-hop transition from
@@ -15,6 +15,7 @@ import { PluginConfigForm } from "./plugin-config-form";
import { PluginManifestCard } from "./plugin-manifest-card";
import { PluginRepoLink } from "./plugin-repo-link";
import { PluginStatusBadge } from "./plugin-status-badge";
import { PluginErrorDiagnostic } from "./plugin-error-diagnostic";
import { UninstallPluginDialog } from "./uninstall-plugin-dialog";
import { usePluginActions } from "./use-plugin-actions";
import { usePluginConfigForm } from "./use-plugin-config-form";
@@ -114,6 +115,7 @@ function PluginDetailHeader({ plugin }: PluginDetailHeaderProps) {
{plugin.description && (
<p className="text-sm text-muted-foreground">{plugin.description}</p>
)}
<PluginErrorDiagnostic plugin={plugin} />
</div>
</div>
</div>
@@ -178,7 +180,8 @@ type PluginDangerZoneProps = {
function PluginDangerZone({ plugin, actions }: PluginDangerZoneProps) {
const busy = actions.busyId === plugin.id;
const canEnable = plugin.status === "disabled" || plugin.status === "registered";
const canEnable =
plugin.status === "disabled" || plugin.status === "registered" || plugin.status === "error";
const canDisable = plugin.status === "active" || plugin.status === "error";
return (
@@ -187,7 +190,7 @@ function PluginDangerZone({ plugin, actions }: PluginDangerZoneProps) {
<Button
variant="outline"
size="sm"
className="cursor-pointer"
className="cursor-pointer min-h-11 sm:min-h-0"
disabled={busy}
onClick={() => actions.handleEnable(plugin)}
>
@@ -198,7 +201,7 @@ function PluginDangerZone({ plugin, actions }: PluginDangerZoneProps) {
<Button
variant="outline"
size="sm"
className="cursor-pointer"
className="cursor-pointer min-h-11 sm:min-h-0"
disabled={busy}
onClick={() => actions.handleDisable(plugin)}
>
@@ -208,7 +211,7 @@ function PluginDangerZone({ plugin, actions }: PluginDangerZoneProps) {
<Button
variant="ghost"
size="sm"
className="cursor-pointer text-destructive hover:text-destructive"
className="cursor-pointer min-h-11 text-destructive hover:text-destructive sm:min-h-0"
disabled={busy}
onClick={() => actions.openUninstall(plugin)}
>
@@ -0,0 +1,46 @@
import { afterEach, describe, expect, it } from "vitest";
import { cleanup, render, screen } from "@testing-library/react";
import { PluginErrorDiagnostic } from "./plugin-error-diagnostic";
import type { PluginRecord } from "@/lib/types/plugins";
afterEach(() => cleanup());
function plugin(overrides: Partial<PluginRecord> = {}): PluginRecord {
return {
id: "acme",
api_version: 1,
version: "1.0.0",
display_name: "Acme",
description: "",
author: "acme",
categories: [],
capabilities: {},
status: "error",
install_path: "/p",
signed: true,
installed_at: "2026-01-01T00:00:00Z",
restart_count: 0,
...overrides,
};
}
describe("PluginErrorDiagnostic", () => {
it("renders the normalized message and machine-readable failure time", () => {
render(
<PluginErrorDiagnostic
plugin={plugin({
last_error: "plugins/runtime: handshake failed",
last_error_at: "2026-08-02T12:34:56Z",
})}
/>,
);
expect(screen.getByRole("alert").textContent).toContain("plugins/runtime: handshake failed");
expect(screen.getByRole("time").getAttribute("dateTime")).toBe("2026-08-02T12:34:56Z");
});
it("renders nothing when there is no persisted diagnostic", () => {
const { container } = render(<PluginErrorDiagnostic plugin={plugin()} />);
expect(container.innerHTML).toBe("");
});
});
@@ -0,0 +1,29 @@
import type { PluginRecord } from "@/lib/types/plugins";
/**
* Renders the host's persisted lifecycle diagnostic without inventing a
* second error state in the UI. The message is already normalized by the
* backend; the timestamp remains machine-readable for assistive tools and
* browser inspection.
*/
export function PluginErrorDiagnostic({ plugin }: { plugin: PluginRecord }) {
if (!plugin.last_error) return null;
return (
<div
role="alert"
data-testid={`plugin-error-${plugin.id}`}
className="rounded-md border border-destructive/30 bg-destructive/5 px-3 py-2 text-xs text-destructive [overflow-wrap:anywhere]"
>
<span>{plugin.last_error}</span>
{plugin.last_error_at && (
<time
dateTime={plugin.last_error_at}
className="mt-1 block text-[11px] text-destructive/70"
>
{new Date(plugin.last_error_at).toLocaleString()}
</time>
)}
</div>
);
}
@@ -112,9 +112,18 @@ function DeclarationList({
}
/** The manifest-declared fields only — runtime bookkeeping (status, install
* path, restart counter) is shown elsewhere and just noise here. */
* path, restart counter, and failure diagnostic) is shown elsewhere and just
* noise here. */
function manifestOnly(plugin: PluginRecord): Record<string, unknown> {
const { status: _s, install_path: _i, restart_count: _r, installed_at: _a, ...manifest } = plugin;
const {
status: _s,
install_path: _i,
restart_count: _r,
installed_at: _a,
last_error: _e,
last_error_at: _ea,
...manifest
} = plugin;
return manifest;
}
@@ -105,6 +105,23 @@ describe("PluginRow repo link", () => {
});
});
describe("PluginRow error recovery", () => {
it("shows the failure diagnostic and an Enable action for errored plugins", () => {
const onEnable = vi.fn();
const p = plugin({
status: "error",
last_error: "plugins/runtime: handshake failed",
last_error_at: "2026-08-02T12:34:56Z",
});
render(<PluginRow {...baseProps} plugin={p} onEnable={onEnable} />);
expect(screen.getByRole("alert").textContent).toContain("plugins/runtime: handshake failed");
const enable = screen.getByRole("button", { name: "Enable" });
fireEvent.click(enable);
expect(onEnable).toHaveBeenCalledWith(p);
});
});
describe("PluginRow auto-update toggle", () => {
it("reflects the global default when the plugin has no override", () => {
render(<PluginRow {...baseProps} plugin={plugin({ auto_update: null })} autoUpdateDefault />);
@@ -7,6 +7,7 @@ import { Switch } from "@kandev/ui/switch";
import Link from "@/components/routing/app-link";
import { PluginRepoLink } from "./plugin-repo-link";
import { PluginStatusBadge } from "./plugin-status-badge";
import { PluginErrorDiagnostic } from "./plugin-error-diagnostic";
import type { MarketplaceEntry, PluginRecord } from "@/lib/types/plugins";
type PluginRowProps = {
@@ -42,7 +43,8 @@ export function PluginRow({
onUpdate,
onSetAutoUpdate,
}: PluginRowProps) {
const canEnable = plugin.status === "disabled" || plugin.status === "registered";
const canEnable =
plugin.status === "disabled" || plugin.status === "registered" || plugin.status === "error";
const canDisable = plugin.status === "active" || plugin.status === "error";
return (
@@ -95,6 +97,7 @@ export function PluginRow({
{plugin.description && (
<div className="text-xs text-muted-foreground">{plugin.description}</div>
)}
<PluginErrorDiagnostic plugin={plugin} />
{plugin.categories.length > 0 && (
<div className="flex flex-wrap gap-1">
{plugin.categories.map((category) => (
@@ -205,7 +208,7 @@ function PluginRowActions({
variant="outline"
size="sm"
data-testid={`plugin-update-${plugin.id}`}
className="cursor-pointer gap-1"
className="cursor-pointer gap-1 min-h-11 sm:min-h-0"
disabled={busy}
onClick={() => onUpdate(update)}
>
@@ -217,7 +220,7 @@ function PluginRowActions({
<Button
variant="outline"
size="sm"
className="cursor-pointer"
className="cursor-pointer min-h-11 sm:min-h-0"
disabled={busy}
onClick={() => onEnable(plugin)}
>
@@ -228,7 +231,7 @@ function PluginRowActions({
<Button
variant="outline"
size="sm"
className="cursor-pointer"
className="cursor-pointer min-h-11 sm:min-h-0"
disabled={busy}
onClick={() => onDisable(plugin)}
>
@@ -238,7 +241,7 @@ function PluginRowActions({
<Button
variant="ghost"
size="sm"
className="cursor-pointer text-destructive hover:text-destructive"
className="cursor-pointer min-h-11 text-destructive hover:text-destructive sm:min-h-0"
disabled={busy}
onClick={() => onUninstall(plugin)}
>
@@ -1,7 +1,7 @@
import { renderHook, act, waitFor } from "@testing-library/react";
import { describe, it, expect, vi, beforeEach } from "vitest";
import type { ReactNode } from "react";
import { StateProvider } from "@/components/state-provider";
import { StateProvider, useAppStore } from "@/components/state-provider";
import type { PluginRecord } from "@/lib/types/plugins";
import type { InstallResult } from "@/lib/api/domains/plugins-api";
@@ -10,6 +10,7 @@ const unloadPlugin = vi.fn();
const installPluginFromUrl = vi.fn<() => Promise<InstallResult>>();
const installPluginUpload = vi.fn<() => Promise<InstallResult>>();
const enablePlugin = vi.fn(async () => ({ enabled: true }));
const getPlugin = vi.fn<() => Promise<PluginRecord>>();
vi.mock("@/lib/plugins/host", () => ({
loadPlugins: (...args: unknown[]) => loadPlugins(...(args as [])),
@@ -25,6 +26,7 @@ vi.mock("@/lib/api/domains/plugins-api", async () => {
installPluginFromUrl: (...args: unknown[]) => installPluginFromUrl(...(args as [])),
installPluginUpload: (...args: unknown[]) => installPluginUpload(...(args as [])),
enablePlugin: (...args: unknown[]) => enablePlugin(...(args as [])),
getPlugin: (...args: unknown[]) => getPlugin(...(args as [])),
};
});
@@ -60,6 +62,7 @@ beforeEach(() => {
installPluginFromUrl.mockReset();
installPluginUpload.mockReset();
enablePlugin.mockClear();
getPlugin.mockReset();
});
describe("usePluginActions — install/update", () => {
@@ -102,4 +105,53 @@ describe("usePluginActions — enable", () => {
const loadOrder = loadPlugins.mock.invocationCallOrder[0];
expect(unloadOrder).toBeLessThan(loadOrder);
});
it("clears a stale failure diagnostic after an errored plugin is enabled", async () => {
const plugin = activeRecord({
status: "error",
last_error: "plugins/runtime: handshake failed",
last_error_at: "2026-08-02T12:34:56Z",
});
const { result } = renderHook(
() => ({ actions: usePluginActions(), stored: useAppStore((s) => s.plugins.items[0]) }),
{ wrapper },
);
await act(async () => {
await result.current.actions.handleEnable(plugin);
});
await waitFor(() => expect(result.current.stored?.status).toBe("active"));
expect(result.current.stored?.last_error).toBeUndefined();
expect(result.current.stored?.last_error_at).toBeUndefined();
});
it("refreshes and upserts the replacement diagnostic after a failed enable", async () => {
const plugin = activeRecord({
status: "error",
last_error: "old handshake failure",
last_error_at: "2026-08-02T12:34:56Z",
});
const refreshed = activeRecord({
status: "error",
last_error: "new executable failure",
last_error_at: "2026-08-02T12:35:56Z",
});
enablePlugin.mockRejectedValueOnce(new Error("enable failed"));
getPlugin.mockResolvedValueOnce(refreshed);
const { result } = renderHook(
() => ({ actions: usePluginActions(), stored: useAppStore((s) => s.plugins.items[0]) }),
{ wrapper },
);
await act(async () => {
await result.current.actions.handleEnable(plugin);
});
await waitFor(() => expect(result.current.stored?.last_error).toBe("new executable failure"));
expect(result.current.stored?.last_error).not.toBe(plugin.last_error);
expect(getPlugin).toHaveBeenCalledWith(plugin.id, { cache: "no-store" });
});
});
@@ -8,6 +8,7 @@ import { useTheme } from "@/components/theme/app-theme";
import {
disablePlugin,
enablePlugin,
getPlugin,
installPluginFromUrl,
installPluginUpload,
listPlugins,
@@ -24,6 +25,9 @@ import type { PluginRecord, PluginStatus, SyncError } from "@/lib/types/plugins"
import type { AppState } from "@/lib/state/store";
function withStatus(plugin: PluginRecord, status: PluginStatus): PluginRecord {
if (status === "active") {
return { ...plugin, status, last_error: undefined, last_error_at: undefined };
}
return { ...plugin, status };
}
@@ -83,6 +87,13 @@ function useEnableDisableActions(upsertPlugin: (p: PluginRecord) => void) {
upsertPlugin(updated);
await loadIfActive(updated, storeApi, resolvedTheme, false);
} catch (err) {
try {
const refreshed = await getPlugin(plugin.id, { cache: "no-store" });
upsertPlugin(refreshed);
} catch {
// Preserve the original Enable failure toast if the diagnostic refresh
// itself cannot reach the backend.
}
toast.error(err instanceof Error ? err.message : `Failed to enable ${plugin.display_name}`);
} finally {
setBusyId(null);
@@ -8,6 +8,7 @@
* was always reachable through the settings menu sheet; this covers the
* plugin's *own* nav item.
*/
import fs from "node:fs";
import path from "node:path";
import type { Browser, BrowserContext, Page } from "@playwright/test";
import { expect, test } from "../../fixtures/test-base";
@@ -21,6 +22,15 @@ const PACKAGE_PATH = path.resolve(
"../../../../../apps/backend/.build/kandev-plugin-e2e-1.0.0.tar.gz",
);
function pluginExecutablePath(installPath: string): string {
const serverDir = path.join(installPath, "server");
const executable = fs
.readdirSync(serverDir, { withFileTypes: true })
.find((entry) => entry.isFile())?.name;
if (!executable) throw new Error(`no plugin executable found under ${serverDir}`);
return path.join(serverDir, executable);
}
/** A desktop-sized client against the same backend, for the parity screenshot. */
async function openDesktopClient(
browser: Browser,
@@ -112,4 +122,135 @@ test.describe("Mobile plugin navigation", () => {
capture.flush();
});
test("retries an errored plugin from the phone settings row", async ({
testPage,
apiClient,
backend,
}) => {
test.setTimeout(120_000);
await testPage.goto("/settings/plugins");
await testPage.getByTestId("install-plugin-trigger").click();
await testPage.getByTestId("install-plugin-tab-upload").click();
await testPage.getByTestId("install-plugin-file-input").setInputFiles(PACKAGE_PATH);
await testPage.getByTestId("install-plugin-upload-submit").click();
const pluginRow = testPage.getByTestId(`plugin-row-${PLUGIN_ID}`);
await expect(pluginRow).toBeVisible({ timeout: 30_000 });
const listResponse = await apiClient.rawRequest("GET", "/api/plugins");
const listBody = (await listResponse.json()) as {
plugins?: Array<{ id: string; install_path: string }>;
};
const installPath = listBody.plugins?.find((plugin) => plugin.id === PLUGIN_ID)?.install_path;
if (!installPath) throw new Error(`plugin ${PLUGIN_ID} install path was not returned`);
const executablePath = pluginExecutablePath(installPath);
const unavailablePath = `${executablePath}.unavailable`;
fs.renameSync(executablePath, unavailablePath);
try {
await backend.restart();
await expect
.poll(
async () => {
const response = await apiClient.rawRequest("GET", `/api/plugins/${PLUGIN_ID}`);
if (!response.ok) return "missing";
return ((await response.json()) as { status?: string }).status ?? "missing";
},
{ timeout: 20_000, intervals: [250, 500, 1000] },
)
.toBe("error");
await testPage.goto("/settings/plugins");
await expect(pluginRow.getByText("Error", { exact: true })).toBeVisible({ timeout: 15_000 });
await expect(pluginRow.getByTestId(`plugin-error-${PLUGIN_ID}`)).toBeVisible();
await expect(pluginRow.getByRole("button", { name: "Enable" })).toBeVisible();
expect(
(await pluginRow.getByRole("button", { name: "Enable" }).boundingBox())?.height,
).toBeGreaterThanOrEqual(44);
fs.renameSync(unavailablePath, executablePath);
await pluginRow.getByRole("button", { name: "Enable" }).click();
await expect(pluginRow.getByText("Active", { exact: true })).toBeVisible({ timeout: 15_000 });
await expect(pluginRow.getByTestId(`plugin-error-${PLUGIN_ID}`)).toHaveCount(0);
} finally {
if (fs.existsSync(unavailablePath) && !fs.existsSync(executablePath)) {
fs.renameSync(unavailablePath, executablePath);
}
}
});
test("wraps a long plugin diagnostic without phone horizontal overflow", async ({
testPage,
apiClient,
backend,
}) => {
test.setTimeout(120_000);
await testPage.goto("/settings/plugins");
await testPage.getByTestId("install-plugin-trigger").click();
await testPage.getByTestId("install-plugin-tab-upload").click();
await testPage.getByTestId("install-plugin-file-input").setInputFiles(PACKAGE_PATH);
await testPage.getByTestId("install-plugin-upload-submit").click();
const pluginRow = testPage.getByTestId(`plugin-row-${PLUGIN_ID}`);
await expect(pluginRow).toBeVisible({ timeout: 30_000 });
const listResponse = await apiClient.rawRequest("GET", "/api/plugins");
const listBody = (await listResponse.json()) as {
plugins?: Array<{ id: string; install_path: string }>;
};
const installPath = listBody.plugins?.find((plugin) => plugin.id === PLUGIN_ID)?.install_path;
if (!installPath) throw new Error(`plugin ${PLUGIN_ID} install path was not returned`);
const executablePath = pluginExecutablePath(installPath);
const unavailablePath = `${executablePath}.unavailable`;
fs.renameSync(executablePath, unavailablePath);
const longDiagnostic = "X".repeat(2048);
try {
await backend.restart();
await expect
.poll(
async () => {
const response = await apiClient.rawRequest("GET", `/api/plugins/${PLUGIN_ID}`);
if (!response.ok) return "missing";
return ((await response.json()) as { status?: string }).status ?? "missing";
},
{ timeout: 20_000, intervals: [250, 500, 1000] },
)
.toBe("error");
await testPage.route("**/api/plugins", async (route) => {
const response = await route.fetch();
const body = (await response.json()) as {
plugins?: Array<Record<string, unknown> & { id?: string }>;
};
body.plugins = body.plugins?.map((plugin) =>
plugin.id === PLUGIN_ID
? {
...plugin,
last_error: longDiagnostic,
last_error_at: "2026-08-02T12:34:56Z",
}
: plugin,
);
await route.fulfill({ response, json: body });
});
await testPage.reload();
const diagnostic = pluginRow.getByTestId(`plugin-error-${PLUGIN_ID}`);
await expect(diagnostic).toBeVisible({ timeout: 15_000 });
await expect(diagnostic.locator("span")).toHaveText(longDiagnostic);
const hasHorizontalOverflow = await testPage.evaluate(
() => document.documentElement.scrollWidth > document.documentElement.clientWidth,
);
expect(hasHorizontalOverflow).toBe(false);
} finally {
await testPage.unroute("**/api/plugins").catch(() => undefined);
if (fs.existsSync(unavailablePath) && !fs.existsSync(executablePath)) {
fs.renameSync(unavailablePath, executablePath);
}
}
});
});
+103
View File
@@ -88,6 +88,26 @@ async function uninstallViaApi(apiClient: ApiClient) {
await apiClient.rawRequest("DELETE", `/api/plugins/${PLUGIN_ID}`).catch(() => undefined);
}
async function installedPluginPath(apiClient: ApiClient): Promise<string> {
const response = await apiClient.rawRequest("GET", "/api/plugins");
if (!response.ok) throw new Error(`GET /api/plugins failed: ${response.status}`);
const body = (await response.json()) as {
plugins?: Array<{ id: string; install_path: string }>;
};
const record = body.plugins?.find((plugin) => plugin.id === PLUGIN_ID);
if (!record) throw new Error(`plugin ${PLUGIN_ID} was not returned by GET /api/plugins`);
return record.install_path;
}
function pluginExecutablePath(installPath: string): string {
const serverDir = path.join(installPath, "server");
const executable = fs
.readdirSync(serverDir, { withFileTypes: true })
.find((entry) => entry.isFile())?.name;
if (!executable) throw new Error(`no plugin executable found under ${serverDir}`);
return path.join(serverDir, executable);
}
/**
* Stages a filesystem sideload: extracts the same fixture package the
* upload tests use directly into `<pluginsDir>/<id>/<version>/`, with no
@@ -318,6 +338,89 @@ test.describe("Plugins — gRPC plugin install/load/live-update/uninstall", () =
await expect(globalToggle).toHaveAttribute("aria-checked", "false");
});
test("shows boot failure diagnostics and retries an errored plugin", async ({
testPage,
apiClient,
backend,
}) => {
test.setTimeout(120_000);
await openInstallDialog(testPage);
await uploadPackage(testPage, PACKAGE_PATH);
const pluginRow = testPage.getByTestId(`plugin-row-${PLUGIN_ID}`);
await expect(pluginRow).toBeVisible({ timeout: 15_000 });
await expect(pluginRow.getByText("Active", { exact: true })).toBeVisible();
const installPath = await installedPluginPath(apiClient);
const executablePath = pluginExecutablePath(installPath);
const unavailablePath = `${executablePath}.unavailable`;
fs.renameSync(executablePath, unavailablePath);
try {
await backend.restart();
await expect
.poll(
async () => {
const response = await apiClient.rawRequest("GET", `/api/plugins/${PLUGIN_ID}`);
if (!response.ok) return "missing";
return ((await response.json()) as { status?: string }).status ?? "missing";
},
{ timeout: 20_000, intervals: [250, 500, 1000] },
)
.toBe("error");
await testPage.goto("/settings/plugins");
await expect(pluginRow.getByText("Error", { exact: true })).toBeVisible({ timeout: 15_000 });
await expect(pluginRow.getByTestId(`plugin-error-${PLUGIN_ID}`)).toBeVisible();
await expect(pluginRow.getByRole("button", { name: "Enable" })).toBeVisible();
await pluginRow.getByTestId(`plugin-row-link-${PLUGIN_ID}`).click();
const detail = testPage.getByTestId(`plugin-detail-${PLUGIN_ID}`);
await expect(detail.getByTestId(`plugin-error-${PLUGIN_ID}`)).toBeVisible();
await expect(detail.getByRole("button", { name: "Enable" })).toBeVisible();
const firstDiagnostic = await detail
.getByTestId(`plugin-error-${PLUGIN_ID}`)
.locator("span")
.textContent();
if (!firstDiagnostic) throw new Error("initial plugin diagnostic was empty");
// The first retry still sees the missing executable. The hook must
// refetch the backend record even though the action remains failed.
await detail.getByRole("button", { name: "Enable" }).click();
await expect(detail.getByText("Error", { exact: true })).toBeVisible({ timeout: 15_000 });
// Change the failure mode without restoring the real binary. Linux
// reports an executable-format error, which must replace the original
// missing-file diagnostic in the detail view.
fs.writeFileSync(executablePath, "not a go-plugin executable\n");
if (process.platform !== "win32") fs.chmodSync(executablePath, 0o755);
await detail.getByRole("button", { name: "Enable" }).click();
await expect(detail.getByText("Error", { exact: true })).toBeVisible({ timeout: 15_000 });
await expect
.poll(
async () =>
(await detail.getByTestId(`plugin-error-${PLUGIN_ID}`).locator("span").textContent()) ??
"",
{ timeout: 15_000, intervals: [250, 500, 1000] },
)
.not.toBe(firstDiagnostic);
fs.rmSync(executablePath, { force: true });
fs.renameSync(unavailablePath, executablePath);
await detail.getByRole("button", { name: "Enable" }).click();
await expect(detail.getByText("Active", { exact: true })).toBeVisible({ timeout: 15_000 });
await expect(detail.getByTestId(`plugin-error-${PLUGIN_ID}`)).toHaveCount(0);
} finally {
if (fs.existsSync(unavailablePath) && fs.existsSync(executablePath)) {
fs.rmSync(executablePath, { force: true });
}
if (fs.existsSync(unavailablePath) && !fs.existsSync(executablePath)) {
fs.renameSync(unavailablePath, executablePath);
}
}
});
test("keybinding declared in the manifest opens a host.openModal demo modal", async ({
testPage,
}) => {
+4
View File
@@ -85,6 +85,10 @@ export interface PluginRecord {
installed_at: string;
/** Crash-restart attempts since install (health-check backoff counter). */
restart_count: number;
/** Host-managed, bounded single-line diagnostic for the last failure. */
last_error?: string | null;
/** UTC timestamp of the last host-managed failure diagnostic. */
last_error_at?: string | null;
last_health_check?: string | null;
/**
* Per-plugin auto-update override. Tri-state: `null`/`undefined` means
+168
View File
@@ -0,0 +1,168 @@
---
status: implemented
created: 2026-08-02
owner: cfl
spec: docs/specs/plugins/spec.md
---
# Plugin reliability repair
## Goal
Make plugin failures recoverable and diagnosable from Settings, and make the
event-buffer overflow signal useful under sustained failure. The repair covers
the installed-plugin symptoms found in `~/kandev/logs.txt`: persisted `error`
states with no reason, an unavailable retry action, and one warning per dropped
buffered event. This remediation also closes the review regressions around
concurrent recovery, diagnostic secrecy, authoritative retry refresh, and phone
layout safety.
## Evidence and root cause
- The backend state machine already permits `error -> active`, but both plugin
settings surfaces only offer **Enable** for `registered` and `disabled`.
- The plugin record persists status but no failure reason or timestamp, so a
restart leaves the operator with an opaque `Error` badge.
- Runtime health callbacks carry only healthy/unhealthy state, so health and
restart failures cannot be persisted with their originating error.
- The delivery ring buffer intentionally retains `error` plugins, but its
overflow path logs every dropped event. The current log contains a sustained
stream of Kandy overflow warnings.
- Restart exhaustion publishes `gaveUp` before its final synchronous status
callback returns. A concurrent `claimStart` holds `Manager.mu` while
`p.stop()` waits for that callback, which can re-enter `RestartCount` and wait
forever on `Manager.mu`. A barrier repro in the runtime package fails on the
current code.
- `handleEnable` catches a failed POST without refetching the backend record, so
the store keeps the previous `last_error` even though the service persisted a
replacement diagnostic.
- `normalizePluginError` only collapses whitespace and truncates raw
`err.Error()`, allowing plugin stdout and credential-like subprocess details to
become durable API data.
- Phone recovery controls use `min-h-9` (36px), and the diagnostic container has
no wrapping rule for an unbroken bounded token.
## Behavioral changes
1. Persist a bounded, single-line `last_error` (maximum 2048 bytes) and UTC
`last_error_at` for failed spawn, handshake, health, restart, and install-path
checks. Do not persist configuration values or secrets. Clear both fields
after a successful handshake and health check.
2. Include those nullable fields in the existing plugin list/detail API response.
Keep old records compatible when the fields are absent.
3. Treat `error` as a recoverable state in both desktop and mobile plugin
settings: show the diagnostic when present and expose **Enable** as the manual
retry action. A failed retry replaces the diagnostic; a successful retry clears
it. Boot must not automatically retry persisted `error` records.
4. Keep ring-buffer capacity and TTL unchanged. Emit at most one overflow warning
per plugin per 60 seconds and include the number of drops aggregated since the
previous warning.
5. A manual Enable concurrent with restart exhaustion must not deadlock. The
manager must not hold its registry mutex while waiting for a process to stop;
the final exhaustion callback and replacement start must both be allowed to
finish.
6. Failed Enable must refetch the authoritative plugin record and upsert it so a
replacement diagnostic is visible immediately. Recovery controls must be at
least 44px high on phone layouts, and diagnostic text must wrap without page
overflow.
7. Persisted diagnostics must redact PATs, bearer tokens, labeled secrets, and
host home paths before the 2048-byte bound is applied.
## Implementation design
### Backend failure diagnostics
- Extend `plugins/store.Record` with nullable `LastError` and `LastErrorAt`
fields, preserving YAML/JSON compatibility for existing records.
- Add registry/service helpers that atomically update status and diagnostic data,
roll back the in-memory record when persistence fails, and normalize errors to a
safe single-line bounded message. Redact credential-like values and home paths
before truncation.
- Change runtime status callbacks to carry the triggering error for unhealthy
transitions. Pass the ping/restart/exit cause through `runtime.Manager` and
`runtime.Process`; healthy callbacks clear the diagnostic.
- Record failures from initial activation, boot activation, config-change restart,
health transitions, restart exhaustion, and missing install paths. Preserve the
last diagnostic while disabled; clear it only on successful recovery.
- Keep `Manager.mu` out of process-stop waits and add a barrier regression test
covering final restart exhaustion against concurrent manual Enable.
- Add unit tests for serialization compatibility, truncation/newline handling,
every service transition, and runtime callback propagation.
### Delivery overflow diagnostics
- Add a per-worker overflow log accumulator with an injectable clock. The first
drop logs immediately; subsequent drops within the 60-second interval are
counted without logging; the first drop after the interval logs the aggregate
count and starts a new interval.
- Keep the existing oldest-drop behavior, TTL purge, recovery flush, delivery
ordering, and queue semantics unchanged.
- Add deterministic delivery tests asserting warning counts and `dropped_count`
aggregation for each plugin independently.
### Frontend recovery
- Allow `error` in `canEnable` for both `PluginRow` and `PluginDetail`.
- Render a compact, accessible diagnostic region from `last_error` in the row and
detail views without adding a new untranslated label; preserve the existing
status badge and responsive action layout.
- Clear diagnostic fields in the optimistic enable action so the UI does not show
stale failure text after a successful retry.
- On failed Enable, fetch `GET /api/plugins/{id}` with `no-store` and upsert the
returned record before showing the toast.
- Use `min-h-11 sm:min-h-0` for phone recovery controls and add wrapping to the
diagnostic region.
- Add component/action tests and a Playwright recovery flow. Exercise the same
settings path at a phone viewport to preserve mobile parity.
### Documentation
- Update `docs/public/plugins.md` with the error-state retry and diagnostic
behavior, and the aggregated overflow-warning policy.
- Run the public-doc validation scripts and the targeted backend/frontend tests.
## Task order
1. `task-01-backend-diagnostics.md` — record contract, runtime propagation, and
lifecycle persistence.
2. `task-02-overflow-logging.md` — per-plugin rate-limited aggregation and tests.
3. `task-03-frontend-recovery.md` — retry action, diagnostics, mobile coverage,
and frontend tests.
4. `task-04-docs-and-verification.md` — public docs and final targeted/full gates.
Tasks 01 and 02 are backend-only and can be implemented sequentially in this
session. Task 03 depends on the API fields from task 01. Task 04 follows the
behavioral implementation.
## Acceptance criteria
- A failed plugin record exposes a useful, bounded `last_error` and timestamp
after backend restart; a successful manual Enable clears both.
- An errored plugin can be retried from both desktop and mobile settings, and a
failed retry remains visibly errored with the new reason after an authoritative
refetch; the runtime deadlock regression test completes under a barrier.
- Persisted diagnostic tests prove PAT, bearer, labeled-secret, and home-path
redaction, while retaining useful non-secret error context.
- Phone recovery targets measure at least 44px and a long diagnostic does not
create horizontal overflow.
- Sustained ring-buffer overflow produces no more than one warning per plugin per
minute, with the suppressed-drop count reported on the next warning.
- Existing lifecycle, buffering, delivery, install, and plugin settings tests
remain green; public plugin docs describe the new operator behavior.
## Out of scope
- Automatic boot retry for persisted `error` records.
- Changing the 100-event/5-minute ring-buffer policy or making the buffer durable.
- Per-plugin event admission rate limiting, package signing, or unrelated plugin
startup failures.
## Validation
- Backend targeted tests for `internal/plugins/store`, `internal/plugins`,
`internal/plugins/runtime`, and `internal/plugins/delivery`.
- Web plugin component/action tests and typecheck/lint.
- Plugin Playwright tests, including a phone viewport recovery flow.
- `node --test scripts/validate-public-docs.test.mjs` and
`node scripts/validate-public-docs.mjs`.
- Required backend format/typecheck/test/lint gates before handoff.
@@ -0,0 +1,51 @@
---
status: done
---
# Task 01: Persist plugin failure diagnostics
## Objective
Carry failure causes through the supervised runtime and persist them on plugin
records so operators can recover an errored plugin with a useful explanation.
## Scope
- `apps/backend/internal/plugins/store/`
- `apps/backend/internal/plugins/registry.go`
- `apps/backend/internal/plugins/service.go`
- `apps/backend/internal/plugins/service_sync.go`
- `apps/backend/internal/plugins/runtime/`
- Related backend tests.
## Requirements
- Add nullable `last_error` and `last_error_at` record fields with backward-
compatible load/save behavior.
- Normalize stored diagnostics to one line and at most 2048 bytes; redact PATs,
bearer tokens, labeled passwords/tokens/secrets/API keys, and the host home
path before persistence. Do not store arbitrary plugin stdout or configuration.
- Propagate the triggering error through unhealthy runtime callbacks.
- Persist diagnostics for activation, boot/config restart, health failure,
restart exhaustion, and missing install paths.
- Clear diagnostics on successful activation/recovery and retain them across an
explicit disable.
- Preserve the existing `error -> active` transition and rollback guarantees.
- Never hold `Manager.mu` while waiting for `process.stop()`, so restart
exhaustion and concurrent manual Enable cannot wait on each other.
## Test-first acceptance
- Store round-trips fields and loads records written before the fields existed.
- Newline and overlong errors are normalized deterministically.
- PAT, bearer, labeled-secret, and home-path values are absent from persisted
diagnostics while safe host error context remains.
- Service tests cover failed enable, successful retry, failed retry replacement,
health failure, recovery, and missing install path.
- Runtime tests prove ping/crash causes reach the manager callback and healthy
recovery carries no error; a barrier-based concurrent Enable/restart-exhaustion
test proves both operations complete.
## Dependencies
None. This is the API contract consumed by task 03.
@@ -0,0 +1,35 @@
---
status: done
---
# Task 02: Aggregate ring-buffer overflow warnings
## Objective
Prevent sustained event-buffer pressure from producing one warning per dropped
event while retaining enough information to diagnose the pressure.
## Scope
- `apps/backend/internal/plugins/delivery/deliverer.go`
- Related delivery tests and logger observers.
## Requirements
- Keep the ring-buffer size, TTL, oldest-drop policy, and delivery order intact.
- Rate-limit overflow warnings independently per plugin to one per 60 seconds.
- Log the first drop immediately; aggregate suppressed drops and include
`dropped_count` in the next warning.
- Use an injectable clock so tests do not sleep.
## Test-first acceptance
- One drop emits one warning with `dropped_count=1`.
- More drops for the same plugin inside the interval emit no additional warning.
- The next drop after the interval emits one warning with the accumulated count.
- Two plugins have independent warning windows and counters.
- Existing ring-buffer recovery/TTL tests remain unchanged in behavior.
## Dependencies
None. This task may land before task 01.
@@ -0,0 +1,49 @@
---
status: done
---
# Task 03: Expose plugin recovery in Settings
## Objective
Make an errored plugin actionable and show the persisted backend diagnostic in
responsive settings UI.
## Scope
- `apps/web/lib/types/plugins.ts`
- `apps/web/components/settings/plugins/plugin-row.tsx`
- `apps/web/components/settings/plugins/plugin-detail.tsx`
- `apps/web/components/settings/plugins/plugin-manifest-card.tsx`
- `apps/web/components/settings/plugins/use-plugin-actions.ts`
- Component/action tests and plugin Playwright coverage.
## Requirements
- Add nullable `last_error` and `last_error_at` client fields.
- Include `error` in the enable-action states for row and detail views.
- Show a compact `role=alert` diagnostic only when a message is present.
- Clear stale diagnostics in local state after a successful enable.
- After a failed Enable, refetch `GET /api/plugins/{id}` with `no-store` and
upsert the authoritative record before showing the toast.
- Keep action controls visible and usable at phone widths; recovery targets are
at least 44px high on phone layouts. Add a phone recovery E2E assertion using
the existing native mobile settings path.
- Add wrapping/overflow protection for long diagnostics.
- Ensure runtime fields are not shown as authored manifest fields.
- Route any newly introduced user-facing copy through i18n.
## Test-first acceptance
- Error rows/details render Enable and the diagnostic.
- Enable action clears diagnostic fields on success and refetches/upserts a
replacement error on failure.
- Manifest raw-field display excludes both new runtime fields.
- Desktop and phone Playwright flows can retry a genuinely errored fixture
plugin, observe a changed diagnostic after consecutive failed retries, measure
recovery targets at least 44px, and prove a long diagnostic does not create
horizontal overflow.
## Dependencies
Task 01 API fields and lifecycle behavior.
@@ -0,0 +1,38 @@
---
status: done
---
# Task 04: Document and verify the repair
## Objective
Make the operator-facing behavior discoverable and run the repository checks
needed to hand off the repair confidently.
## Scope
- `docs/public/plugins.md`
- Any generated/translation updates required by task 03.
- Targeted and final verification commands.
## Requirements
- Document that Error is recoverable with Enable, where failure diagnostics are
shown, and that successful recovery clears them.
- Document that failed Enable refreshes the authoritative diagnostic, while
stored diagnostics redact credential-like values and home paths.
- Document that ring-buffer overflow warnings are aggregated per plugin rather
than emitted for every dropped event.
- Keep public terminology consistent with the existing plugin docs.
## Acceptance
- Public docs validation passes.
- Backend targeted tests, web tests, type checking, linting, and plugin E2E checks pass,
including the concurrent runtime barrier and phone overflow assertions.
- Backend formatting and required repository gates pass, or any environment
limitation is recorded explicitly in the handoff.
## Dependencies
Tasks 01–03.
+25 -5
View File
@@ -116,12 +116,16 @@ Either path runs the same pipeline:
3. Parse and validate `manifest.yaml` **before any code runs**: schema, `id`
pattern, the `categories` and UI-surface enums, and that
`runtime.executables` contains an entry for the host's OS/arch.
4. Extract to `~/.kandev/plugins/<id>/<version>/` and record the
installation.
4. Extract to `~/.kandev/plugins/<id>/<version>/` and record the installation
in `~/.kandev/plugins/<id>.yml`.
5. Spawn the platform-matched binary and complete the go-plugin handshake.
Status is `registered` while this is pending, `active` once the
handshake succeeds, or `error` if spawn/handshake fails (the operator can
retry via **Enable**).
retry via **Enable**). The record keeps a bounded, single-line diagnostic
and its failure timestamp so the reason is visible in Settings > Plugins.
Before persistence, credential-like values such as PATs, bearer tokens,
labeled secrets/API keys, and the host home path are redacted; plugin stdout
is not stored verbatim.
A successful install that failed to spawn returns HTTP 201 with a
`warning` field rather than failing outright — the package is installed,
@@ -169,10 +173,21 @@ binary an operator hasn't explicitly approved via install or Sync.
- **Disable** stops the subprocess. Config and state are preserved; no
events or webhooks are delivered while disabled.
- **Enable** respawns the subprocess and re-completes the handshake.
- **Enable** respawns the subprocess and re-completes the handshake. It is also
the manual recovery action for an `error` plugin; the Settings row and detail
page show the last failure diagnostic when one is available. A successful
retry clears the diagnostic, while a failed retry re-reads the plugin record
so the row/detail immediately shows the replacement reason and keeps the
plugin in `error`.
- **Uninstall** stops the subprocess and deletes the plugin's package,
registration record, and all persisted state — there is no grace period.
When a plugin is in `error`, its declared events remain buffered in the bounded
100-event/5-minute ring buffer. If that buffer overflows, kandev drops the
oldest event and emits at most one warning per plugin per minute, reporting the
number of drops accumulated since the previous warning instead of writing one
log line per dropped event.
## Per-plugin settings
A plugin can declare `config_schema` in its manifest, generating a settings
@@ -212,7 +227,7 @@ identically to an unsigned one; signing is not required in v1.
```
~/.kandev/plugins/
├── <id>.yml # registration record (status, install_path, signed, ...)
├── <id>.yml # registration record (signed, status, install_path, last_error, last_error_at, ...)
├── <id>.config.yml # operator-editable config (PATCH /api/plugins/{id})
└── <id>/
├── <version>/ # extracted package (InstallPath)
@@ -224,6 +239,11 @@ identically to an unsigned one; signing is not required in v1.
</details>
Each `<id>.yml` registration record stores the installed package metadata and
host-managed runtime fields, including `signed`, `last_error`, and
`last_error_at`. The diagnostic fields are empty until a runtime failure is
recorded and are cleared after successful recovery.
## Security posture
- **Auth is the spawn relationship.** Kandev spawns the plugin subprocess
+69 -10
View File
@@ -133,10 +133,23 @@ ui: # Native frontend plugin (see "Fr
status: "active"
version: "1.0.0"
install_path: "~/.kandev/plugins/kandev-plugin-slack/1.0.0"
signed: false
installed_at: "2026-04-26T10:00:00Z"
restart_count: 0
last_error: null
last_error_at: null
```
`last_error` and `last_error_at` are host-managed runtime diagnostics. When a
spawn, handshake, health check, restart, or install-path check moves a plugin to
`error`, kandev stores a single-line diagnostic (bounded to 2048 bytes) and the
UTC timestamp of the failure. Before persistence, kandev redacts credential-like
values (including PATs, bearer tokens, and labeled passwords, tokens, secrets,
or API keys) and replaces the host home path with `~`; arbitrary plugin stdout
and untrusted subprocess details are never persisted verbatim. The diagnostic
is bounded after redaction. A successful handshake and health check clears both
fields. Existing records without these fields load as null.
`capabilities.api_read` / `capabilities.api_write` gate the **Host data API** Host
RPCs (both reads and writes live now) — the vocabulary is a list of
resource names: `tasks`, `sessions`, `messages`, `workspaces`, `workflows`,
@@ -188,7 +201,9 @@ uploaded tarball:
6. Kandev spawns the platform-matched binary via `hashicorp/go-plugin`, completes the
handshake (§2 of GRPC-CONTRACT.md) — status `registered` while this is pending.
7. Handshake succeeds → status `active`. Handshake/spawn failure → status `error`
(restart retried with backoff; see "State machine").
(restart retried with backoff; see "State machine"). The failure diagnostic is
persisted on the record so the operator can see why activation failed and
retry with **Enable**.
Uninstall stops the subprocess and removes the record, all installed versions, and
plugin state (no 24-hour grace period in v1). `POST /api/plugins/register` is removed;
@@ -556,8 +571,9 @@ Mattermost-webapp model), not iframes. The full contract lives in
```
registered -> active -> disabled -> uninstalled
| |
+-> error -+
registered|active|disabled --failure--> error
error --successful Enable/health recovery--> active
error --Disable--> disabled
```
| State | Meaning |
@@ -572,7 +588,13 @@ Health monitoring: kandev's go-plugin client calls `Ping()` on the plugin every
seconds (injectable). 3 consecutive failures -> `error` + inbox item + restart attempt
with backoff. A subprocess crash (unexpected process exit) triggers an immediate
restart with backoff (max 5 attempts, then `error`). Next successful handshake/`Ping`
-> `active`, queued events delivered in order.
-> `active`, queued events delivered in order, and the persisted failure diagnostic
is cleared. An operator can manually enable a plugin in `error` to retry its spawn
and handshake; boot does not automatically retry a persisted `error` state. A
manual Enable racing the final restart-exhaustion callback must complete without
deadlock: the manager never waits for a stopping process while holding its
process registry lock, so the final callback and replacement start can both
complete.
## Permissions
@@ -604,8 +626,17 @@ restart with backoff (max 5 attempts, then `error`). Next successful handshake/`
- **3 consecutive `Ping` failures (90s), or crash with restart attempts exhausted
(max 5, backoff)**: status -> `error`. Events buffered (100, 5min TTL). Webhooks
return 503. Inbox item created.
- **Buffer overflows (>100 events or >5min)**: oldest events dropped and logged.
return 503. Inbox item created and the bounded failure diagnostic is persisted
with its timestamp.
- **Failed spawn/handshake, missing install path, or failed restart**: status ->
`error` with the bounded diagnostic persisted; the Plugins UI exposes **Enable**
as the manual retry action.
- **Diagnostic safety**: persisted failure text is normalized, credential/path
redacted, and bounded before it is returned by plugin list/detail APIs. Plugin
stdout is not a durable diagnostic channel.
- **Buffer overflows (>100 events or >5min)**: oldest events dropped. Kandev emits
at most one overflow warning per plugin per minute, aggregating the number of
dropped events since the previous warning.
- **Plugin returns a gRPC error (or times out) on `DeliverEvent`**: retry up to 3 times
with exponential backoff (5s, 15s, 45s). After exhaustion, event is logged as failed
and dropped.
@@ -617,8 +648,9 @@ restart with backoff (max 5 attempts, then `error`). Next successful handshake/`
## Persistence guarantees
- Plugin installation records (`id`, `version`, `install_path`, capabilities, status)
persist to disk under `~/.kandev/plugins/<id>/` and survive backend restarts.
- Plugin installation records (`id`, `version`, `install_path`, capabilities, status,
`signed`, `last_error`, `last_error_at`) persist to disk as
`~/.kandev/plugins/<id>.yml` and survive backend restarts.
- Extracted plugin packages persist at `~/.kandev/plugins/<id>/<version>/` until
uninstall.
- Plugin state in SQLite survives restarts.
@@ -666,7 +698,32 @@ restart with backoff (max 5 attempts, then `error`). Next successful handshake/`
plugin `error` while buffering events (up to 100 or 5 minutes), and creates an inbox
item "Plugin kandev-plugin-slack is unreachable". **WHEN** a subsequent restart
attempt succeeds and the handshake completes, **THEN** status returns to `active`
and buffered events are delivered in order.
and buffered events are delivered in order, with the persisted failure diagnostic
cleared.
- **GIVEN** a plugin in `error` with a persisted `last_error`, **WHEN** the operator
opens Settings > Plugins, **THEN** the row shows the diagnostic and an **Enable**
action. **WHEN** the operator clicks **Enable** and the plugin starts successfully,
**THEN** status returns to `active`, the diagnostic fields are cleared, and normal
delivery resumes. **WHEN** the retry fails, **THEN** status remains `error`, the
client refetches the authoritative plugin record, and the new diagnostic replaces
the previous one in the row and detail views.
- **GIVEN** restart exhaustion is entering its final unhealthy callback, **WHEN** an
operator concurrently clicks **Enable**, **THEN** the callback and Enable both
complete and the replacement process is either started or reports its own result;
neither operation waits forever on the other.
- **GIVEN** a persisted diagnostic containing an unbroken long token, **WHEN** the
operator opens the plugin row or detail view on a phone, **THEN** the diagnostic
wraps inside its container and the page has no horizontal overflow. Recovery
actions have a minimum 44 CSS-pixel phone target.
- **GIVEN** a plugin whose event ring buffer is full, **WHEN** additional events are
dropped, **THEN** kandev logs one warning for the first drop and suppresses further
warnings for that plugin for one minute while counting them. **WHEN** the next
warning is emitted, **THEN** it includes the aggregate number of drops since the
previous warning.
- **GIVEN** a plugin whose manifest declares `secrets: false`, **WHEN** the plugin
calls `Host.RevealSecret`, **THEN** kandev's server interceptor returns gRPC status
@@ -763,7 +820,9 @@ restart with backoff (max 5 attempts, then `error`). Next successful handshake/`
- **Hot reload.** Upgrading a plugin requires a new install (new version directory);
there is no in-place manifest or binary swap on a running process.
- **Multi-instance plugins.** Each plugin ID maps to exactly one supervised subprocess.
- **Rate limiting.** No per-plugin rate limits in v1. Misbehaving plugins can be disabled manually.
- **Event admission rate limiting.** No per-plugin rate limits in v1. Misbehaving
plugins can be disabled manually. Diagnostic log aggregation is still required
for bounded event-buffer overflow warnings.
- **Plugin database namespaces.** Plugins do not get their own SQLite schemas. KV state is sufficient for v1.
- **Broader write surface.** The Host data API writes cover task create/update
(conservative field mask) and sending a message to a task session. Deleting or