diff --git a/docs/explanation/decisions/012-server-classification-terminology.md b/docs/explanation/decisions/012-server-classification-terminology.md new file mode 100644 index 000000000..0e07b6e61 --- /dev/null +++ b/docs/explanation/decisions/012-server-classification-terminology.md @@ -0,0 +1,109 @@ +# ADR-012: Server Classification Terminology + +## Status + +Accepted + +## Context + +The muster codebase, documentation, plans, and issues use inconsistent terminology for downstream MCP servers. The same server category is variously referred to as: + +- "SSO server" +- "OAuth server" +- "OAuth-protected server" +- "non-OAuth server" +- "non-SSO server" + +This causes real problems: + +1. **Concept conflation**: "SSO server" is used as shorthand for any server with `StatusAuthRequired`, when SSO (token forwarding and token exchange) is just one of three authentication mechanisms. Manual OAuth login produces the exact same per-session architecture but is not SSO. + +2. **AI agent confusion**: AI agents (and humans) reading the code, plans, or issues see "SSO" and mentally narrow the scope to automatic authentication flows, missing that manual OAuth servers follow the identical code paths. + +3. **Misleading function names**: Functions like `handleNonOAuthCapabilityChanged` and `refreshNonOAuthCapabilities` suggest an OAuth-specific concern when the actual distinction is about the connection model (global client vs. per-session client). + +### The actual distinction + +There are exactly **two server categories** in the aggregator, defined by their **connection model**: + +| Property | Non-authenticated servers | Authenticated servers | +|---|---|---| +| Status | `StatusConnected` | `StatusAuthRequired` | +| Client model | Single persistent `ServerInfo.Client` shared by all sessions | Per-session clients pooled in `SessionConnectionPool` | +| Tool storage | `ServerInfo.Tools` (global) | `CapabilityStore` (keyed by `sessionID + serverName`) | +| Auth state | None | `SessionAuthStore` (keyed by `sessionID + serverName`) | +| Notification refresh | Via persistent client's `OnNotification` | Via pooled client's `OnNotification` (session-scoped) | + +**Within authenticated servers**, there are three auth mechanisms that do NOT affect the connection model: + +| Mechanism | Trigger | Configuration | +|---|---|---| +| Manual OAuth login | User runs `core_auth_login` | Default for servers returning 401 during registration | +| Token forwarding | Automatic at session creation | `auth.forwardToken: true` | +| Token exchange | Automatic at session creation | `auth.tokenExchange.enabled: true` | + +The difference between these mechanisms is *how the initial connection is established*. Once connected, the per-session client, capability store, auth store, and connection pool interactions are identical. + +## Decision + +### Standard terminology + +Use these terms consistently across all code, documentation, issues, and plans: + +| Term | Meaning | Use when... | +|---|---|---| +| **non-authenticated server** | A server with `StatusConnected` and a persistent global client | Describing the connection model | +| **authenticated server** | A server with `StatusAuthRequired` and per-session pooled clients | Describing the connection model | +| **token forwarding** | An auth mechanism where muster forwards its own ID token | Describing a specific auth mechanism | +| **token exchange** | An auth mechanism using RFC 8693 token exchange | Describing a specific auth mechanism | +| **manual OAuth login** | An auth mechanism where the user authenticates via browser | Describing a specific auth mechanism | +| **SSO** | Collective term for token forwarding and token exchange (automatic auth) | Describing automatic authentication, never as a server category | + +### Terms to avoid + +| Avoid | Use instead | Why | +|---|---|---| +| "SSO server" | "authenticated server" | SSO is an auth mechanism, not a server category | +| "OAuth server" | "authenticated server" | OAuth is an auth mechanism, not a server category | +| "non-OAuth server" | "non-authenticated server" | Defines the category by what it lacks, using the wrong axis | +| "OAuth-protected server" | "authenticated server" | Same issue as "OAuth server" | + +### Code naming conventions + +For new code, use names that reflect the connection model: + +- Functions operating on non-authenticated servers: `*Global*` or `*Persistent*` (e.g., `refreshGlobalCapabilities`) +- Functions operating on authenticated servers: `*Session*` or `*PerSession*` (e.g., `refreshSessionCapabilities` -- already correct) +- Functions operating on SSO specifically: `*SSO*` is fine when the code is actually specific to automatic auth (e.g., `initSSOForSession`, `ssoTracker`) + +### Existing code + +Existing function names like `refreshNonOAuthCapabilities`, `handleNonOAuthCapabilityChanged`, and `wirePoolNotificationCallback` are not renamed in this ADR. Renaming should happen opportunistically when those files are modified for other reasons, to avoid unnecessary churn. The terminology standard applies to: + +- All new code +- All new documentation and ADRs +- All new issues and plans +- Comments and doc strings when they are touched for other reasons + +## Consequences + +### Positive + +1. **Correct mental model**: The two-tier architecture is described by its actual axis (connection model), not by a subset of its authentication mechanisms. +2. **Reduced AI confusion**: AI agents reading "authenticated server" correctly understand this includes all three auth mechanisms, not just SSO. +3. **Clearer design discussions**: Layer 3 (polling) and Layer 4 (health tracking) can be discussed in terms of "iterate authenticated servers" instead of the misleading "iterate SSO servers". + +### Negative + +1. **Terminology gap**: Existing code, earlier ADRs (#011), and closed issues still use the old terms. This is acceptable as a living codebase; the standard applies going forward. + +### Neutral + +- The code constants `StatusConnected` and `StatusAuthRequired` remain unchanged -- they already align with this terminology. +- `ShouldUseTokenForwarding` and `ShouldUseTokenExchange` remain unchanged -- they correctly describe specific auth mechanisms, not server categories. + +## Related Decisions + +- [ADR-008: Unified Authentication Architecture](008-unified-authentication.md) -- defines `core_auth_login`/`core_auth_logout` as the auth interface +- [ADR-009: SSO Token Forwarding](009-sso-token-forwarding.md) -- defines token forwarding and exchange as SSO mechanisms +- [ADR-011: Session Connection Pool](011-session-connection-pool.md) -- establishes the three-store model for authenticated servers diff --git a/docs/explanation/decisions/README.md b/docs/explanation/decisions/README.md index 34577cdce..712172421 100644 --- a/docs/explanation/decisions/README.md +++ b/docs/explanation/decisions/README.md @@ -19,4 +19,5 @@ Each ADR follows this structure: - [ADR-008: Unified Authentication Architecture](008-unified-authentication.md) - [ADR-009: SSO Token Forwarding](009-sso-token-forwarding.md) - [ADR-010: Server-Side Meta-Tools Migration](010-server-side-meta-tools.md) -- [ADR-011: Session Connection Pool and Auth/Capability Separation](011-session-connection-pool.md) \ No newline at end of file +- [ADR-011: Session Connection Pool and Auth/Capability Separation](011-session-connection-pool.md) +- [ADR-012: Server Classification Terminology](012-server-classification-terminology.md) \ No newline at end of file diff --git a/internal/aggregator/capability_poller.go b/internal/aggregator/capability_poller.go new file mode 100644 index 000000000..998c2b8d8 --- /dev/null +++ b/internal/aggregator/capability_poller.go @@ -0,0 +1,111 @@ +package aggregator + +import ( + "time" + + "github.com/giantswarm/muster/pkg/logging" +) + +// DefaultCapabilityPollInterval is how often the poller re-fetches +// capabilities from all connected downstream MCP servers. This catches +// silent redeployments that don't trigger a CR change or a +// notifications/tools/list_changed notification. +const DefaultCapabilityPollInterval = 5 * time.Minute + +// runCapabilityPoller periodically re-fetches capabilities from every +// connected downstream MCP server and updates the registry/store when +// something changed. +// +// For non-authenticated servers (StatusConnected with a persistent +// ServerInfo.Client), it calls refreshNonOAuthCapabilities which +// deep-compares and updates ServerInfo.Tools/Resources/Prompts. +// +// For authenticated servers (StatusAuthRequired with per-session pooled +// clients), it iterates all active pooled sessions via +// SessionConnectionPool.GetSessionsForServer and calls +// refreshSessionCapabilities for each. +// +// All refresh calls are deduplicated via the shared notifRefreshGroup +// singleflight so that a poll overlapping with a notification-triggered +// refresh is coalesced. +func (a *AggregatorServer) runCapabilityPoller() { + defer a.wg.Done() + + interval := a.config.CapabilityPollInterval + if interval <= 0 { + interval = DefaultCapabilityPollInterval + } + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + logging.Info("Aggregator", "Capability poller started (interval=%v)", interval) + + for { + select { + case <-a.ctx.Done(): + logging.Debug("Aggregator", "Capability poller stopped") + return + case <-ticker.C: + a.pollAllServers() + } + } +} + +// pollAllServers iterates all registered servers and re-fetches +// capabilities, using the existing refresh methods from +// notification_subscriber.go. Each refresh is deduplicated via the +// shared notifRefreshGroup singleflight. +func (a *AggregatorServer) pollAllServers() { + servers := a.registry.GetAllServers() + if len(servers) == 0 { + return + } + + logging.Debug("Aggregator", "Capability poll: checking %d servers", len(servers)) + + for serverName, info := range servers { + switch { + case info.Status == StatusAuthRequired: + a.pollAuthServer(serverName) + + case info.IsConnected() && info.Client != nil: + a.pollNonAuthServer(serverName) + } + } +} + +// pollNonAuthServer re-fetches capabilities for a non-authenticated +// server using the shared singleflight key "notif-caps/". +func (a *AggregatorServer) pollNonAuthServer(serverName string) { + sfKey := "notif-caps/" + serverName + _, _, _ = a.notifRefreshGroup.Do(sfKey, func() (interface{}, error) { + a.refreshNonOAuthCapabilities(serverName) + return nil, nil + }) +} + +// pollAuthServer iterates all active pooled sessions for an authenticated +// server and re-fetches capabilities for each. Servers with no active +// pooled clients are skipped. +func (a *AggregatorServer) pollAuthServer(serverName string) { + if a.connPool == nil { + return + } + + sessions := a.connPool.GetSessionsForServer(serverName) + if len(sessions) == 0 { + logging.Debug("Aggregator", "Capability poll: no active sessions for auth server %s, skipping", serverName) + return + } + + ctx := a.refreshContext() + + for _, ps := range sessions { + sfKey := ps.SessionID + "/" + serverName + _, _, _ = a.notifRefreshGroup.Do(sfKey, func() (interface{}, error) { + a.refreshSessionCapabilities(ctx, serverName, ps.SessionID, ps.Client) + return nil, nil + }) + } +} diff --git a/internal/aggregator/capability_poller_test.go b/internal/aggregator/capability_poller_test.go new file mode 100644 index 000000000..3096edfa4 --- /dev/null +++ b/internal/aggregator/capability_poller_test.go @@ -0,0 +1,257 @@ +package aggregator + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/mark3labs/mcp-go/mcp" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestPollNonAuthServer_RefreshesCapabilities(t *testing.T) { + registry := NewServerRegistry("x") + a := &AggregatorServer{registry: registry} + + initial := []mcp.Tool{{Name: "old-tool", Description: "v1"}} + client := ¬ifMockClient{tools: initial} + + ctx := context.Background() + require.NoError(t, registry.Register(ctx, "srv", client, "")) + + updated := []mcp.Tool{ + {Name: "old-tool", Description: "v1"}, + {Name: "new-tool", Description: "v2"}, + } + client.setTools(updated) + + a.pollNonAuthServer("srv") + + info, _ := registry.GetServerInfo("srv") + info.mu.RLock() + assert.Len(t, info.Tools, 2) + info.mu.RUnlock() +} + +func TestPollNonAuthServer_NoChangeSkipsUpdate(t *testing.T) { + registry := NewServerRegistry("x") + a := &AggregatorServer{registry: registry} + + tools := []mcp.Tool{{Name: "tool", Description: "d"}} + client := ¬ifMockClient{tools: tools} + + ctx := context.Background() + require.NoError(t, registry.Register(ctx, "srv", client, "")) + + select { + case <-registry.GetUpdateChannel(): + default: + } + + a.pollNonAuthServer("srv") + + select { + case <-registry.GetUpdateChannel(): + t.Fatal("expected no update when tools haven't changed") + default: + } +} + +func TestPollNonAuthServer_SingleflightDedup(t *testing.T) { + registry := NewServerRegistry("x") + a := &AggregatorServer{registry: registry} + + tools := []mcp.Tool{{Name: "t1"}} + client := ¬ifMockClient{tools: tools} + + ctx := context.Background() + require.NoError(t, registry.Register(ctx, "srv", client, "")) + + baseCount := atomic.LoadInt32(&client.listToolsCalls) + + var wg sync.WaitGroup + for i := 0; i < 20; i++ { + wg.Add(1) + go func() { + defer wg.Done() + a.pollNonAuthServer("srv") + }() + } + wg.Wait() + + calls := atomic.LoadInt32(&client.listToolsCalls) - baseCount + assert.LessOrEqual(t, calls, int32(5), + "singleflight should deduplicate concurrent calls, got %d", calls) +} + +func TestPollAuthServer_IteratesActiveSessions(t *testing.T) { + capStore := NewInMemoryCapabilityStore(time.Hour) + pool := NewSessionConnectionPool(time.Hour) + defer pool.Stop() + + registry := NewServerRegistry("x") + a := &AggregatorServer{ + registry: registry, + capabilityStore: capStore, + connPool: pool, + } + + oldCaps := &Capabilities{Tools: []mcp.Tool{{Name: "old"}}} + require.NoError(t, capStore.Set(context.Background(), "sess-1", "auth-srv", oldCaps)) + + updated := []mcp.Tool{{Name: "old"}, {Name: "new"}} + client := ¬ifMockClient{tools: updated} + pool.Put("sess-1", "auth-srv", client) + + a.pollAuthServer("auth-srv") + + caps, err := capStore.Get(context.Background(), "sess-1", "auth-srv") + require.NoError(t, err) + require.NotNil(t, caps) + assert.Len(t, caps.Tools, 2) +} + +func TestPollAuthServer_SkipsWhenNoActiveSessions(t *testing.T) { + pool := NewSessionConnectionPool(time.Hour) + defer pool.Stop() + + a := &AggregatorServer{ + registry: NewServerRegistry("x"), + connPool: pool, + } + + a.pollAuthServer("no-sessions-server") +} + +func TestPollAuthServer_NilConnPool(t *testing.T) { + a := &AggregatorServer{ + registry: NewServerRegistry("x"), + connPool: nil, + } + + a.pollAuthServer("srv") +} + +func TestPollAllServers_ConnectedAndAuthRequired(t *testing.T) { + capStore := NewInMemoryCapabilityStore(time.Hour) + pool := NewSessionConnectionPool(time.Hour) + defer pool.Stop() + + registry := NewServerRegistry("x") + a := &AggregatorServer{ + registry: registry, + capabilityStore: capStore, + connPool: pool, + } + + nonAuthClient := ¬ifMockClient{tools: []mcp.Tool{{Name: "t1"}}} + require.NoError(t, registry.Register(context.Background(), "connected-srv", nonAuthClient, "")) + + _ = registry.RegisterPendingAuth("auth-srv", "http://localhost:9999", "", nil) + + authClient := ¬ifMockClient{tools: []mcp.Tool{{Name: "auth-t1"}}} + pool.Put("sess-1", "auth-srv", authClient) + require.NoError(t, capStore.Set(context.Background(), "sess-1", "auth-srv", + &Capabilities{Tools: []mcp.Tool{{Name: "auth-t1"}}})) + + baseNonAuth := atomic.LoadInt32(&nonAuthClient.listToolsCalls) + + a.pollAllServers() + + assert.Equal(t, baseNonAuth+1, atomic.LoadInt32(&nonAuthClient.listToolsCalls), + "pollAllServers should trigger exactly one ListTools call for the non-auth server") +} + +func TestRunCapabilityPoller_StopsOnContextCancel(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + + registry := NewServerRegistry("x") + client := ¬ifMockClient{tools: []mcp.Tool{{Name: "t1"}}} + require.NoError(t, registry.Register(context.Background(), "srv", client, "")) + + baseCount := atomic.LoadInt32(&client.listToolsCalls) + + a := &AggregatorServer{ + ctx: ctx, + registry: registry, + config: AggregatorConfig{CapabilityPollInterval: 50 * time.Millisecond}, + connPool: NewSessionConnectionPool(time.Hour), + } + defer a.connPool.Stop() + + a.wg.Add(1) + go a.runCapabilityPoller() + + require.Eventually(t, func() bool { + return atomic.LoadInt32(&client.listToolsCalls)-baseCount >= 1 + }, 5*time.Second, 10*time.Millisecond, + "poller should have polled at least once before cancel") + + cancel() + + done := make(chan struct{}) + go func() { + a.wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("poller did not stop within timeout after context cancel") + } +} + +func TestRunCapabilityPoller_PollsOnTick(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + registry := NewServerRegistry("x") + client := ¬ifMockClient{tools: []mcp.Tool{{Name: "t1"}}} + require.NoError(t, registry.Register(context.Background(), "srv", client, "")) + + baseCount := atomic.LoadInt32(&client.listToolsCalls) + + a := &AggregatorServer{ + ctx: ctx, + registry: registry, + config: AggregatorConfig{CapabilityPollInterval: 100 * time.Millisecond}, + connPool: NewSessionConnectionPool(time.Hour), + } + defer a.connPool.Stop() + + a.wg.Add(1) + go a.runCapabilityPoller() + + require.Eventually(t, func() bool { + return atomic.LoadInt32(&client.listToolsCalls)-baseCount >= 2 + }, 5*time.Second, 50*time.Millisecond, + "poller should have re-fetched tools at least twice") + + cancel() + a.wg.Wait() +} + +func TestDefaultCapabilityPollInterval(t *testing.T) { + assert.Equal(t, 5*time.Minute, DefaultCapabilityPollInterval) +} + +func TestRunCapabilityPoller_UsesDefaultInterval(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + a := &AggregatorServer{ + ctx: ctx, + registry: NewServerRegistry("x"), + config: AggregatorConfig{}, + connPool: NewSessionConnectionPool(time.Hour), + } + defer a.connPool.Stop() + + a.wg.Add(1) + go a.runCapabilityPoller() + a.wg.Wait() +} diff --git a/internal/aggregator/server.go b/internal/aggregator/server.go index b85d92d7d..18575b05a 100644 --- a/internal/aggregator/server.go +++ b/internal/aggregator/server.go @@ -393,6 +393,10 @@ func (a *AggregatorServer) Start(ctx context.Context) error { a.wg.Add(1) go a.runSSOTrackerCleanup() + // Start periodic capability polling for silent redeployment detection + a.wg.Add(1) + go a.runCapabilityPoller() + // Subscribe to tool update events from workflow and other managers // This ensures the aggregator stays synchronized with core muster components logging.Info("Aggregator", "Subscribing to tool update events...") diff --git a/internal/aggregator/session_connection_pool.go b/internal/aggregator/session_connection_pool.go index b44ec5c03..7bac1a226 100644 --- a/internal/aggregator/session_connection_pool.go +++ b/internal/aggregator/session_connection_pool.go @@ -342,6 +342,33 @@ func (p *SessionConnectionPool) Len() int { return len(p.pool) } +// PooledSession pairs a session ID with a live MCP client for iteration. +// Used by the capability poller to re-fetch capabilities for authenticated +// servers across all active sessions. +type PooledSession struct { + SessionID string + Client MCPClient +} + +// GetSessionsForServer returns a snapshot of all currently pooled sessions +// for the given server name. No lock is held after return; callers must +// handle the case where a session was evicted between snapshot and use. +func (p *SessionConnectionPool) GetSessionsForServer(serverName string) []PooledSession { + p.mu.RLock() + defer p.mu.RUnlock() + + var sessions []PooledSession + for key, entry := range p.pool { + if key.ServerName == serverName && entry.Client != nil { + sessions = append(sessions, PooledSession{ + SessionID: key.SessionID, + Client: entry.Client, + }) + } + } + return sessions +} + // evictedPoolEntry pairs a poolKey with a snapshot of the poolEntry that was // removed. Used by evictIdle to defer Close calls outside the write lock. type evictedPoolEntry struct { diff --git a/internal/aggregator/session_connection_pool_test.go b/internal/aggregator/session_connection_pool_test.go index a7becf82e..285178d19 100644 --- a/internal/aggregator/session_connection_pool_test.go +++ b/internal/aggregator/session_connection_pool_test.go @@ -521,6 +521,48 @@ func TestSessionConnectionPool_SetNotificationCallback_ReplacesCallback(t *testi assert.Equal(t, int32(1), secondCount.Load(), "new callback should be invoked") } +func TestSessionConnectionPool_GetSessionsForServer_ReturnsMatching(t *testing.T) { + pool := newTestPool() + defer pool.Stop() + + c1 := &poolTestClient{} + c2 := &poolTestClient{} + c3 := &poolTestClient{} + + pool.Put("s1", "srv-a", c1) + pool.Put("s2", "srv-a", c2) + pool.Put("s1", "srv-b", c3) + + sessions := pool.GetSessionsForServer("srv-a") + assert.Len(t, sessions, 2, "should return 2 sessions for srv-a") + + ids := make(map[string]bool) + for _, ps := range sessions { + ids[ps.SessionID] = true + assert.NotNil(t, ps.Client) + } + assert.True(t, ids["s1"]) + assert.True(t, ids["s2"]) +} + +func TestSessionConnectionPool_GetSessionsForServer_EmptyPool(t *testing.T) { + pool := newTestPool() + defer pool.Stop() + + sessions := pool.GetSessionsForServer("nonexistent") + assert.Empty(t, sessions) +} + +func TestSessionConnectionPool_GetSessionsForServer_NoMatchingServer(t *testing.T) { + pool := newTestPool() + defer pool.Stop() + + pool.Put("s1", "srv-a", &poolTestClient{}) + + sessions := pool.GetSessionsForServer("srv-b") + assert.Empty(t, sessions) +} + func TestSessionConnectionPool_EvictIdleMixedEntries(t *testing.T) { maxAge := 100 * time.Millisecond pool := NewSessionConnectionPool(maxAge) diff --git a/internal/aggregator/types.go b/internal/aggregator/types.go index a25c43f10..dfc15a53b 100644 --- a/internal/aggregator/types.go +++ b/internal/aggregator/types.go @@ -203,6 +203,10 @@ type AggregatorConfig struct { // Debug enables debug logging Debug bool + + // CapabilityPollInterval is the interval between periodic capability + // re-fetches from downstream servers. Zero uses DefaultCapabilityPollInterval. + CapabilityPollInterval time.Duration } // OAuthServerConfig holds OAuth server configuration for protecting the Muster Server. diff --git a/internal/app/services.go b/internal/app/services.go index 4ef9bb9f4..96bf57334 100644 --- a/internal/app/services.go +++ b/internal/app/services.go @@ -2,6 +2,7 @@ package app import ( "fmt" + "time" mcpserverPkg "github.com/giantswarm/muster/internal/mcpserver" aggregatorService "github.com/giantswarm/muster/internal/services/aggregator" @@ -240,15 +241,23 @@ func InitializeServices(cfg *Config) (*Services, error) { effectiveClientID := mergedOAuthMCPClientConfig.GetEffectiveClientID() // Convert config types + var capPollInterval time.Duration + if raw := cfg.MusterConfig.Aggregator.CapabilityPollInterval; raw != "" { + if d, err := time.ParseDuration(raw); err == nil { + capPollInterval = d + } + } + aggConfig := aggregator.AggregatorConfig{ - Port: cfg.MusterConfig.Aggregator.Port, - Host: cfg.MusterConfig.Aggregator.Host, - Transport: cfg.MusterConfig.Aggregator.Transport, - MusterPrefix: cfg.MusterConfig.Aggregator.MusterPrefix, - Version: cfg.Version, - Yolo: cfg.Yolo, - ConfigDir: cfg.ConfigPath, - Debug: cfg.Debug, + Port: cfg.MusterConfig.Aggregator.Port, + Host: cfg.MusterConfig.Aggregator.Host, + Transport: cfg.MusterConfig.Aggregator.Transport, + MusterPrefix: cfg.MusterConfig.Aggregator.MusterPrefix, + Version: cfg.Version, + Yolo: cfg.Yolo, + ConfigDir: cfg.ConfigPath, + Debug: cfg.Debug, + CapabilityPollInterval: capPollInterval, OAuth: aggregator.OAuthProxyConfig{ Enabled: oauthMCPClientEnabled, PublicURL: oauthPublicURL, diff --git a/internal/config/types.go b/internal/config/types.go index de8a8fe5a..d9b658ec1 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -37,6 +37,11 @@ type AggregatorConfig struct { Transport string `yaml:"transport,omitempty"` // Transport to use (default: streamable-http) MusterPrefix string `yaml:"musterPrefix,omitempty"` // Pre-prefix for all tools (default: "x") + // CapabilityPollInterval is the interval between periodic capability + // re-fetches from downstream servers (e.g. "5m", "30s"). Zero or empty + // uses the default (5 minutes). + CapabilityPollInterval string `yaml:"capabilityPollInterval,omitempty"` + // OAuth contains all OAuth-related configuration with explicit mcpClient/server roles. // - oauth.mcpClient: muster as OAuth client/proxy for authenticating TO remote MCP servers // - oauth.server: muster as OAuth resource server for protecting ITSELF diff --git a/internal/testing/mock/http_server.go b/internal/testing/mock/http_server.go index 26eba1ad5..b5e4f9f66 100644 --- a/internal/testing/mock/http_server.go +++ b/internal/testing/mock/http_server.go @@ -252,6 +252,12 @@ func (s *HTTPServer) RemoveDynamicTool(toolName string) { s.mockServer.RemoveDynamicTool(toolName) } +// AddDynamicToolSilently adds a tool without sending notifications. +// Used to simulate silent server redeployments for capability poller testing. +func (s *HTTPServer) AddDynamicToolSilently(toolConfig ToolConfig) { + s.mockServer.AddDynamicToolSilently(toolConfig) +} + // GetError returns any error that occurred during server operation func (s *HTTPServer) GetError() error { s.mu.RLock() diff --git a/internal/testing/mock/server.go b/internal/testing/mock/server.go index 414fa3afe..deb5bd579 100644 --- a/internal/testing/mock/server.go +++ b/internal/testing/mock/server.go @@ -6,8 +6,10 @@ import ( "fmt" "os" "path/filepath" + "reflect" "strings" "sync" + "unsafe" "github.com/giantswarm/muster/internal/template" @@ -160,6 +162,61 @@ func (s *Server) RemoveDynamicTool(toolName string) { } } +// AddDynamicToolSilently adds a tool to the server so that ListTools +// returns it, but WITHOUT sending a notifications/tools/list_changed +// notification. This simulates a silent server redeployment where the +// tool list changes without the connected client being notified. +// +// It bypasses the mcp-go library's AddTool (which always sends a +// notification) by writing directly to the internal tools map via +// reflect + unsafe. This is acceptable because this code is used +// exclusively in test scenarios. +func (s *Server) AddDynamicToolSilently(toolConfig ToolConfig) { + handler := NewToolHandler(toolConfig, s.templateEngine, s.debug) + + s.mu.Lock() + s.toolHandlers[toolConfig.Name] = handler + s.mu.Unlock() + + tool := mcp.NewTool(toolConfig.Name, mcp.WithDescription(toolConfig.Description)) + st := server.ServerTool{ + Tool: tool, + Handler: s.createToolHandler(toolConfig.Name), + } + + injectToolSilently(s.mcpServer, toolConfig.Name, st) + + if s.debug { + fmt.Fprintf(os.Stderr, "Silently added tool '%s' to mock server '%s' (no notification)\n", toolConfig.Name, s.name) + } +} + +// injectToolSilently writes a ServerTool directly into the MCPServer's +// unexported tools map, acquiring the toolsMu lock but skipping the +// notification that AddTool/AddTools normally sends. This is test-only +// code that uses reflect + unsafe to access unexported struct fields. +func injectToolSilently(srv *server.MCPServer, name string, st server.ServerTool) { + v := reflect.ValueOf(srv).Elem() + + muField := v.FieldByName("toolsMu") + if !muField.IsValid() { + panic("injectToolSilently: mcp-go MCPServer no longer has field 'toolsMu' -- check for upstream changes") + } + //nolint:gosec // Test-only code: accessing unexported field via unsafe. + mu := (*sync.RWMutex)(unsafe.Pointer(muField.UnsafeAddr())) + + toolsField := v.FieldByName("tools") + if !toolsField.IsValid() { + panic("injectToolSilently: mcp-go MCPServer no longer has field 'tools' -- check for upstream changes") + } + //nolint:gosec // Test-only code: accessing unexported field via unsafe. + toolsPtr := (*map[string]server.ServerTool)(unsafe.Pointer(toolsField.UnsafeAddr())) + + mu.Lock() + (*toolsPtr)[name] = st + mu.Unlock() +} + // Start starts the mock MCP server using stdio transport func (s *Server) Start(ctx context.Context) error { if s.debug { diff --git a/internal/testing/muster_manager.go b/internal/testing/muster_manager.go index 34f6307dd..bf62b4b47 100644 --- a/internal/testing/muster_manager.go +++ b/internal/testing/muster_manager.go @@ -1205,10 +1205,19 @@ func (m *musterInstanceManager) generateConfigFilesWithMocks(configPath string, }, } - // Apply custom main config if provided + // Apply custom main config if provided (shallow-merging one level of nested map values). if config != nil && config.MainConfig != nil { for key, value := range config.MainConfig.Config { - mainConfig[key] = value + existing, existsInMain := mainConfig[key] + existingMap, existingIsMap := existing.(map[string]interface{}) + valueMap, valueIsMap := value.(map[string]interface{}) + if existsInMain && existingIsMap && valueIsMap { + for k, v := range valueMap { + existingMap[k] = v + } + } else { + mainConfig[key] = value + } } } diff --git a/internal/testing/scenarios/poll-capability-refresh-nonoauth.yaml b/internal/testing/scenarios/poll-capability-refresh-nonoauth.yaml new file mode 100644 index 000000000..4e85f56e9 --- /dev/null +++ b/internal/testing/scenarios/poll-capability-refresh-nonoauth.yaml @@ -0,0 +1,76 @@ +name: "poll-capability-refresh-nonoauth" +description: "Test periodic capability polling detects tool changes without notifications" +category: "behavioral" +concept: "mcpserver" +tags: ["mcpserver", "polling", "tools", "capability-freshness"] +timeout: "2m" + +# Test Story: Capability Poller Detects Silent Tool Changes +# Given: A non-OAuth MCP server with one initial tool +# When: A new tool is silently added (no notifications/tools/list_changed sent) +# Then: The capability poller detects the change and the new tool becomes available + +pre_configuration: + main_config: + config: + aggregator: + capabilityPollInterval: "2s" + mcp_servers: + - name: "poll-server" + config: + type: "streamable-http" + tools: + - name: "initial_tool" + description: "A tool that exists from startup" + input_schema: + type: "object" + properties: + msg: + type: "string" + responses: + - response: + status: "ok" + message: "initial: {{ .msg }}" + +steps: + # Phase 1: Verify the initial tool is available + - id: "verify-initial-tool" + tool: "x_poll-server_initial_tool" + args: + msg: "hello" + expected: + success: true + contains: ["initial: hello"] + timeout: "30s" + + # Phase 2: Silently add a tool (no notification sent). + # The test handler waits for the poller to detect the change before returning. + - id: "add-tool-silently" + tool: "test_add_mock_tool_silently" + args: + server: "poll-server" + tool_name: "silent_tool" + tool_description: "A tool added without notification" + expected: + success: true + contains: ["poller detected"] + timeout: "60s" + + # Phase 3: Call the silently-added tool (already detected by poller in previous step) + - id: "verify-silent-tool-detected" + tool: "x_poll-server_silent_tool" + args: {} + expected: + success: true + contains: ["silent_tool"] + timeout: "10s" + + # Phase 4: Verify the initial tool still works + - id: "verify-initial-tool-still-works" + tool: "x_poll-server_initial_tool" + args: + msg: "still here" + expected: + success: true + contains: ["initial: still here"] + timeout: "10s" diff --git a/internal/testing/test_tools.go b/internal/testing/test_tools.go index 145fb7e88..a8cb7b81d 100644 --- a/internal/testing/test_tools.go +++ b/internal/testing/test_tools.go @@ -54,6 +54,10 @@ const ( // TestToolRemoveMockTool dynamically removes a tool from a running mock MCP server. // This triggers a notifications/tools/list_changed notification to all connected clients. TestToolRemoveMockTool = "test_remove_mock_tool" + // TestToolAddMockToolSilently adds a tool to a mock MCP server without + // sending a notifications/tools/list_changed notification. This simulates + // a silent server redeployment and is used to test the capability poller. + TestToolAddMockToolSilently = "test_add_mock_tool_silently" ) // TestToolsHandler handles test-specific tools that operate on mock infrastructure. @@ -166,7 +170,8 @@ func IsTestTool(toolName string) bool { TestToolSimulateMusterReauth, TestToolMusterAuthLogin, TestToolAddMockTool, - TestToolRemoveMockTool: + TestToolRemoveMockTool, + TestToolAddMockToolSilently: return true } return false @@ -207,6 +212,8 @@ func (h *TestToolsHandler) HandleTestTool(ctx context.Context, toolName string, return h.handleAddMockTool(ctx, args) case TestToolRemoveMockTool: return h.handleRemoveMockTool(ctx, args) + case TestToolAddMockToolSilently: + return h.handleAddMockToolSilently(ctx, args) default: return nil, fmt.Errorf("unknown test tool: %s", toolName) } @@ -1640,3 +1647,66 @@ func (h *TestToolsHandler) handleRemoveMockTool(ctx context.Context, args map[st "tool": toolName, }, nil } + +// handleAddMockToolSilently adds a tool to a mock server without sending +// a notifications/tools/list_changed notification. The tool is visible via +// ListTools but the aggregator's notification subscriber will NOT be +// triggered. Only the capability poller will detect the change. +// +// Args: +// - server: Required. Name of the mock MCP server. +// - tool_name: Required. Name of the new tool. +// - tool_description: Optional. Description of the new tool. +func (h *TestToolsHandler) handleAddMockToolSilently(ctx context.Context, args map[string]interface{}) (interface{}, error) { + serverName, ok := args["server"].(string) + if !ok || serverName == "" { + return nil, fmt.Errorf("server argument is required") + } + + toolName, ok := args["tool_name"].(string) + if !ok || toolName == "" { + return nil, fmt.Errorf("tool_name argument is required") + } + + if h.instanceManager == nil || h.currentInstance == nil { + return nil, fmt.Errorf("instance manager or current instance not available") + } + + description, _ := args["tool_description"].(string) + + toolConfig := mock.ToolConfig{ + Name: toolName, + Description: description, + Responses: []mock.ToolResponse{ + {Response: map[string]interface{}{"status": "ok", "tool": toolName}}, + }, + } + + httpServer := h.instanceManager.GetMockHTTPServer(h.currentInstance.ID, serverName) + if httpServer == nil { + return nil, fmt.Errorf("mock HTTP server %s not found for instance %s", serverName, h.currentInstance.ID) + } + httpServer.AddDynamicToolSilently(toolConfig) + + if h.debug { + h.logger.Debug("Silently added mock tool '%s' to server '%s' (no notification)\n", toolName, serverName) + } + + expectedToolName := fmt.Sprintf("x_%s_%s", serverName, toolName) + pollCtx, cancel := context.WithTimeout(ctx, 30*time.Second) + defer cancel() + if err := h.waitForToolVisibility(pollCtx, expectedToolName, true); err != nil { + return nil, fmt.Errorf("tool silently added to mock server but poller did not detect it: %w", err) + } + + if h.debug { + h.logger.Debug("Poller detected silently added tool '%s' on server '%s'\n", toolName, serverName) + } + + return map[string]interface{}{ + "success": true, + "message": fmt.Sprintf("Silently added tool '%s' to mock server '%s' and poller detected it", toolName, serverName), + "server": serverName, + "tool": toolName, + }, nil +}