diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs index 6184d0c..8193e52 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs @@ -167,14 +167,18 @@ public sealed class A2AWatchRegistry : IAsyncDisposable return Task.FromResult(true); } - /// Re-arm every persisted watch of the delivery's known targets; delivers results kept undelivered. Returns the count re-armed. + /// + /// Re-arm every persisted watch of the delivery's known targets, then deliver results kept undelivered (all watches + /// first: a delivery may run a whole chat turn). Returns the count re-armed. + /// public async Task RestoreAsync(A2AClientSettings settings, CancellationToken cancellationToken) { var count = 0; + var targets = new List<(string Target, string Workspace)>(); foreach (var target in _delivery.KnownTargets().Distinct(StringComparer.Ordinal)) + if (_delivery.WorkspaceFor(target) is { } workspace) targets.Add((target, workspace)); + foreach (var (target, workspace) in targets) { - if (_delivery.WorkspaceFor(target) is not { } workspace) continue; - await FlushUndeliveredAsync(target, workspace, cancellationToken).ConfigureAwait(false); using (await LockAsync(workspace, cancellationToken).ConfigureAwait(false)) { foreach (var stored in PendingFile.Read(workspace).Where(r => r.Target == target && r.Undelivered is null)) @@ -193,12 +197,15 @@ public sealed class A2AWatchRegistry : IAsyncDisposable } } } + foreach (var (target, workspace) in targets) + await FlushUndeliveredAsync(target, workspace, cancellationToken).ConfigureAwait(false); return count; } - /// Deliver results that were kept because the target had no live conversation. + /// Deliver results that were kept because the target had no live conversation. After they stay kept. public async Task FlushUndeliveredAsync(string target, string workspace, CancellationToken cancellationToken) { + if (_stopped) return; var claimed = new List(); using (await LockAsync(workspace, cancellationToken).ConfigureAwait(false)) { @@ -214,6 +221,7 @@ public sealed class A2AWatchRegistry : IAsyncDisposable { foreach (var record in claimed) { + if (_stopped) break; // shutting down: the rest stay in pending.json for the next start if (!await TryDeliverAsync(target, record.Undelivered!, cancellationToken).ConfigureAwait(false)) { failed = true; continue; } using (await LockAsync(workspace).ConfigureAwait(false)) PendingFile.Remove(workspace, record.WatchId); diff --git a/hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AClientSetup.cs b/hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AClientSetup.cs index db77e6b..6dc5d5c 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AClientSetup.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AClientSetup.cs @@ -8,7 +8,10 @@ namespace Volcanic.AgenticCore.Telegram.A2A; /// The A2A client (off unless A2AClient__Enabled=true, independent of the server's A2A__Enabled). Only the /// per-chat sessions get (), never A2A server sessions. /// Call before , which copies the server options: when the server is on, its push -/// receiver is routed into the watch registry and its public URL becomes the push base URL. +/// receiver is routed into the watch registry and its public URL becomes the push base URL. Call before the polling +/// service is registered too: hosted services start in order, and must point the +/// registry at the chats before the first chat session (else early results go through the registry's live-session +/// default). /// public static class TelegramA2AClientSetup { diff --git a/hosts/Volcanic.AgenticCore.Telegram/Program.cs b/hosts/Volcanic.AgenticCore.Telegram/Program.cs index cfc46d8..51ffa0b 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Program.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Program.cs @@ -104,6 +104,10 @@ builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddSingleton(); +// Before PollingService (hosted services start in registration order): the watch hosted service points the registry at +// the chats before the first chat session exists. Before the server setup too, which copies a2a: this sets a2a.OnPush +// (push receiver -> watches) and the client's PushBaseUrl. +TelegramA2AClientSetup.Register(builder.Services, a2aClient, a2a, A2AWatchRegistry.Shared); builder.Services.AddSingleton(); builder.Services.AddHostedService(sp => sp.GetRequiredService()); if (scheduling.Enabled) SchedulerSetup.Register(builder.Services); // the hosted service starts after PollingService @@ -111,8 +115,6 @@ if (options.Triggers.IsEnabled) TriggersSetup.Register(builder.Services, options // Watchers get a short-lived GOOGLE_ACCESS_TOKEN for their chat's calendar, never the client secret or refresh token. if (scheduling.Enabled && calendar is not null) builder.Services.AddSingleton(); builder.Services.AddHostedService(); -// Before the server setup copies a2a: sets a2a.OnPush (push receiver -> watches) and the client's PushBaseUrl. -TelegramA2AClientSetup.Register(builder.Services, a2aClient, a2a, A2AWatchRegistry.Shared); TelegramA2ASetup.Register(builder.Services, options, a2a); await using var host = builder.Build(); diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs index 5896b32..3407f8a 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs @@ -485,6 +485,45 @@ public sealed class A2AWatchTests : IDisposable Assert.Empty(delivery.Delivered); } + [Fact] + public async Task Flush_AfterStop_KeepsTheResults() + { + PendingFile.Upsert(_ws, new WatchRecord("w1", "1", "remote", "t1", "c1", null, DateTimeOffset.UtcNow, WatchSources.Poll, Undelivered: "kept")); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, new HttpClientHandler()); + registry.UseDelivery(delivery); + await registry.StopAsync(); + await registry.FlushUndeliveredAsync("1", _ws, CancellationToken.None); + Assert.Empty(delivery.Delivered); + Assert.Equal("kept", Assert.Single(PendingFile.Read(_ws)).Undelivered); + } + + [Fact] + public async Task Restore_RearmsWatchesBeforeFlushingKeptResults() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("slow").WithDelay(TimeSpan.FromSeconds(5))); + WriteAgents(); + var settings = new A2AClientSettings(); + AgentTask task; + await using (var first = new A2AWatchRegistry(TimeProvider.System, remote.Handler)) + { + using var agent = await first.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + task = await StartRemoteTaskAsync(agent); + } + PendingFile.Upsert(_ws, new WatchRecord("w1", "1", "remote", task.Id, task.ContextId, null, DateTimeOffset.UtcNow, WatchSources.Poll)); + PendingFile.Upsert(_ws, new WatchRecord("w2", "1", "remote", "done", "c2", null, DateTimeOffset.UtcNow, WatchSources.Poll, Undelivered: "kept")); + var delivery = new GatedDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { PollDelays = [TimeSpan.FromSeconds(10)] }; + registry.UseDelivery(delivery); + var restore = registry.RestoreAsync(settings, CancellationToken.None); + await WaitUntilAsync(() => delivery.Entered >= 1); // a slow chat turn delivering the kept result + Assert.Equal("w1", Assert.Single(registry.List("1")).WatchId); // already re-armed + delivery.Release(); + Assert.Equal(1, await restore); + Assert.Equal("kept", Assert.Single(delivery.Delivered)); + } + [Fact] public async Task StopWatching_RacingFinish_DoesNotResurrectTheRecord() {