Resync session tools when backend health changes

docs/arch/10-virtual-mcp-architecture.md

@@ -791,6 +791,62 @@ connector wiring), `pkg/vmcp/aggregator/aggregator.go` and
 `runListChangedResync`, `resyncSessionTools`, `resyncSessionResources`,
 `resyncSessionPrompts`) with `Server.resyncBaseCtx` cancelled on `Stop`.
 
+### Health-driven tools resync (#5786, PR1: passthrough mode)
+
+The propagation above only fires when a connected backend itself emits a
+`list_changed` notification. A backend that flips
+unhealthy ⇄ healthy, or is added to / removed from the group, emits nothing —
+yet the advertised catalog changes, because capability aggregation
+health-filters the backend set (`filterHealthyBackends`). Before #5786,
+already-connected sessions kept serving the capability set snapshotted at
+registration until they reconnected; health transitions were a status-only
+signal (logged, reported to the CRD) with no data-path consumer.
+
+The health monitor (`pkg/vmcp/health`) now exposes a change callback:
+`Monitor.OnChange(fn ChangeListener)` (also on the `health.Reporter`
+interface). It fires when a backend's **advertisability** flips — a transition
+across the healthy/degraded ⇄ unhealthy/unknown/unauthenticated partition,
+detected at the `statusTracker.RecordSuccess`/`RecordFailure` transition
+points, mirroring `filterHealthyBackends`' inclusion rule — and when
+`UpdateBackends` adds or removes backends. Delivery is **debounced to the
+monitor's check interval** (leading edge immediate, further changes in the
+window coalesced into one trailing delivery carrying a monotonically
+increasing generation), so a flapping backend or a multi-backend partition
+cannot storm listeners. Listeners run on a dedicated goroutine, never on a
+health-check path, and `Monitor.Stop` waits for in-flight deliveries.
+
+`Serve` subscribes the transport layer
+(`pkg/vmcp/server/serve_health_resync.go`): the server keeps a registry of
+each live session's **KindTools resync worker** (the same
+`listChangedResyncWorker` the backend-notification path builds — identity and
+forwarded-header capture, the liveness guard, cache invalidation, replace
+semantics, and the SDK's automatic downstream `notifications/tools/list_changed`
+emission are all shared) and, on each delivery, triggers a tools resync for
+every registered session. Sessions register on successful registration and
+deregister on the termination paths the server observes; sessions that end
+without server involvement (TTL expiry, SDK-initiated DELETE) are pruned
+lazily when a triggered resync's liveness guard finds them gone.
+
+**Scope**: passthrough mode, tools only. When the optimizer is enabled the
+fan-out is a no-op — the advertised `find_tool`/`call_tool` meta-tools do not
+change on a health flip, and rebuilding the optimizer's per-session backing
+index for live sessions is the deferred optimizer-mode follow-up (PR2 of
+#5786). Resources/resource-templates/prompts re-derivation on health change is
+likewise not wired (a recovered backend's resources appear to new sessions,
+and to existing sessions on the backend's own `list_changed`). `UpdateBackends`
+notifies on membership changes only: a property change to an existing backend
+(URL/transport) restarts its health-check goroutine but does not notify —
+if the relocated backend serves a different tool set, existing sessions pick
+it up via the backend's own `list_changed` or on reconnect, matching the
+agreed membership-only scope.
+
+**Implementation**: `pkg/vmcp/health/change_notifier.go` (`ChangeListener`,
+debounce), `pkg/vmcp/health/monitor.go` (`OnChange`, fire points),
+`pkg/vmcp/health/status.go` (advertisability-transition detection),
+`pkg/vmcp/server/serve_health_resync.go` (`healthResyncRegistry`,
+`resyncSessionsOnBackendHealthChange`), subscription in
+`pkg/vmcp/server/serve.go`.
+
 ### Mid-call forwarding (elicitation / sampling / progress / logging)
 
 While a backend `tools/call` (or other request) is in flight, the backend may issue **server-initiated** requests and notifications back toward the client: elicitation, sampling, progress, and logging. vMCP forwards these mid-call in both directions through a per-call forwarder that bridges the backend connection to the originating client session, so a backend that needs user input (elicitation) or model completions (sampling), or that emits progress/log notifications, reaches the real client transparently. This is distinct from composite-tool elicitation (which the composer drives during a workflow); the mid-call forwarder handles the general request-scoped case for a single backend call.

pkg/vmcp/health/change_notifier.go

@@ -0,0 +1,138 @@
+// SPDX-FileCopyrightText: Copyright 2026 Stacklok, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package health
+
+import (
+	"slices"
+	"sync"
+	"time"
+)
+
+// ChangeListener receives backend-catalog change notifications from the
+// Monitor (see Monitor.OnChange). generation is a monotonically increasing
+// counter identifying the latest change folded into this notification;
+// intermediate generations are coalesced away by debouncing, so a listener
+// never sees the same generation twice. Generations are assigned in
+// dispatch order, but each delivery runs on its own goroutine: a listener
+// still processing one notification when the next window's delivery starts
+// can observe them overlapping or out of order. Listeners must therefore
+// treat a notification as "re-read current state", not as an ordered event
+// log (the tools-resync listener re-derives from the live health view, so
+// this is inherently safe there).
+//
+// Listeners are invoked on a dedicated goroutine — never on a health-check or
+// UpdateBackends call path — so they may safely call back into the Monitor's
+// read methods. Listeners MUST NOT call Monitor.Stop: Stop waits for in-flight
+// notifications to complete and would deadlock.
+type ChangeListener func(generation uint64)
+
+// changeNotifier coalesces backend-catalog change events and fans them out to
+// subscribed listeners, debounced to a fixed window: the first event after a
+// quiet period is delivered immediately, and further events inside the window
+// collapse into a single trailing notification carrying the latest
+// generation. This bounds delivery to at most two notifications per window
+// (one leading, one trailing) no matter how many backends change at once, so
+// a flapping backend cannot storm listeners.
+//
+// The zero value is not usable; construct with newChangeNotifier.
+type changeNotifier struct {
+	// window is the debounce window between deliveries.
+	window time.Duration
+
+	// fireWG tracks in-flight delivery goroutines so stop can wait for them
+	// (keeps shutdown deterministic and goroutine-leak-free).
+	fireWG sync.WaitGroup
+
+	mu           sync.Mutex
+	listeners    []ChangeListener
+	generation   uint64 // bumped on every notify()
+	deliveredGen uint64 // latest generation handed to listeners
+	lastFire     time.Time
+	timer        *time.Timer // non-nil while a trailing delivery is scheduled
+	stopped      bool
+}
+
+// newChangeNotifier returns a notifier that debounces deliveries to window.
+func newChangeNotifier(window time.Duration) *changeNotifier {
+	return &changeNotifier{window: window}
+}
+
+// subscribe registers fn for future deliveries — every delivery fired after
+// subscription, including a trailing delivery already scheduled for changes
+// that predate the subscription. fn never observes deliveries that fired
+// before it subscribed.
+func (n *changeNotifier) subscribe(fn ChangeListener) {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+	n.listeners = append(n.listeners, fn)
+}
+
+// notify records one catalog change and schedules its delivery: immediately
+// when the window has elapsed since the last delivery, otherwise via a single
+// trailing timer that folds every notify within the window into one delivery.
+func (n *changeNotifier) notify() {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+
+	n.generation++
+	if n.stopped || n.timer != nil {
+		// Stopped: never deliver. Timer pending: the scheduled trailing
+		// delivery picks up the generation bumped above.
+		return
+	}
+	if delay := n.window - time.Since(n.lastFire); delay > 0 {
+		n.timer = time.AfterFunc(delay, n.fireTrailing)
+		return
+	}
+	n.fireLocked()
+}
+
+// stop cancels any pending trailing delivery, suppresses all future delivery,
+// and waits for in-flight listener invocations to return. Idempotent.
+func (n *changeNotifier) stop() {
+	n.mu.Lock()
+	n.stopped = true
+	if n.timer != nil {
+		n.timer.Stop()
+		n.timer = nil
+	}
+	n.mu.Unlock()
+
+	n.fireWG.Wait()
+}
+
+// fireTrailing runs when the trailing timer expires: it delivers the current
+// generation unless it was already delivered (or the notifier stopped).
+func (n *changeNotifier) fireTrailing() {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+
+	n.timer = nil
+	if n.stopped || n.generation == n.deliveredGen {
+		return
+	}
+	n.fireLocked()
+}
+
+// fireLocked delivers the current generation to all listeners on a fresh
+// goroutine, so no delivery ever runs on a health-check or UpdateBackends
+// call path (listener code may take the Monitor's locks). Caller must hold
+// n.mu.
+func (n *changeNotifier) fireLocked() {
+	n.lastFire = time.Now()
+	n.deliveredGen = n.generation
+	if len(n.listeners) == 0 {
+		return
+	}
+
+	gen := n.generation
+	listeners := slices.Clone(n.listeners)
+	n.fireWG.Add(1)
+	go func() {
+		defer n.fireWG.Done()
+		for _, fn := range listeners {
+			fn(gen)
+		}
+	}()
+}

pkg/vmcp/health/change_notifier_test.go

@@ -0,0 +1,148 @@
+// SPDX-FileCopyrightText: Copyright 2026 Stacklok, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package health
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// notifierRecorder subscribes to a changeNotifier and records delivered
+// generations on a buffered channel so tests can assert delivery counts and
+// ordering without unbounded blocking.
+type notifierRecorder struct {
+	fires chan uint64
+}
+
+func newNotifierRecorder(t *testing.T, n *changeNotifier) *notifierRecorder {
+	t.Helper()
+	r := &notifierRecorder{fires: make(chan uint64, 64)}
+	n.subscribe(func(gen uint64) { r.fires <- gen })
+	return r
+}
+
+// next waits for one delivery, failing the test after a bounded timeout.
+func (r *notifierRecorder) next(t *testing.T) uint64 {
+	t.Helper()
+	select {
+	case gen := <-r.fires:
+		return gen
+	case <-time.After(2 * time.Second):
+		t.Fatal("timeout waiting for change notification")
+		return 0
+	}
+}
+
+// none asserts no delivery arrives within wait.
+func (r *notifierRecorder) none(t *testing.T, wait time.Duration) {
+	t.Helper()
+	select {
+	case gen := <-r.fires:
+		t.Fatalf("unexpected change notification with generation %d", gen)
+	case <-time.After(wait):
+	}
+}
+
+func TestChangeNotifier_DeliversLeadingEdgeImmediately(t *testing.T) {
+	t.Parallel()
+
+	n := newChangeNotifier(time.Hour) // window never elapses within the test
+	t.Cleanup(n.stop)
+	rec := newNotifierRecorder(t, n)
+
+	n.notify()
+
+	assert.Equal(t, uint64(1), rec.next(t),
+		"first change after a quiet period must be delivered immediately")
+	rec.none(t, 50*time.Millisecond)
+}
+
+func TestChangeNotifier_CoalescesBurstIntoOneTrailingDelivery(t *testing.T) {
+	t.Parallel()
+
+	// A wide window keeps the burst below reliably inside it even under CI
+	// scheduling delays; the trailing wait is event-driven, so the window
+	// bounds the test's runtime rather than padding it.
+	n := newChangeNotifier(500 * time.Millisecond)
+	t.Cleanup(n.stop)
+	rec := newNotifierRecorder(t, n)
+
+	// Leading delivery for the first change...
+	n.notify()
+	assert.Equal(t, uint64(1), rec.next(t))
+
+	// ...then a burst inside the window coalesces into exactly one trailing
+	// delivery carrying the latest generation.
+	n.notify()
+	n.notify()
+	n.notify()
+	assert.Equal(t, uint64(4), rec.next(t),
+		"burst must collapse into one trailing delivery with the latest generation")
+	rec.none(t, 200*time.Millisecond)
+}
+
+func TestChangeNotifier_NoTrailingDeliveryWithoutNewChange(t *testing.T) {
+	t.Parallel()
+
+	n := newChangeNotifier(50 * time.Millisecond)
+	t.Cleanup(n.stop)
+	rec := newNotifierRecorder(t, n)
+
+	n.notify()
+	assert.Equal(t, uint64(1), rec.next(t))
+
+	// A single change produces a single delivery: nothing trails it.
+	rec.none(t, 150*time.Millisecond)
+}
+
+func TestChangeNotifier_StopCancelsPendingTrailingDelivery(t *testing.T) {
+	t.Parallel()
+
+	// Wide window so the second notify reliably lands inside it (scheduling a
+	// trailing delivery) rather than firing leading-edge under CI delays.
+	n := newChangeNotifier(500 * time.Millisecond)
+	rec := newNotifierRecorder(t, n)
+
+	n.notify()
+	require.Equal(t, uint64(1), rec.next(t))
+	n.notify() // schedules a trailing delivery
+
+	n.stop()
+
+	rec.none(t, 600*time.Millisecond)
+}
+
+func TestChangeNotifier_StoppedNeverDelivers(t *testing.T) {
+	t.Parallel()
+
+	n := newChangeNotifier(time.Millisecond)
+	rec := newNotifierRecorder(t, n)
+
+	n.stop()
+	n.notify()
+
+	rec.none(t, 50*time.Millisecond)
+}
+
+func TestChangeNotifier_SubsequentWindowsDeliverAgain(t *testing.T) {
+	t.Parallel()
+
+	n := newChangeNotifier(30 * time.Millisecond)
+	t.Cleanup(n.stop)
+	rec := newNotifierRecorder(t, n)
+
+	n.notify()
+	first := rec.next(t)
+
+	// After the window has elapsed, the next change is again delivered
+	// immediately (leading edge), with a strictly greater generation.
+	time.Sleep(60 * time.Millisecond)
+	n.notify()
+	second := rec.next(t)
+
+	assert.Greater(t, second, first, "generations must be strictly increasing across deliveries")
+}