AI tools: 22 Task bridges gone, 17 uncancellable tool calls fixed, the ThreadPool hop made drainable

src/MeshWeaver.AI/AgentChatClient.cs

@@ -2065,62 +2065,6 @@ private void CreateAgentsSync()
         logger.LogDebug("[AgentChatClient] Created {Count} agents", agents.Count);
     }
 
-    /// <summary>
-    /// Creates ChatClientAgent instances for all loaded configurations.
-    /// </summary>
-    [Obsolete("Use CreateAgentsSync — CreateAgentsAsync deadlocks in Orleans")]
-    private async Task CreateAgentsAsync()
-    {
-        if (chatClientFactories.Count == 0)
-        {
-            logger.LogWarning("[AgentChatClient] No IChatClientFactory available, cannot create agents");
-            return;
-        }
-
-        if (agentsInitialized)
-            return;
-
-        // Select the appropriate factory based on the requested model
-        var factory = GetFactoryForModel(currentModelName);
-        if (factory == null)
-        {
-            logger.LogWarning("[AgentChatClient] No factory can serve model: {ModelName}", currentModelName);
-            throw new ArgumentException($"No factory can serve model: {currentModelName}");
-        }
-
-        isPersistentFactory = factory.IsPersistent;
-        logger.LogDebug("[AgentChatClient] Using factory {FactoryName} for model {ModelName} (persistent={IsPersistent})",
-            factory.Name, currentModelName ?? "default", isPersistentFactory);
-
-        var configs = loadedAgents.Select(a => a.AgentConfiguration).ToImmutableList();
-        var createdAgents = ImmutableDictionary<string, ChatClientAgent>.Empty;
-
-        // Order agents: non-delegating first, delegating second, default last
-        var orderedConfigs = OrderAgentsForCreation(configs);
-
-        // First pass: Create all agents in order
-        foreach (var agentConfig in orderedConfigs)
-        {
-            var agent = await factory.CreateAgentAsync(
-                agentConfig, this, createdAgents, configs, currentModelName);
-            createdAgents = createdAgents.SetItem(agentConfig.Id, agent);
-            agents = agents.SetItem(agentConfig.Id, agent);
-        }
-
-        // Second pass: Update agents with cyclic dependencies
-        var cyclicAgents = FindCyclicDelegations(configs);
-        foreach (var agentConfig in cyclicAgents)
-        {
-            var updatedAgent = await factory.CreateAgentAsync(
-                agentConfig, this, createdAgents, configs, currentModelName);
-            createdAgents = createdAgents.SetItem(agentConfig.Id, updatedAgent);
-            agents = agents.SetItem(agentConfig.Id, updatedAgent);
-        }
-
-        agentsInitialized = true;
-        logger.LogDebug("[AgentChatClient] Created {Count} agents", agents.Count);
-    }
-
     /// <summary>
     /// Resolves <see cref="currentModelName"/> to a model that can actually serve this round. Two
     /// cases land here, and they are NOT the same thing:

src/MeshWeaver.AI/Plugins/AgentFilesPlugin.cs

@@ -1,6 +1,5 @@
 using System.ComponentModel;
 using System.Reactive.Linq;
-using System.Reactive.Threading.Tasks;
 using System.Text;
 using MeshWeaver.AI.Stores;
 using MeshWeaver.Messaging;
@@ -68,20 +67,29 @@ public IEnumerable<AITool> CreateTools() =>
     ];
 
     // Boundary adapters. AIFunctionFactory requires Task<string>; the bodies are reactive and each
-    // ends in exactly one FirstAsync().ToTask() — the store's own API is IObservable throughout.
+    // ends in the ONE shared bridge, ToolTask.Bridge — the store's own API is IObservable throughout.
+    //
+    // 🚨 These used to end in `.Catch(...).FirstAsync().ToTask(ct)`. The token was observed and the
+    // fault was handled fluently, so the two loud terminals were covered — but the THIRD one was
+    // not. A source that completes WITHOUT emitting reaches FirstAsync, which signals
+    // InvalidOperationException("Sequence contains no elements") — past the .Catch, which sits
+    // upstream of it — so the model got a stack trace where it needed an answer. And that terminal
+    // is reachable here, not theoretical: MeshNodeStreamCache completes its per-path subjects on
+    // eviction and on dispose, which is exactly what a thread whose hub is going away does.
+    // ToolTask.Bridge makes the empty completion an ANSWER, and keeps the other three.
 
     private Task<string> ReadFile(
         [Description("Path of the file within your working area, e.g. 'notes.md' or 'research/findings.md'.")]
         string path,
         CancellationToken cancellationToken) =>
         Store is not { } store
             ? Task.FromResult(NoThread)
