diff --git a/src/Netclaw.Cli.Tests/Mcp/McpOAuthEndToEndTests.cs b/src/Netclaw.Cli.Tests/Mcp/McpOAuthEndToEndTests.cs index 8df1176b9..69feff9b3 100644 --- a/src/Netclaw.Cli.Tests/Mcp/McpOAuthEndToEndTests.cs +++ b/src/Netclaw.Cli.Tests/Mcp/McpOAuthEndToEndTests.cs @@ -135,6 +135,19 @@ public async ValueTask InitializeAsync( return new McpClientInitialization(tools.Cast().ToList()); } + public ValueTask> ListToolsAsync( + McpClient client, + CancellationToken cancellationToken) + => ListToolsCoreAsync(client, cancellationToken); + + private async ValueTask> ListToolsCoreAsync( + McpClient client, + CancellationToken cancellationToken) + { + var tools = await client.ListToolsAsync(cancellationToken: cancellationToken); + return tools.Cast().ToList(); + } + public ValueTask InvokeAsync( AIFunction function, AIFunctionArguments? arguments, diff --git a/src/Netclaw.Daemon.Tests/Mcp/McpCatalogRefreshTests.cs b/src/Netclaw.Daemon.Tests/Mcp/McpCatalogRefreshTests.cs new file mode 100644 index 000000000..7745ceca0 --- /dev/null +++ b/src/Netclaw.Daemon.Tests/Mcp/McpCatalogRefreshTests.cs @@ -0,0 +1,249 @@ +// ----------------------------------------------------------------------- +// +// Copyright (C) 2026 - 2026 Petabridge, LLC +// +// ----------------------------------------------------------------------- +using System.Text.Json; +using Microsoft.Extensions.AI; +using Microsoft.Extensions.Time.Testing; +using Netclaw.Daemon.Mcp; +using Xunit; + +namespace Netclaw.Daemon.Tests.Mcp; + +public sealed class McpCatalogRefreshTests +{ + private static readonly McpServerName ServerName = new("test"); + private static readonly DateTimeOffset InitialTime = DateTimeOffset.Parse("2026-07-22T12:00:00Z"); + + [Fact] + public async Task CatalogChange_RepublishesSnapshotAndGeneration() + { + var runtime = new McpClientManagerLifecycleTests.ControlledMcpClientRuntime(); + var plan = runtime.Enqueue(new McpClientManagerLifecycleTests.ClientPlan("old_tool")); + var time = new FakeTimeProvider(InitialTime); + await using var harness = CreateHarness(runtime, time); + await harness.Manager.StartAsync(TestContext.Current.CancellationToken); + + // Connect marked the catalog fresh; the throttle must elapse before a refresh. + time.Advance(McpClientManager.CatalogRefreshInterval); + plan.ToolNames = ["old_tool", "new_tool"]; + Assert.True(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + + var snapshot = Assert.IsType(harness.Manager.GetSnapshot(ServerName)); + Assert.Equal(2, snapshot.Generation); + Assert.Equal(2, snapshot.ToolFunctions.Count); + Assert.Equal(2, snapshot.Status.ToolCount); + Assert.Equal(1, plan.RefreshCount); + Assert.Equal(1, runtime.CreateCount); // no reconnect + AssertPublishedTools(harness, "new_tool", "old_tool"); + } + + [Fact] + public async Task NoCatalogChange_DoesNotBumpGeneration() + { + var runtime = new McpClientManagerLifecycleTests.ControlledMcpClientRuntime(); + var plan = runtime.Enqueue(new McpClientManagerLifecycleTests.ClientPlan("stable_tool")); + var time = new FakeTimeProvider(InitialTime); + await using var harness = CreateHarness(runtime, time); + await harness.Manager.StartAsync(TestContext.Current.CancellationToken); + + time.Advance(McpClientManager.CatalogRefreshInterval); + Assert.False(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + + var snapshot = Assert.IsType(harness.Manager.GetSnapshot(ServerName)); + Assert.Equal(1, snapshot.Generation); + Assert.Equal(1, plan.RefreshCount); + Assert.Equal(1, runtime.CreateCount); + AssertPublishedTools(harness, "stable_tool"); + } + + [Fact] + public async Task RefreshIsThrottledWithinInterval() + { + var runtime = new McpClientManagerLifecycleTests.ControlledMcpClientRuntime(); + var plan = runtime.Enqueue(new McpClientManagerLifecycleTests.ClientPlan("tool_a")); + var time = new FakeTimeProvider(InitialTime); + await using var harness = CreateHarness(runtime, time); + await harness.Manager.StartAsync(TestContext.Current.CancellationToken); + + // First refresh immediately after connect is throttled (connect marked it fresh). + Assert.False(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + + time.Advance(McpClientManager.CatalogRefreshInterval); + plan.ToolNames = ["tool_a", "tool_b"]; + Assert.True(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + + // Second refresh within the interval is throttled even though the catalog changed. + plan.ToolNames = ["tool_a", "tool_b", "tool_c"]; + Assert.False(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + + var snapshot = Assert.IsType(harness.Manager.GetSnapshot(ServerName)); + Assert.Equal(2, snapshot.Generation); + Assert.Equal(1, plan.RefreshCount); // only the middle call actually re-listed + } + + [Fact] + public async Task FailedRefresh_KeepsLastGoodCatalogAndGeneration() + { + var runtime = new McpClientManagerLifecycleTests.ControlledMcpClientRuntime(); + var plan = runtime.Enqueue(new McpClientManagerLifecycleTests.ClientPlan("old_tool")); + var time = new FakeTimeProvider(InitialTime); + await using var harness = CreateHarness(runtime, time); + await harness.Manager.StartAsync(TestContext.Current.CancellationToken); + + time.Advance(McpClientManager.CatalogRefreshInterval); + plan.ListFailure = new InvalidOperationException("server blew up"); + Assert.False(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + Assert.Equal(1, plan.RefreshCount); // prove the failure path actually ran + + var snapshot = Assert.IsType(harness.Manager.GetSnapshot(ServerName)); + Assert.Equal(1, snapshot.Generation); + Assert.Equal("old_tool", Assert.Single(snapshot.ToolFunctions).Key); + Assert.Equal(McpConnectionState.Connected, snapshot.Status.State); + AssertPublishedTools(harness, "old_tool"); + } + + [Fact] + public async Task FailedRefresh_RollsBackThrottleSoNextTickRetries() + { + var runtime = new McpClientManagerLifecycleTests.ControlledMcpClientRuntime(); + var plan = runtime.Enqueue(new McpClientManagerLifecycleTests.ClientPlan("tool_a")); + var time = new FakeTimeProvider(InitialTime); + await using var harness = CreateHarness(runtime, time); + await harness.Manager.StartAsync(TestContext.Current.CancellationToken); + + time.Advance(McpClientManager.CatalogRefreshInterval); + plan.ListFailure = new InvalidOperationException("transient"); + Assert.False(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + Assert.Equal(1, plan.RefreshCount); + + // The claim was rolled back, so a 30s advance allows an immediate retry rather + // than forcing a 5-minute wait. Catalog is unchanged, so no generation bump. + plan.ListFailure = null; + time.Advance(TimeSpan.FromSeconds(30)); + Assert.False(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + Assert.Equal(2, plan.RefreshCount); + Assert.Equal(1, harness.Manager.GetSnapshot(ServerName)?.Generation); + } + + [Fact] + public async Task EmptyCatalogRefresh_KeepsLastGoodTools() + { + var runtime = new McpClientManagerLifecycleTests.ControlledMcpClientRuntime(); + var plan = runtime.Enqueue(new McpClientManagerLifecycleTests.ClientPlan("tool_a", "tool_b")); + var time = new FakeTimeProvider(InitialTime); + await using var harness = CreateHarness(runtime, time); + await harness.Manager.StartAsync(TestContext.Current.CancellationToken); + + time.Advance(McpClientManager.CatalogRefreshInterval); + plan.ToolNames = []; // server now reports no tools + Assert.False(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + + var snapshot = Assert.IsType(harness.Manager.GetSnapshot(ServerName)); + Assert.Equal(1, snapshot.Generation); + Assert.Equal(2, snapshot.ToolFunctions.Count); + Assert.Equal(McpConnectionState.Connected, snapshot.Status.State); + AssertPublishedTools(harness, "tool_a", "tool_b"); + } + + [Fact] + public async Task EmptyCatalogRefresh_RollsBackThrottleSoNextTickRetries() + { + var runtime = new McpClientManagerLifecycleTests.ControlledMcpClientRuntime(); + var plan = runtime.Enqueue(new McpClientManagerLifecycleTests.ClientPlan("tool_a", "tool_b")); + var time = new FakeTimeProvider(InitialTime); + await using var harness = CreateHarness(runtime, time); + await harness.Manager.StartAsync(TestContext.Current.CancellationToken); + + time.Advance(McpClientManager.CatalogRefreshInterval); + plan.ToolNames = []; // server now reports no tools + Assert.False(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + Assert.Equal(1, plan.RefreshCount); + + // The last-good, previously non-empty catalog stays published. + var snapshotAfterEmpty = Assert.IsType(harness.Manager.GetSnapshot(ServerName)); + Assert.Equal(1, snapshotAfterEmpty.Generation); + Assert.Equal(2, snapshotAfterEmpty.ToolFunctions.Count); + AssertPublishedTools(harness, "tool_a", "tool_b"); + + // The claim was rolled back, so a 30s advance allows an immediate retry rather + // than forcing a 5-minute wait. The server recovers with a changed catalog, so + // the re-list runs and the snapshot generation bumps. + plan.ToolNames = ["tool_a", "tool_b", "tool_c"]; + time.Advance(TimeSpan.FromSeconds(30)); + Assert.True(await harness.Manager.TryRefreshCatalogAsync(ServerName, TestContext.Current.CancellationToken)); + Assert.Equal(2, plan.RefreshCount); + + var snapshotAfterRecovery = Assert.IsType(harness.Manager.GetSnapshot(ServerName)); + Assert.Equal(2, snapshotAfterRecovery.Generation); + Assert.Equal(3, snapshotAfterRecovery.ToolFunctions.Count); + AssertPublishedTools(harness, "tool_a", "tool_b", "tool_c"); + } + + [Fact] + public async Task RefreshOnUnknownServer_IsNoOp() + { + var runtime = new McpClientManagerLifecycleTests.ControlledMcpClientRuntime(); + runtime.Enqueue(new McpClientManagerLifecycleTests.ClientPlan("tool_a")); + await using var harness = CreateHarness(runtime); + await harness.Manager.StartAsync(TestContext.Current.CancellationToken); + + Assert.False(await harness.Manager.TryRefreshCatalogAsync(new McpServerName("missing"), TestContext.Current.CancellationToken)); + } + + [Fact] + public void Fingerprint_IgnoresToolOrder() + { + var a = AIFunctionFactory.Create(() => "unused", name: "tool_a", description: "desc"); + var b = AIFunctionFactory.Create(() => "unused", name: "tool_b", description: "desc"); + + Assert.Equal( + McpClientManager.ComputeCatalogFingerprint([a, b]), + McpClientManager.ComputeCatalogFingerprint([b, a])); + } + + [Fact] + public void Fingerprint_ChangesOnDescriptionOrToolAdd() + { + var baseline = AIFunctionFactory.Create(() => "unused", name: "tool", description: "desc"); + var newDescription = AIFunctionFactory.Create(() => "unused", name: "tool", description: "changed"); + var addedTool = AIFunctionFactory.Create(() => "unused", name: "other", description: "desc"); + + var baselineHash = McpClientManager.ComputeCatalogFingerprint([baseline]); + Assert.NotEqual(baselineHash, McpClientManager.ComputeCatalogFingerprint([newDescription])); + Assert.NotEqual(baselineHash, McpClientManager.ComputeCatalogFingerprint([baseline, addedTool])); + } + + [Fact] + public void CanonicalSchema_IgnoresKeyOrderAndWhitespace() + { + var a = JsonDocument.Parse("""{"z":1,"a":{"y":true,"b":"x"}}""").RootElement; + var b = JsonDocument.Parse(""" { "a": { "b": "x", "y": true }, "z": 1 } """).RootElement; + + Assert.Equal(McpClientManager.CanonicalSchema(a), McpClientManager.CanonicalSchema(b)); + } + + [Fact] + public void CanonicalSchema_ChangesOnSchemaEdit() + { + var a = JsonDocument.Parse("""{"type":"object","properties":{"a":{"type":"number"}}}""").RootElement; + var b = JsonDocument.Parse("""{"type":"object","properties":{"a":{"type":"string"}}}""").RootElement; + + Assert.NotEqual(McpClientManager.CanonicalSchema(a), McpClientManager.CanonicalSchema(b)); + } + + private static void AssertPublishedTools(McpClientManagerLifecycleTests.ManagerHarness harness, params string[] expected) + { + Assert.Equal(expected, harness.Manager.GetToolNames(ServerName)); + Assert.Equal(expected.Length, harness.Manager.GetServerStatuses()[ServerName].ToolCount); + } + + private static McpClientManagerLifecycleTests.ManagerHarness CreateHarness(McpClientManagerLifecycleTests.ControlledMcpClientRuntime runtime) + => new(runtime, new FakeTimeProvider(InitialTime)); + + private static McpClientManagerLifecycleTests.ManagerHarness CreateHarness( + McpClientManagerLifecycleTests.ControlledMcpClientRuntime runtime, + FakeTimeProvider time) + => new(runtime, time); +} diff --git a/src/Netclaw.Daemon.Tests/Mcp/McpClientManagerLifecycleTests.cs b/src/Netclaw.Daemon.Tests/Mcp/McpClientManagerLifecycleTests.cs index 1912f9b17..7a4e777f4 100644 --- a/src/Netclaw.Daemon.Tests/Mcp/McpClientManagerLifecycleTests.cs +++ b/src/Netclaw.Daemon.Tests/Mcp/McpClientManagerLifecycleTests.cs @@ -487,12 +487,19 @@ private static ManagerHarness CreateHarness( IOperationalNotificationSink notificationSink) => new(runtime, timeProvider, notificationSink); - private sealed class ManagerHarness : IAsyncDisposable + internal sealed class ManagerHarness : IAsyncDisposable { private readonly McpOAuthFlowBroker _flowBroker; private bool _stopFailureObserved; private bool _managerDisposed; + public ManagerHarness( + ControlledMcpClientRuntime runtime, + FakeTimeProvider timeProvider) + : this(runtime, timeProvider, NullNotificationSink.Instance) + { + } + public ManagerHarness( ControlledMcpClientRuntime runtime, FakeTimeProvider timeProvider, @@ -555,7 +562,7 @@ public async ValueTask DisposeAsync() } } - private sealed class ControlledMcpClientRuntime : IMcpClientRuntime + internal sealed class ControlledMcpClientRuntime : IMcpClientRuntime { private readonly ConcurrentQueue _plans = new(); private readonly ConcurrentDictionary _clients = new(); @@ -598,6 +605,23 @@ public async ValueTask InitializeAsync( if (plan.Initialize is not null) await plan.Initialize(cancellationToken); + var functions = BuildFunctions(plan); + return new McpClientInitialization(functions.Values.ToList()); + } + + public ValueTask> ListToolsAsync( + McpClient client, + CancellationToken cancellationToken) + { + var plan = _clients[client]; + Interlocked.Increment(ref plan.RefreshCountStorage); + if (plan.ListFailure is not null) + return ValueTask.FromException>(plan.ListFailure); + return ValueTask.FromResult>(BuildFunctions(plan).Values.ToList()); + } + + private IReadOnlyDictionary BuildFunctions(ClientPlan plan) + { var functions = new Dictionary(StringComparer.OrdinalIgnoreCase); foreach (var name in plan.ToolNames) { @@ -606,7 +630,7 @@ public async ValueTask InitializeAsync( _functions[function] = plan; } - return new McpClientInitialization(functions.Values.ToList()); + return functions; } public async ValueTask InvokeAsync( @@ -632,9 +656,9 @@ public ValueTask DisposeAsync(McpClient client) } } - private sealed class ClientPlan(params string[] toolNames) + internal sealed class ClientPlan(params string[] toolNames) { - public string[] ToolNames { get; } = toolNames; + public string[] ToolNames { get; set; } = toolNames; public Func? Initialize { get; init; } @@ -642,6 +666,8 @@ private sealed class ClientPlan(params string[] toolNames) public Exception? DisposeFailure { get; init; } + public Exception? ListFailure { get; set; } + public TaskCompletionSource Created { get; } = new(TaskCreationOptions.RunContinuationsAsynchronously); @@ -654,9 +680,13 @@ private sealed class ClientPlan(params string[] toolNames) public int DisposeCountStorage; + public int RefreshCountStorage; + public int InvocationCount => Volatile.Read(ref InvocationCountStorage); public int DisposeCount => Volatile.Read(ref DisposeCountStorage); + + public int RefreshCount => Volatile.Read(ref RefreshCountStorage); } private sealed class RecordingNotificationSink : IOperationalNotificationSink diff --git a/src/Netclaw.Daemon.Tests/Mcp/McpReconnectionServiceTests.cs b/src/Netclaw.Daemon.Tests/Mcp/McpReconnectionServiceTests.cs index fcdc7cb5f..1a909382a 100644 --- a/src/Netclaw.Daemon.Tests/Mcp/McpReconnectionServiceTests.cs +++ b/src/Netclaw.Daemon.Tests/Mcp/McpReconnectionServiceTests.cs @@ -33,6 +33,8 @@ public async Task SkipsNonUnreachableServers() await service.CheckAndReconnectAsync(CancellationToken.None); Assert.Equal(0, _reconnectable.ReconnectCallCount); + // Only the Connected server gets a catalog refresh; AwaitingAuth/AuthFailed/Disabled do not. + Assert.Equal(1, _reconnectable.RefreshCallCount); } [Fact] @@ -181,9 +183,12 @@ private sealed class FakeMcpReconnectable : IMcpReconnectable { private readonly Dictionary _statuses = new(); private int _reconnectCallCount; + private int _refreshCallCount; public int ReconnectCallCount => Volatile.Read(ref _reconnectCallCount); + public int RefreshCallCount => Volatile.Read(ref _refreshCallCount); + public Func>? OnReconnect { get; set; } public void SetStatus(string name, McpConnectionState state, int toolCount = 0) @@ -204,6 +209,12 @@ public async Task TryReconnectAsync(McpServerName serverName, Cancellation return await OnReconnect(serverName, ct); return false; } + + public Task TryRefreshCatalogAsync(McpServerName serverName, CancellationToken ct = default) + { + Interlocked.Increment(ref _refreshCallCount); + return Task.FromResult(true); + } } private sealed class FakeNotificationSink : IOperationalNotificationSink diff --git a/src/Netclaw.Daemon.Tests/Mcp/McpSdkOAuthFlowIntegrationTests.cs b/src/Netclaw.Daemon.Tests/Mcp/McpSdkOAuthFlowIntegrationTests.cs index 2381fc9c0..84f01a0d9 100644 --- a/src/Netclaw.Daemon.Tests/Mcp/McpSdkOAuthFlowIntegrationTests.cs +++ b/src/Netclaw.Daemon.Tests/Mcp/McpSdkOAuthFlowIntegrationTests.cs @@ -837,6 +837,19 @@ public async ValueTask InitializeAsync( return new McpClientInitialization(tools.Cast().ToList()); } + public ValueTask> ListToolsAsync( + McpClient client, + CancellationToken cancellationToken) + => ListToolsCoreAsync(client, cancellationToken); + + private async ValueTask> ListToolsCoreAsync( + McpClient client, + CancellationToken cancellationToken) + { + var tools = await client.ListToolsAsync(cancellationToken: cancellationToken); + return tools.Cast().ToList(); + } + public ValueTask InvokeAsync( AIFunction function, AIFunctionArguments? arguments, diff --git a/src/Netclaw.Daemon/Mcp/IMcpReconnectable.cs b/src/Netclaw.Daemon/Mcp/IMcpReconnectable.cs index cd5cf7033..bc126bcfa 100644 --- a/src/Netclaw.Daemon/Mcp/IMcpReconnectable.cs +++ b/src/Netclaw.Daemon/Mcp/IMcpReconnectable.cs @@ -12,4 +12,11 @@ internal interface IMcpReconnectable IReadOnlyDictionary GetServerStatuses(); Task TryReconnectAsync(McpServerName serverName, CancellationToken ct = default); + + /// + /// Re-lists a healthy server's tool catalog on its live client. Throttled by the + /// implementer; returns false when the server is not connected, refreshed too + /// recently, or the catalog is unchanged. + /// + Task TryRefreshCatalogAsync(McpServerName serverName, CancellationToken ct = default); } diff --git a/src/Netclaw.Daemon/Mcp/McpClientManager.cs b/src/Netclaw.Daemon/Mcp/McpClientManager.cs index b776949cf..87b898549 100644 --- a/src/Netclaw.Daemon/Mcp/McpClientManager.cs +++ b/src/Netclaw.Daemon/Mcp/McpClientManager.cs @@ -3,16 +3,22 @@ // Copyright (C) 2026 - 2026 Petabridge, LLC // // ----------------------------------------------------------------------- +using System.Buffers.Binary; using System.Collections.Concurrent; using System.Collections.ObjectModel; using System.Net; using System.Runtime.ExceptionServices; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using System.Text.Json.Nodes; using Microsoft.Extensions.AI; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using ModelContextProtocol; using ModelContextProtocol.Authentication; using ModelContextProtocol.Client; +using ModelContextProtocol.Protocol; using Netclaw.Actors.Tools; using Netclaw.Configuration; using Netclaw.Security; @@ -42,6 +48,20 @@ internal sealed class McpClientManager : IHostedService, IDisposable, IMcpToolIn private bool _disposed; private Task? _stopTask; + /// + /// Minimum time between catalog refresh polls on a healthy connection. The + /// reconnection service ticks every 30 seconds; this throttle keeps the + /// re-list RPC quiet on stable servers while still converging within a few + /// minutes of a live catalog change. + /// + internal static readonly TimeSpan CatalogRefreshInterval = TimeSpan.FromMinutes(5); + + /// + /// Bounds a single catalog re-list so a server that stops answering cannot stall + /// the poll loop, block a reconnect on the per-server gate, or delay shutdown. + /// + internal static readonly TimeSpan CatalogRefreshTimeout = TimeSpan.FromSeconds(15); + public McpClientManager( Dictionary serverEntries, ToolRegistry toolRegistry, @@ -176,6 +196,154 @@ public async Task TryReconnectAsync(McpServerName serverName, Cancellation return await ReconnectAsync(lifecycle, entry, observed, ct, null); } + /// + /// Re-lists a connected server's tools on the live client and republishes the + /// snapshot when the catalog changed. Throttled to + /// per server. A failed refresh never empties the catalog: the last good snapshot + /// and generation stay published until a later refresh succeeds, and transport-level + /// failures are left to the invocation path and the reconnection service to handle. + /// + public async Task TryRefreshCatalogAsync(McpServerName serverName, CancellationToken ct = default) + { + if (IsStopping + || !_serverEntries.TryGetValue(serverName.Value, out var entry) + || !entry.Enabled) + return false; + + if (!_servers.TryGetValue(serverName, out var lifecycle) + || lifecycle.Snapshot is not { IsConnected: true }) + return false; + + using var candidateCancellation = CancellationTokenSource.CreateLinkedTokenSource( + ct, _lifetimeCancellation.Token); + candidateCancellation.CancelAfter(CatalogRefreshTimeout); + try + { + await lifecycle.Gate.WaitAsync(candidateCancellation.Token); + } + catch (OperationCanceledException) when (_lifetimeCancellation.IsCancellationRequested) + { + return false; + } + + try + { + if (IsStopping) + return false; + + var snapshot = lifecycle.Snapshot; + if (snapshot is not { IsConnected: true }) + return false; + + if (_flowBroker.TryGetActive(snapshot.Name, out _)) + return false; + + if (!lifecycle.TryClaimCatalogRefresh( + _timeProvider.GetUtcNow().ToUnixTimeMilliseconds(), + (long)CatalogRefreshInterval.TotalMilliseconds, + out var previousRefreshMs)) + { + return false; + } + + return await RefreshCatalogCoreAsync( + lifecycle, entry, snapshot, previousRefreshMs, candidateCancellation.Token); + } + finally + { + lifecycle.Gate.Release(); + } + } + + private async Task RefreshCatalogCoreAsync( + McpServerLifecycle lifecycle, + McpServerEntry entry, + McpServerSnapshot current, + long previousRefreshMs, + CancellationToken ct) + { + try + { + var tools = await _clientRuntime.ListToolsAsync(current.Client!, ct); + var functions = CreateFunctionMap(tools); + + // A server that was serving tools now reports none. Publishing an empty + // catalog would wipe the model-visible index, and the server is still + // Connected so the invocation path can't rescue it either. Treat it as + // transient: keep the last good snapshot, retry on the next poll window. + if (functions.Count == 0 && current.ToolFunctions.Count > 0) + { + // Roll back the throttle claim so the next 30s tick retries instead of + // waiting 5 minutes, matching the failed-refresh path below. + lifecycle.RollbackCatalogRefreshClaim(previousRefreshMs); + _logger.LogWarning( + "MCP server '{Name}' catalog refresh returned no tools; keeping {ToolCount} existing tool(s)", + current.Name.Value, + current.ToolFunctions.Count); + return false; + } + + var fingerprint = ComputeCatalogFingerprint(functions.Values); + if (string.Equals(fingerprint, current.CatalogFingerprint, StringComparison.Ordinal)) + return false; + + var publishedTools = ToolRegistrationExtensions.PrepareMcpTools( + current.Name.Value, + tools, + entry.GrantCategory, + this, + _maxToolDescriptionChars, + _maxToolSchemaWarnChars, + _logger); + LogToolDrift(current.Name, tools); + + McpServerSnapshot replacement; + lock (_shutdownSync) + { + if (_stopping) + return false; + + replacement = current with + { + ToolFunctions = functions, + Generation = checked(current.Generation + 1), + Status = new McpServerStatus( + current.Name, + McpConnectionState.Connected, + functions.Count, + null, + current.Status.LastErrorAt), + CatalogFingerprint = fingerprint, + }; + // Connection first, tools second — same ordering as the connect path. + lifecycle.Publish(replacement); + _toolRegistry.PublishMcpServerTools(current.Name.Value, publishedTools); + } + + _logger.LogInformation( + "MCP server '{Name}' catalog refreshed as generation {Generation} ({ToolCount} tools)", + current.Name.Value, + replacement.Generation, + functions.Count); + return true; + } + catch (OperationCanceledException) when (_lifetimeCancellation.IsCancellationRequested) + { + return false; + } + catch (Exception ex) + { + // Invariant: a failed refresh must never empty the catalog. Roll back the + // throttle claim so the next 30s tick retries instead of waiting 5 minutes. + lifecycle.RollbackCatalogRefreshClaim(previousRefreshMs); + _logger.LogWarning(ex, + "MCP server '{Name}' catalog refresh failed; keeping generation {Generation} unchanged", + current.Name.Value, + current.Generation); + return false; + } + } + internal async Task StartAuthorizationAsync( McpServerName serverName, CancellationToken requestCancellation) @@ -388,6 +556,7 @@ private async Task BuildAndPublishCandidateAsync( var initialization = await _clientRuntime.InitializeAsync(candidate, ct); var tools = initialization.Tools; var functions = CreateFunctionMap(tools); + var catalogFingerprint = ComputeCatalogFingerprint(functions.Values); var publishedTools = ToolRegistrationExtensions.PrepareMcpTools( current.Name.Value, tools, @@ -430,11 +599,14 @@ private async Task BuildAndPublishCandidateAsync( candidate, functions, checked(current.Generation + 1), - connectedStatus); + connectedStatus, + catalogFingerprint); // Connection first, tools second. A tool the model can see is then always // dispatchable, because dispatch resolves it from this snapshot. lifecycle.Publish(replacement); _toolRegistry.PublishMcpServerTools(current.Name.Value, publishedTools); + // The connect path just listed the catalog; skip the next poll window. + lifecycle.MarkCatalogRefreshed(_timeProvider.GetUtcNow().ToUnixTimeMilliseconds()); candidate = null; oauthCache = null; } @@ -1202,6 +1374,61 @@ internal static IReadOnlyDictionary CreateFunctionMap(IReadO return new ReadOnlyDictionary(map); } + /// + /// Computes a content checksum over the model-visible surface of a server's tool + /// catalog: name, description, input schema, and return schema of every tool. + /// Order-independent (tools are sorted by name) and schema-canonical (object keys + /// sorted, whitespace normalized), so a server reordering either is not mistaken + /// for a change. Equal fingerprints mean the catalog is unchanged; any add, remove, + /// rename, or schema edit changes the checksum. + /// + internal static string ComputeCatalogFingerprint(IEnumerable tools) + { + using var stream = new MemoryStream(); + foreach (var tool in tools.OrderBy(t => t.Name, StringComparer.Ordinal)) + { + WriteField(stream, tool.Name); + WriteField(stream, tool.Description ?? string.Empty); + WriteField(stream, CanonicalSchema(tool.JsonSchema)); + WriteField(stream, tool.ReturnJsonSchema is { } returnSchema ? CanonicalSchema(returnSchema) : string.Empty); + } + + return Convert.ToHexString(SHA256.HashData(stream.ToArray())); + } + + /// Canonicalizes a schema (sorted object keys, normalized whitespace) so + /// semantically identical schemas hash the same. Exposed for tests. + internal static string CanonicalSchema(JsonElement schema) + { + if (schema.ValueKind is JsonValueKind.Undefined or JsonValueKind.Null) + return string.Empty; + + return CanonicalNode(schema)?.ToJsonString() ?? string.Empty; + } + + private static JsonNode? CanonicalNode(JsonElement element) => element.ValueKind switch + { + JsonValueKind.Object => new JsonObject(element.EnumerateObject() + .OrderBy(property => property.Name, StringComparer.Ordinal) + .Select(property => new KeyValuePair(property.Name, CanonicalNode(property.Value)))), + JsonValueKind.Array => new JsonArray(element.EnumerateArray().Select(CanonicalNode).ToArray()), + JsonValueKind.String => JsonValue.Create(element.GetString()), + JsonValueKind.Number => JsonValue.Create(element), + JsonValueKind.True => JsonValue.Create(true), + JsonValueKind.False => JsonValue.Create(false), + JsonValueKind.Null => null, + _ => JsonValue.Create(element.GetRawText()), + }; + + private static void WriteField(Stream stream, string value) + { + var bytes = Encoding.UTF8.GetBytes(value); + Span length = stackalloc byte[4]; + BinaryPrimitives.WriteInt32LittleEndian(length, bytes.Length); + stream.Write(length); + stream.Write(bytes); + } + private void LogToolDrift(McpServerName serverName, IReadOnlyList discoveredTools) { var profiles = _toolConfig.AudienceProfiles; @@ -1288,6 +1515,11 @@ ValueTask InitializeAsync( McpClient client, CancellationToken cancellationToken); + /// Re-lists a connected client's tools from the server (no reconnect). + ValueTask> ListToolsAsync( + McpClient client, + CancellationToken cancellationToken); + ValueTask InvokeAsync( AIFunction function, AIFunctionArguments? arguments, @@ -1307,9 +1539,17 @@ public Task CreateAsync( public async ValueTask InitializeAsync( McpClient client, CancellationToken cancellationToken) + { + var tools = await ListToolsAsync(client, cancellationToken); + return new McpClientInitialization(tools); + } + + public async ValueTask> ListToolsAsync( + McpClient client, + CancellationToken cancellationToken) { var tools = await client.ListToolsAsync(cancellationToken: cancellationToken); - return new McpClientInitialization(tools.Cast().ToList()); + return tools.Cast().ToList(); } public ValueTask InvokeAsync( @@ -1331,12 +1571,35 @@ internal sealed record McpClientCandidate( internal sealed class McpServerLifecycle(McpServerSnapshot initialSnapshot) { private McpServerSnapshot? _snapshot = initialSnapshot; + private long _lastCatalogRefreshMs; public SemaphoreSlim Gate { get; } = new(1, 1); public McpServerSnapshot? Snapshot => Volatile.Read(ref _snapshot); public void Publish(McpServerSnapshot? snapshot) => Volatile.Write(ref _snapshot, snapshot); + + /// + /// Claims the next catalog refresh slot when has + /// elapsed since the last refresh or connect. Callers hold , so a + /// plain field is sufficient. Outputs the previous claim so a failed attempt can roll + /// back and retry sooner than the full interval. + /// + public bool TryClaimCatalogRefresh(long nowMs, long minIntervalMs, out long previousMs) + { + previousMs = _lastCatalogRefreshMs; + if (nowMs - previousMs < minIntervalMs) + return false; + + _lastCatalogRefreshMs = nowMs; + return true; + } + + /// Restores the claim timestamp after a failed refresh so the next tick retries. + public void RollbackCatalogRefreshClaim(long previousMs) => _lastCatalogRefreshMs = previousMs; + + /// Records a successful catalog refresh (or connect) so the poll throttle starts from now. + public void MarkCatalogRefreshed(long nowMs) => _lastCatalogRefreshMs = nowMs; } internal sealed record McpServerSnapshot( @@ -1344,7 +1607,8 @@ internal sealed record McpServerSnapshot( McpClient? Client, IReadOnlyDictionary ToolFunctions, long Generation, - McpServerStatus Status) + McpServerStatus Status, + string CatalogFingerprint = "") { private static readonly IReadOnlyDictionary EmptyFunctions = new ReadOnlyDictionary(new Dictionary()); diff --git a/src/Netclaw.Daemon/Mcp/McpReconnectionService.cs b/src/Netclaw.Daemon/Mcp/McpReconnectionService.cs index 05151377e..bfe269ceb 100644 --- a/src/Netclaw.Daemon/Mcp/McpReconnectionService.cs +++ b/src/Netclaw.Daemon/Mcp/McpReconnectionService.cs @@ -58,6 +58,12 @@ internal async Task CheckAndReconnectAsync(CancellationToken ct) if (status.State is not McpConnectionState.Unreachable) { _backoff.TryRemove(serverName, out _); + + // Healthy connections get a throttled catalog refresh so live tool + // changes surface without a reconnect. The manager owns the cadence. + if (status.State is McpConnectionState.Connected) + await TryRefreshCatalogAsync(serverName, ct); + continue; } @@ -135,5 +141,25 @@ private void EmitReconnectedAlert(McpServerName serverName) })); } + private async Task TryRefreshCatalogAsync(McpServerName serverName, CancellationToken ct) + { + try + { + await _mcpReconnectable.TryRefreshCatalogAsync(serverName, ct); + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + // A refresh failure is not a connection failure — the manager keeps the + // last good catalog and the next tick retries. Log and move on. + _logger.LogDebug(ex, + "MCP server '{Name}' catalog refresh threw an exception", + serverName.Value); + } + } + internal readonly record struct BackoffState(int FailureCount, long LastAttemptMs); }