fix(a2a-client): stopped registry starts no watch loops

extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs

@@ -55,6 +55,9 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
     private readonly ConcurrentDictionary<string, byte> _flushRequested = new(StringComparer.Ordinal);
     private volatile IA2AResultDelivery _delivery;
 
+    /// <summary>Set by <see cref="StopAsync"/>: later starts and re-arms still persist their record but run no loop.</summary>
+    private volatile bool _stopped;
+
     public A2AWatchRegistry(TimeProvider time, HttpMessageHandler handler, ILogger? logger = null)
     {
         Time = time;
@@ -97,6 +100,8 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
         var watch = new Watch(new WatchRecord(watchId, target, agent.Entry.Name, task.Id, task.ContextId, null, Time.GetUtcNow(), source), workspace);
         using (await LockAsync(workspace, cancellationToken).ConfigureAwait(false))
         {
+            // A watch that is finishing (left the map, delivering) isn't seen here, so a start racing it may watch the
+            // task again: a rare duplicate result, within the registry's at-least-once delivery.
             if (_watches.Values.FirstOrDefault(w => w.Record.Target == target && w.Record.TaskId == task.Id) is { } existing)
             {
                 watch.DisposeStop();
@@ -138,6 +143,13 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
             watch.DisposeStop();
             throw;
         }
+        if (_stopped)
+        {
+            // Shutting down: the record stays in pending.json for the next restore, no loop outlives the registry.
+            _watches.TryRemove(new(watchId, watch));
+            watch.DisposeStop();
+            return watch.Record;
+        }
         watch.Loop = Task.Run(() => RunAsync(watch, settings));
         return watch.Record;
     }
@@ -174,6 +186,8 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
                     var watch = new Watch(record, workspace);
                     if (!_watches.TryAdd(record.WatchId, watch)) { watch.DisposeStop(); continue; }
                     if (!ReferenceEquals(record, stored)) PendingFile.Upsert(workspace, record);
+                    // Checked after the watch is in the map: a StopAsync that set the flag later sees (and stops) it.
+                    if (_stopped) { _watches.TryRemove(new(record.WatchId, watch)); watch.DisposeStop(); return count; }
                     watch.Loop = Task.Run(() => RunAsync(watch, settings));
                     count++;
                 }
@@ -255,6 +269,13 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
             var watch = new Watch(record with { Undelivered = null }, workspace);
             if (!_watches.TryAdd(record.WatchId, watch)) { watch.DisposeStop(); return; }
             PendingFile.Upsert(workspace, watch.Record);
+            if (_stopped)
+            {
+                // Kept in pending.json for the next restore; no loop after StopAsync.
+                _watches.TryRemove(new(record.WatchId, watch));
+                watch.DisposeStop();
+                return;
+            }
             watch.Loop = Task.Run(() => RunAsync(watch, settings));
         }
     }
@@ -513,6 +534,7 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
     /// </summary>
     public async Task StopAsync()
     {
+        _stopped = true;
         var watches = _watches.ToArray();
         foreach (var pair in watches) _watches.TryRemove(pair);
         await Task.WhenAll(watches.Select(p => p.Value.StopAsync())).ConfigureAwait(false);

tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs

@@ -462,6 +462,29 @@ public sealed class A2AWatchTests : IDisposable
         Assert.Single(PendingFile.Read(_ws));
     }
 
+    [Fact]
+    public async Task Start_AfterStop_KeepsTheRecordButStartsNoLoop()
+    {
+        await using var remote = await RemoteAgentServer.StartAsync();
+        remote.Faux.Enqueue(FauxTurn.Say("Later"));
+        WriteAgents();
+        var delivery = new RecordingDelivery(_ws);
+        await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = WatchSources.Poll, PollDelays = [TimeSpan.FromMilliseconds(50)] };
+        registry.UseDelivery(delivery);
+        var settings = new A2AClientSettings();
+        using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None);
+        var task = await StartRemoteTaskAsync(agent);
+        await registry.StopAsync(); // shutting down while a tool call hands a task over
+        var record = await registry.StartAsync("1", _ws, agent, task, settings, CancellationToken.None);
+        await registry.RearmAsync(_ws, record with { WatchId = "w2", TaskId = "other" }, settings);
+        Assert.Empty(registry.List("1"));
+        Assert.Equal(2, PendingFile.Read(_ws).Count); // restored on the next start
+        Assert.Equal(0, await registry.RestoreAsync(settings, CancellationToken.None));
+        Assert.Empty(registry.List("1"));
+        await Task.Delay(500);
+        Assert.Empty(delivery.Delivered);
+    }
+
     [Fact]
     public async Task StopWatching_RacingFinish_DoesNotResurrectTheRecord()
     {

tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs

@@ -106,8 +106,9 @@ public sealed class A2AClientHost : IAsyncDisposable
 
     public async ValueTask DisposeAsync()
     {
-        await Registry.DisposeAsync();
+        // Sessions first: a delivery or session-start flush in flight finishes against a live registry.
         foreach (var session in _sessions) await session.DisposeAsync();
+        await Registry.DisposeAsync();
         await _services.DisposeAsync();
         try { Directory.Delete(_root, true); } catch (IOException) { } catch (UnauthorizedAccessException) { }
     }