-            : store.Read(path)
-                .Take(1)
-                .Select(content => content ?? $"No file at '{path}'.")
-                .Catch<string, Exception>(ex => Observable.Return($"Could not read '{path}': {ex.Message}"))
-                .FirstAsync()
-                .ToTask(cancellationToken);
+            : ToolTask.Bridge(
+                store.Read(path).Select(content => content ?? $"No file at '{path}'."),
+                cancellationToken,
+                answer => answer,
+                ex => $"Could not read '{path}': {ex.Message}",
+                () => $"No file at '{path}'.");
 
     private Task<string> WriteFile(
         [Description("Path of the file within your working area, e.g. 'notes.md' or 'research/findings.md'.")]
@@ -91,24 +99,26 @@ private Task<string> WriteFile(
         CancellationToken cancellationToken) =>
         Store is not { } store
             ? Task.FromResult(NoThread)
-            : store.Write(path, content)
-                .Select(node => $"Saved '{path}' ({content.Length} characters) at {node.Path}.")
-                .Catch<string, Exception>(ex => Observable.Return($"Could not save '{path}': {ex.Message}"))
-                .FirstAsync()
-                .ToTask(cancellationToken);
+            : ToolTask.Bridge(
+                store.Write(path, content)
+                    .Select(node => $"Saved '{path}' ({content.Length} characters) at {node.Path}."),
+                cancellationToken,
+                answer => answer,
+                ex => $"Could not save '{path}': {ex.Message}",
+                () => $"Could not save '{path}': the write produced no result.");
 
     private Task<string> ListFiles(
         [Description("Folder within your working area to list. Empty string lists the top level.")]
         string directory,
         CancellationToken cancellationToken) =>
         Store is not { } store
             ? Task.FromResult(NoThread)
-            : store.ListChildren(directory ?? "")
-                .Take(1)
-                .Select(FormatListing)
-                .Catch<string, Exception>(ex => Observable.Return($"Could not list '{directory}': {ex.Message}"))
-                .FirstAsync()
-                .ToTask(cancellationToken);
+            : ToolTask.Bridge(
+                store.ListChildren(directory ?? "").Select(FormatListing),
+                cancellationToken,
+                answer => answer,
+                ex => $"Could not list '{directory}': {ex.Message}",
+                () => "(empty)");
 
     private Task<string> SearchFiles(
         [Description("Folder within your working area to search. Empty string searches from the top level.")]
@@ -122,24 +132,26 @@ private Task<string> SearchFiles(
         CancellationToken cancellationToken) =>
         Store is not { } store
             ? Task.FromResult(NoThread)
-            : store.Search(directory ?? "", pattern, string.IsNullOrWhiteSpace(glob) ? null : glob, recursive)
-                .Take(1)
-                .Select(FormatMatches)
-                .Catch<string, Exception>(ex => Observable.Return($"Could not search '{directory}': {ex.Message}"))
-                .FirstAsync()
-                .ToTask(cancellationToken);
+            : ToolTask.Bridge(
+                store.Search(directory ?? "", pattern, string.IsNullOrWhiteSpace(glob) ? null : glob, recursive)
+                    .Select(FormatMatches),
+                cancellationToken,
+                answer => answer,
+                ex => $"Could not search '{directory}': {ex.Message}",
+                () => "No matches.");
 
     private Task<string> DeleteFile(
         [Description("Path of the file within your working area to delete.")]
         string path,
         CancellationToken cancellationToken) =>
         Store is not { } store
             ? Task.FromResult(NoThread)
-            : store.Delete(path)
-                .Select(deleted => deleted ? $"Deleted '{path}'." : $"No file at '{path}'.")
-                .Catch<string, Exception>(ex => Observable.Return($"Could not delete '{path}': {ex.Message}"))
-                .FirstAsync()
-                .ToTask(cancellationToken);
+            : ToolTask.Bridge(
+                store.Delete(path).Select(deleted => deleted ? $"Deleted '{path}'." : $"No file at '{path}'."),
+                cancellationToken,
+                answer => answer,
+                ex => $"Could not delete '{path}': {ex.Message}",
+                () => $"No file at '{path}'.");
 
     private const string NoThread =
         "No working area is available — this conversation has no thread yet. Try again after the "

src/MeshWeaver.AI/Plugins/CollaborationPlugin.cs

@@ -1,5 +1,4 @@
 using System.ComponentModel;
-using System.Reactive.Concurrency;
 using System.Reactive.Linq;
 using MeshWeaver.Graph;
 using MeshWeaver.Markdown.Collaboration;
@@ -14,10 +13,12 @@ namespace MeshWeaver.AI.Plugins;
 /// Agent plugin providing tools for adding comments and suggesting edits on
 /// Markdown documents via the collaborative editing infrastructure.
 ///
-/// Every method is await-free — reads are moved onto the <see cref="TaskPoolScheduler"/>
-/// with <c>.SubscribeOn(TaskPoolScheduler.Default)</c> (NEVER
-/// <c>Observable.FromAsync</c>, which is forbidden outside <c>IoPool</c>) so blocking
-/// enumeration never touches the hub scheduler, and writes go through
+/// Every method is await-free — reads leave the caller's scheduler through
+/// <see cref="ToolTask.Pooled{T}"/> (NEVER <c>Observable.FromAsync</c>, which is forbidden
+/// outside <c>IoPool</c>) so blocking enumeration never touches the hub scheduler AND the
+/// read stays visible to <c>IoPool.Drain()</c> — the bare
+/// <c>SubscribeOn(TaskPoolScheduler.Default)</c> this replaces achieved the first and
+/// silently gave up the second. Writes go through
 /// <c>hub.Observe(...)</c>, with <see cref="ToolTask.Bridge{T}"/> bridging the off-hub
 /// callback thread back to the caller — every terminal settles (value, error, EMPTY
 /// completion) and the round's cancellation token is observed (#1956).
@@ -76,8 +77,7 @@ public Task<string> AddComment(
         // completed WITHOUT emitting, and `cancellationToken` appeared only in the doc comment — so
         // a parked AddComment held its Ai-pool gate permit through IoPool.Drain and Stop no-opped.
         return ToolTask.Bridge(
-            ops.Get(resolvedInput)
-                .SubscribeOn(TaskPoolScheduler.Default)
+            ToolTask.Pooled(hub, ops.Get(resolvedInput))
                 .Take(1)
                 .SelectMany(docJson => AddCommentContinuation(
                     docJson, resolvedPath, documentPath, selectedText, commentText)),
@@ -171,8 +171,7 @@ public Task<string> SuggestEdit(
         // path reports for an address that routes nowhere. Same three-terminal bridge as
         // AddComment: value, error, and the EMPTY completion that used to park the round (#1956).
         return ToolTask.Bridge(
-            ops.Get(resolvedInput)
-                .SubscribeOn(TaskPoolScheduler.Default)
+            ToolTask.Pooled(hub, ops.Get(resolvedInput))
                 .Take(1)
                 .SelectMany(docJson => SuggestEditContinuation(
                     docJson, resolvedPath, documentPath, originalText, newText)),

src/MeshWeaver.AI/Plugins/DelegationTool.cs

@@ -83,6 +83,15 @@ public record SubThreadInfo(
 /// </summary>
 public static class DelegationTool
 {
+    /// <summary>
+    /// Moves a delegation's SUBSCRIBE off the caller's scheduler, through the mesh-scoped
+    /// <c>AgentStore</c> I/O pool when a hub is available so <c>IoPool.Drain()</c> can see and join
+    /// it, and through the bare ThreadPool otherwise (the legacy, workspace-less path, which is
+    /// exactly what it had before).
+    /// </summary>
+    private static IObservable<T> HopOffCallerScheduler<T>(IObservable<T> source, IMessageHub? hub) =>
+        hub is null ? source.SubscribeOn(TaskPoolScheduler.Default) : ToolTask.Pooled(hub, source);
+
     /// <summary>
     /// Creates the suite of delegation tools: <c>delegate_to_agent</c>, plus
     /// <c>list_sub_threads</c> and <c>send_to_sub_thread</c> when the parent
@@ -211,9 +220,24 @@ Task<string> Delegate(
 
             // Settle ONCE and release what the wait was holding. Every resolution path goes through
             // here — a bare tcs.TrySet* would resolve the caller and leave the subscriptions live.
+            //
+            // 🚨 DISPOSE FIRST, THEN SETTLE — the same invariant, and for the same reason, as
+            // ToolTask.Settle. This read `if (set()) pending.Dispose();`, i.e. the caller was told
+            // the delegation had ended and the subscriptions were released afterwards. Those are not
+            // interchangeable: `tcs` is created RunContinuationsAsynchronously, so TrySet* completes
+            // the task and schedules the parent round's continuation on the pool, which then runs
+            // CONCURRENTLY with the rest of this callback. On the cancellation path the round could
+            // therefore observe the cancel — release its IoPoolNames.Ai permit, let IoPool.Drain()
+            // through, let teardown dispose the service scope and unload collectible node ALCs —
+            // while this call still held a live subscription to the sub-thread's node stream.
+            // Disposing first makes "the work stopped" established before "the call ended" is
+            // observable. Dispose and TrySet* are both idempotent, so a losing terminal disposes an
+            // already-disposed bag and sets nothing; the lambdas keep returning bool because the
+            // chunk-aggregate path logs only when its TrySetResult actually won.
             void Settle(Func<bool> set)
             {
-                if (set()) pending.Dispose();
+                pending.Dispose();
+                set();
             }
 
             // 🚨 THE ROUND'S CANCELLATION MUST REACH THIS TASK. Nothing else can end the wait
@@ -336,16 +360,28 @@ void Settle(Func<bool> set)
             // above which reads Thread.Summary). No await foreach, no Task.Run, no
             // Observable.Create wrapper.
             //
-            // SubscribeOn(TaskPoolScheduler.Default): the parent agent loop's
-            // continuation (Orleans grain scheduler in prod, single-threaded pump in
-            // the DelegationDeadlockTest) MUST NOT host the sub-thread drain. The
-            // typical executeAsync impl is Observable.Create<async> over an
-            // IAsyncEnumerable, which captures the Subscribe-time SynchronizationContext
-            // for every MoveNextAsync — if that's the grain scheduler, every sub-thread
-            // continuation posts back through it and wedges the grain. Hop the Subscribe
-            // to ThreadPool so the entire drain runs free of the caller's context.
-            pending.Add(executeAsync(agentName, task, context, cancellationToken)
-            .SubscribeOn(TaskPoolScheduler.Default)
+            // The parent agent loop's continuation (Orleans grain scheduler in prod,
+            // single-threaded pump in the DelegationDeadlockTest) MUST NOT host the
+            // sub-thread drain. The typical executeAsync impl is Observable.Create<async>
+            // over an IAsyncEnumerable, which captures the Subscribe-time
+            // SynchronizationContext for every MoveNextAsync — if that's the grain
+            // scheduler, every sub-thread continuation posts back through it and wedges
+            // the grain. So the Subscribe hops off the caller's context.
+            //
+            // 🚨 It hops through ToolTask.Pooled — the mesh-scoped AgentStore pool — and no
+            // longer through a bare `.SubscribeOn(TaskPoolScheduler.Default)`. Both leave the
+            // caller's scheduler, which is all the paragraph above ever asked for; only one of
+            // them is visible to IoPool.Drain(). A sub-thread drain running on the raw ThreadPool
+            // is counted by nothing and waited for by nobody, so teardown disposes the service
+            // scope and unloads collectible node ALCs while it is still executing — the
+            // use-after-unload precondition Drain exists to rule out, and the reason
+            // IIoPool.SubscribeThroughPool documents itself as "the tracked, drainable
+            // replacement for a bare .SubscribeOn(TaskPoolScheduler.Default), which the drain
+            // cannot reach". AgentStore, never Ai: this call already sits inside a round holding
+            // an Ai permit. With no workspace (the legacy path) there is no hub to resolve the
+            // registry from, so that path keeps the untracked hop — which is what it had.
+            pending.Add(HopOffCallerScheduler(
+                    executeAsync(agentName, task, context, cancellationToken), workspace?.Hub)
             .Subscribe(
                 chunk => sb.Append(chunk),
                 ex =>