fix(a2a-client): delivery set before polling, flushes respect stop, restore re-arms first

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

@@ -167,14 +167,18 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
         return Task.FromResult(true);
     }
 
-    /// <summary>Re-arm every persisted watch of the delivery's known targets; delivers results kept undelivered. Returns the count re-armed.</summary>
+    /// <summary>
+    /// 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.
+    /// </summary>
     public async Task<int> 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;
     }
 
-    /// <summary>Deliver results that were kept because the target had no live conversation.</summary>
+    /// <summary>Deliver results that were kept because the target had no live conversation. After <see cref="StopAsync"/> they stay kept.</summary>
     public async Task FlushUndeliveredAsync(string target, string workspace, CancellationToken cancellationToken)
     {
+        if (_stopped) return;
         var claimed = new List<WatchRecord>();
         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);

hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AClientSetup.cs

@@ -8,7 +8,10 @@ namespace Volcanic.AgenticCore.Telegram.A2A;
 /// The A2A client (off unless <c>A2AClient__Enabled=true</c>, independent of the server's <c>A2A__Enabled</c>). Only the
 /// per-chat sessions get <see cref="A2AClientExtension"/> (<see cref="Chats.ChatRegistry"/>), never A2A server sessions.
 /// Call before <see cref="TelegramA2ASetup.Register"/>, 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 <see cref="A2AWatchHostedService"/> must point the
+/// registry at the chats before the first chat session (else early results go through the registry's live-session
+/// default).
 /// </summary>
 public static class TelegramA2AClientSetup
 {

hosts/Volcanic.AgenticCore.Telegram/Program.cs

@@ -104,6 +104,10 @@ builder.Services.AddSingleton<ChatCommands>();
 builder.Services.AddSingleton<InboundMedia>();
 builder.Services.AddSingleton<InboundComposer>();
 builder.Services.AddSingleton<IUpdateRouter, UpdateRouter>();
+// 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<PollingService>();
 builder.Services.AddHostedService(sp => sp.GetRequiredService<PollingService>());
 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<IScriptEnvironment, CalendarScriptEnvironment>();
 builder.Services.AddHostedService<BotSetupService>();
-// 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();

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()
     {