diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs index 3cc5560..6184d0c 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs @@ -55,6 +55,9 @@ public sealed class A2AWatchRegistry : IAsyncDisposable private readonly ConcurrentDictionary _flushRequested = new(StringComparer.Ordinal); private volatile IA2AResultDelivery _delivery; + /// Set by : later starts and re-arms still persist their record but run no loop. + 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 /// 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); diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs index 4d2dd57..5896b32 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs +++ b/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() { diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs index 32349bd..e2a1883 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs +++ b/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) { } }