diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AResultDelivery.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AResultDelivery.cs new file mode 100644 index 0000000..0615906 --- /dev/null +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AResultDelivery.cs @@ -0,0 +1,49 @@ +using System.Collections.Concurrent; +using Volcanic.AgenticCore.Abstractions.Extensions; + +namespace Volcanic.AgenticCore.Extensions.A2A; + +/// Where finished remote tasks go. Targets are host ids (Telegram: chat ids). +public interface IA2AResultDelivery +{ + /// Run as a follow-up prompt in the target's conversation. False: not deliverable now. + Task DeliverAsync(string target, string text, CancellationToken cancellationToken); + + /// The target's workspace. A path computation: must not create or wake a session. + string? WorkspaceFor(string target); + + /// Targets whose workspace may hold a pending.json (restore at startup). + IEnumerable KnownTargets(); +} + +/// Default delivery: the target's live session, if any (registered on session start, removed on shutdown). +public sealed class LiveSessionDelivery : IA2AResultDelivery +{ + private readonly ConcurrentDictionary _live = new(StringComparer.Ordinal); + + public void Attach(string target, string workspace, IExtensionApi api) => _live[target] = (api, workspace); + + /// Detaches only if it is still the target's live session (a newer one may have attached). + public void Detach(string target, IExtensionApi api) + { + // TryRemove(pair) removes only if the value is still this exact entry, so a racing Attach is never undone. + if (_live.TryGetValue(target, out var entry) && ReferenceEquals(entry.Api, api)) _live.TryRemove(new(target, entry)); + } + + public Task DeliverAsync(string target, string text, CancellationToken cancellationToken) + { + if (!_live.TryGetValue(target, out var entry)) return Task.FromResult(false); + try + { + entry.Api.SendUserMessage(text, null, MessageDelivery.FollowUp); + return Task.FromResult(true); + } + catch (Exception ex) when (ex is InvalidOperationException or ObjectDisposedException) + { + return Task.FromResult(false); + } + } + + public string? WorkspaceFor(string target) => _live.TryGetValue(target, out var entry) ? entry.Workspace : null; + public IEnumerable KnownTargets() => _live.Keys.ToList(); +} diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs new file mode 100644 index 0000000..8b30c74 --- /dev/null +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs @@ -0,0 +1,315 @@ +using System.Collections.Concurrent; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using System.Threading.Channels; +using A2A; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Volcanic.AgenticCore.Extensions.A2A; + +/// +/// Process-wide (): one HTTP handler for every remote call, and the watches of long remote tasks. +/// A watch follows its task by push (when the card supports it and the host gave us a push base URL), else by +/// subscribing (streaming agents), else by polling; it delivers the outcome through and +/// is persisted in the workspace's pending.json until delivered. +/// +public sealed class A2AWatchRegistry : IAsyncDisposable +{ + public static A2AWatchRegistry Shared { get; } = new(TimeProvider.System, + new SocketsHttpHandler { PooledConnectionLifetime = TimeSpan.FromMinutes(5) }); + + public static readonly TimeSpan[] DefaultPollDelays = + [TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(30), TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(2), TimeSpan.FromMinutes(5)]; + + private static readonly TimeSpan[] ReconnectDelays = [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(15)]; + + private readonly HttpMessageHandler _handler; + private readonly ConcurrentDictionary _watches = new(StringComparer.Ordinal); + private volatile IA2AResultDelivery _delivery; + + public A2AWatchRegistry(TimeProvider time, HttpMessageHandler handler, ILogger? logger = null) + { + Time = time; + _handler = handler; + Logger = logger ?? NullLogger.Instance; + _delivery = Live; + } + + public TimeProvider Time { get; } + public ILogger Logger { get; set; } + public LiveSessionDelivery Live { get; } = new(); + + internal string? SourceOverride { get; init; } + internal TimeSpan[] PollDelays { get; init; } = DefaultPollDelays; + internal TimeSpan SafetyPollInterval { get; init; } = TimeSpan.FromMinutes(10); + + public void UseDelivery(IA2AResultDelivery delivery) => _delivery = delivery; + + internal RemoteAgentConnector Connector(A2AClientSettings settings) => + new(_handler, TimeSpan.FromSeconds(settings.CardTimeoutSeconds)); + + internal static TimeSpan DelayAt(TimeSpan[] delays, int attempt) => delays[Math.Min(attempt, delays.Length - 1)]; + + public IReadOnlyList List(string target) => + _watches.Values.Where(w => w.Record.Target == target).Select(w => w.Record).ToList(); + + /// Hand a still-running remote task to a watch. + internal async Task StartAsync(string target, string workspace, RemoteAgent agent, AgentTask task, + A2AClientSettings settings, CancellationToken cancellationToken) + { + var watchId = Guid.NewGuid().ToString("N"); + string? token = null; + var source = SourceOverride ?? (agent.Streaming ? WatchSources.Subscribe : WatchSources.Poll); + if (SourceOverride is null && settings.PushBaseUrl is { Length: > 0 } pushBase && agent.SupportsPush) + { + token = Convert.ToHexString(RandomNumberGenerator.GetBytes(32)); + try + { + await agent.Client.CreateTaskPushNotificationConfigAsync(new CreateTaskPushNotificationConfigRequest + { + TaskId = task.Id, + ConfigId = watchId, + Config = new PushNotificationConfig { Id = watchId, Url = $"{pushBase.TrimEnd('/')}/a2a/push/{watchId}", Token = token }, + }, cancellationToken).ConfigureAwait(false); + source = WatchSources.Push; + } + catch (Exception ex) when (ex is A2AException or HttpRequestException) + { + Logger.LogInformation("A2A: {Agent} refused push for task {TaskId} ({Error}); watching by {Source}", agent.Entry.Name, task.Id, ex.Message, source); + token = null; + } + } + var record = new WatchRecord(watchId, target, agent.Entry.Name, task.Id, task.ContextId, token, Time.GetUtcNow(), source); + PendingFile.Upsert(workspace, record); + Arm(record, workspace, settings); + return record; + } + + /// + /// Push receiver: a known watch with the right token wakes up and checks its task (the body is never trusted). + /// False → 404. + /// + public Task OnPushAsync(string watchId, string? token, JsonElement body, CancellationToken cancellationToken) + { + if (!_watches.TryGetValue(watchId, out var watch) || watch.Record.PushToken is not { } expected || token is null || + !CryptographicOperations.FixedTimeEquals(Encoding.UTF8.GetBytes(expected), Encoding.UTF8.GetBytes(token))) + return Task.FromResult(false); + watch.Wake.Writer.TryWrite(true); + return Task.FromResult(true); + } + + /// Re-arm every persisted watch of the delivery's known targets; delivers results kept undelivered. Returns the count re-armed. + public async Task RestoreAsync(A2AClientSettings settings, CancellationToken cancellationToken) + { + var count = 0; + foreach (var target in _delivery.KnownTargets().Distinct(StringComparer.Ordinal)) + { + if (_delivery.WorkspaceFor(target) is not { } workspace) continue; + await FlushUndeliveredAsync(target, workspace, cancellationToken).ConfigureAwait(false); + foreach (var record in PendingFile.Read(workspace).Where(r => r.Target == target && r.Undelivered is null)) + { + if (Arm(record, workspace, settings)) count++; + } + } + return count; + } + + /// Deliver results that were kept because the target had no live conversation. + public async Task FlushUndeliveredAsync(string target, string workspace, CancellationToken cancellationToken) + { + foreach (var record in PendingFile.Read(workspace).Where(r => r.Target == target && r.Undelivered is not null)) + if (await TryDeliverAsync(target, record.Undelivered!, cancellationToken).ConfigureAwait(false)) + PendingFile.Remove(workspace, record.WatchId); + } + + /// Stop watching a task (after cancel); true when a watch existed. + public async Task StopWatchingAsync(string target, string workspace, string taskId) + { + var watch = _watches.Values.FirstOrDefault(w => w.Record.Target == target && w.Record.TaskId == taskId); + if (watch is not null) + { + _watches.TryRemove(watch.Record.WatchId, out _); + await watch.StopAsync().ConfigureAwait(false); + } + foreach (var record in PendingFile.Read(workspace).Where(r => r.Target == target && r.TaskId == taskId)) + PendingFile.Remove(workspace, record.WatchId); + return watch is not null; + } + + private bool Arm(WatchRecord record, string workspace, A2AClientSettings settings) + { + var watch = new Watch(record, workspace); + if (!_watches.TryAdd(record.WatchId, watch)) return false; + watch.Loop = Task.Run(() => RunAsync(watch, settings)); + return true; + } + + private async Task RunAsync(Watch watch, A2AClientSettings settings) + { + var record = watch.Record; + var ct = watch.Stop.Token; + var deadline = record.StartedAt + settings.WatchTimeout; + var source = record.Source; + var failures = 0; + var polls = 0; + try + { + while (!ct.IsCancellationRequested) + { + if (Time.GetUtcNow() >= deadline) + { + await FinishAsync(watch, $"[A2A result from \"{record.Agent}\", task {record.TaskId} (no result)]\n" + + $"\"{record.Agent}\" gave no result within {settings.WatchTimeout}. Check it with action=status, agent={record.Agent}, taskId={record.TaskId}.", null).ConfigureAwait(false); + return; + } + if (AgentsFile.Find(watch.Workspace, record.Agent) is not { } entry) + { + await FinishAsync(watch, $"[A2A result from \"{record.Agent}\", task {record.TaskId} (stopped)]\n" + + $"\"{record.Agent}\" is no longer in agents.json, so its task is no longer watched.", null).ConfigureAwait(false); + return; + } + + TimeSpan wait; + try + { + using var agent = await Connector(settings).ConnectAsync(entry, ct).ConfigureAwait(false); + var task = source == WatchSources.Subscribe + ? await SubscribeAsync(agent, record.TaskId, deadline, ct).ConfigureAwait(false) + : await agent.Client.GetTaskAsync(new GetTaskRequest { Id = record.TaskId }, ct).ConfigureAwait(false); + if (task is not null && A2AResults.IsSettled(task.Status.State)) + { + var saved = A2AResults.SaveFiles(watch.Workspace, record.Agent, task, settings.MaxFileBytes); + await FinishAsync(watch, A2AResults.DeliveryText(record.Agent, task, saved.Saved, saved.Skipped), agent).ConfigureAwait(false); + return; + } + failures = 0; + wait = source switch + { + WatchSources.Push => SafetyPollInterval, + WatchSources.Subscribe => DelayAt(ReconnectDelays, 0), + _ => DelayAt(PollDelays, polls++), + }; + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + failures++; + Logger.LogInformation("A2A: watching task {TaskId} of {Agent} failed ({Error})", record.TaskId, record.Agent, ex.Message); + if (source == WatchSources.Subscribe && failures >= 3) source = WatchSources.Poll; + wait = source == WatchSources.Subscribe ? DelayAt(ReconnectDelays, failures) : DelayAt(PollDelays, polls++); + } + await WaitAsync(watch, Min(wait, deadline - Time.GetUtcNow()), ct).ConfigureAwait(false); + } + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + } + catch (Exception ex) + { + Logger.LogWarning(ex, "A2A: watch {WatchId} crashed", record.WatchId); + } + } + + /// + /// Follows the task's stream until it settles, the stream ends, or the watch deadline passes; then reads the task. + /// + private async Task SubscribeAsync(RemoteAgent agent, string taskId, DateTimeOffset deadline, CancellationToken ct) + { + AgentTask? current = null; + var remaining = deadline - Time.GetUtcNow(); + using var until = CancellationTokenSource.CreateLinkedTokenSource(ct); + if (remaining > TimeSpan.Zero) until.CancelAfter(remaining); + else await until.CancelAsync().ConfigureAwait(false); + try + { + await foreach (var ev in agent.Client.SubscribeToTaskAsync(new SubscribeToTaskRequest { Id = taskId }, until.Token).ConfigureAwait(false)) + { + current = TaskProjection.Apply(current, ev); + if (current is not null && A2AResults.IsSettled(current.Status.State)) return current; + } + } + catch (A2AException) // e.g. the task is already terminal: read it instead + { + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) // the deadline: the loop reports it + { + } + return await agent.Client.GetTaskAsync(new GetTaskRequest { Id = taskId }, ct).ConfigureAwait(false); + } + + /// Waits , or less when a push wakes the watch. + private async Task WaitAsync(Watch watch, TimeSpan wait, CancellationToken ct) + { + if (wait <= TimeSpan.Zero) return; + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); + var delay = Task.Delay(wait, Time, timeout.Token); + var woken = watch.Wake.Reader.WaitToReadAsync(timeout.Token).AsTask(); + await Task.WhenAny(delay, woken).ConfigureAwait(false); + await timeout.CancelAsync().ConfigureAwait(false); + while (watch.Wake.Reader.TryRead(out _)) { } + ct.ThrowIfCancellationRequested(); + } + + private async Task FinishAsync(Watch watch, string text, RemoteAgent? agent) + { + // Stopped (cancel) in the meantime: the user no longer wants this result. + if (!_watches.TryRemove(new(watch.Record.WatchId, watch))) return; + if (agent is not null && watch.Record.PushToken is not null) + { + try + { + await agent.Client.DeleteTaskPushNotificationConfigAsync(new DeleteTaskPushNotificationConfigRequest + { TaskId = watch.Record.TaskId, Id = watch.Record.WatchId }, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) when (ex is A2AException or HttpRequestException) { } + } + if (await TryDeliverAsync(watch.Record.Target, text, CancellationToken.None).ConfigureAwait(false)) + PendingFile.Remove(watch.Workspace, watch.Record.WatchId); + else + PendingFile.Upsert(watch.Workspace, watch.Record with { Undelivered = text }); + } + + private async Task TryDeliverAsync(string target, string text, CancellationToken ct) + { + try + { + return await _delivery.DeliverAsync(target, text, ct).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + Logger.LogWarning(ex, "A2A: delivering a result to {Target} failed", target); + return false; + } + } + + private static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b; + + /// Stops every watch (the records stay in pending.json for the next restore). + public async Task StopAsync() + { + foreach (var watch in _watches.Values) await watch.StopAsync().ConfigureAwait(false); + _watches.Clear(); + } + + public async ValueTask DisposeAsync() => await StopAsync().ConfigureAwait(false); + + private sealed class Watch(WatchRecord record, string workspace) + { + public WatchRecord Record => record; + public string Workspace => workspace; + public CancellationTokenSource Stop { get; } = new(); + public Channel Wake { get; } = Channel.CreateBounded(new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.DropOldest }); + public Task Loop { get; set; } = Task.CompletedTask; + + public async Task StopAsync() + { + await Stop.CancelAsync().ConfigureAwait(false); + try { await Loop.ConfigureAwait(false); } catch (OperationCanceledException) { } + } + } +} diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs new file mode 100644 index 0000000..1f05d18 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs @@ -0,0 +1,228 @@ +using System.Collections.Concurrent; +using System.Text.Json; +using A2A; +using Volcanic.AgenticCore.Extensions.A2A; +using Volcanic.AgenticCore.Extensions.Tests.A2A.Support; +using Volcanic.AgenticCore.Testing; +using A2AMessage = global::A2A.Message; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A; + +public sealed class A2AWatchTests : IDisposable +{ + private readonly string _ws = Path.Combine(Path.GetTempPath(), "a2a-watch", Guid.NewGuid().ToString("N")); + + private sealed class RecordingDelivery(string workspace) : IA2AResultDelivery + { + public ConcurrentQueue<(string Target, string Text)> Delivered { get; } = new(); + public volatile bool Accept = true; + public Task DeliverAsync(string target, string text, CancellationToken cancellationToken) + { + if (Accept) Delivered.Enqueue((target, text)); + return Task.FromResult(Accept); + } + public string? WorkspaceFor(string target) => workspace; + public IEnumerable KnownTargets() => ["1"]; + } + + private void WriteAgents(string key = RemoteAgentServer.Key) + { + Directory.CreateDirectory(Path.GetDirectoryName(AgentsFile.PathFor(_ws))!); + File.WriteAllText(AgentsFile.PathFor(_ws), $$"""[ { "name": "remote", "url": "{{RemoteAgentServer.BaseUrl}}", "key": "{{key}}" } ]"""); + } + + private static async Task StartRemoteTaskAsync(RemoteAgent agent, string text = "go") => + (await agent.Client.SendMessageAsync(new SendMessageRequest + { + Message = new A2AMessage { Role = Role.User, MessageId = Guid.NewGuid().ToString("N"), Parts = [Part.FromText(text)] }, + Configuration = new SendMessageConfiguration { ReturnImmediately = true }, + })).Task!; + + private static async Task WaitUntilAsync(Func condition, int timeoutMs = 15_000) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException("condition not met"); + await Task.Delay(20); + } + } + + [Theory] + [InlineData(WatchSources.Subscribe)] + [InlineData(WatchSources.Poll)] + public async Task Watch_DeliversResult(string source) + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Trip booked").WithDelay(TimeSpan.FromMilliseconds(500))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = source, PollDelays = [TimeSpan.FromMilliseconds(100)] }; + 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); + var record = await registry.StartAsync("1", _ws, agent, task, settings, CancellationToken.None); + Assert.Equal(source, record.Source); + Assert.Contains(PendingFile.Read(_ws), r => r.WatchId == record.WatchId); // persisted before the watch can finish + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); + var (target, text) = Assert.Single(delivery.Delivered); + Assert.Equal("1", target); + Assert.StartsWith($"[A2A result from \"remote\", task {task.Id} (completed)]", text); + Assert.Contains("Trip booked", text); + await WaitUntilAsync(() => PendingFile.Read(_ws).Count == 0); + Assert.Empty(registry.List("1")); + } + + [Fact] + public async Task Watch_Push_DeliversFollowUp() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Pushed result").WithDelay(TimeSpan.FromMilliseconds(500))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SafetyPollInterval = TimeSpan.FromMinutes(10) }; + registry.UseDelivery(delivery); + var push = new PushToRegistryHandler(registry); + remote.PushHandler = push; + var settings = new A2AClientSettings { PushBaseUrl = "http://local.test" }; + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + var task = await StartRemoteTaskAsync(agent); + var record = await registry.StartAsync("1", _ws, agent, task, settings, CancellationToken.None); + Assert.Equal(WatchSources.Push, record.Source); + Assert.NotNull(record.PushToken); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); + Assert.True(push.Received > 0); + Assert.Contains("Pushed result", Assert.Single(delivery.Delivered).Text); + var configs = await agent.Client.ListTaskPushNotificationConfigAsync(new ListTaskPushNotificationConfigRequest { TaskId = task.Id }); + Assert.DoesNotContain(configs.Configs ?? [], c => c.Id == record.WatchId); // removed once the result was delivered + } + + [Fact] + public async Task Watch_PushBadToken_Rejected() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("slow").WithDelay(TimeSpan.FromSeconds(5))); + WriteAgents(); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler); + registry.UseDelivery(new RecordingDelivery(_ws)); + var settings = new A2AClientSettings { PushBaseUrl = "http://local.test" }; + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + var record = await registry.StartAsync("1", _ws, agent, await StartRemoteTaskAsync(agent), settings, CancellationToken.None); + var body = JsonDocument.Parse("{}").RootElement; + Assert.False(await registry.OnPushAsync(record.WatchId, "wrong", body, CancellationToken.None)); + Assert.False(await registry.OnPushAsync(record.WatchId, null, body, CancellationToken.None)); + Assert.False(await registry.OnPushAsync("unknown", record.PushToken, body, CancellationToken.None)); + Assert.True(await registry.OnPushAsync(record.WatchId, record.PushToken, body, CancellationToken.None)); + } + + [Theory] + [InlineData(WatchSources.Poll)] + [InlineData(WatchSources.Subscribe)] + public async Task Watch_Timeout_DeliversNotice(string source) + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("never").WithDelay(TimeSpan.FromSeconds(10))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = source, PollDelays = [TimeSpan.FromMilliseconds(100)] }; + registry.UseDelivery(delivery); + var settings = new A2AClientSettings { WatchTimeout = TimeSpan.FromMilliseconds(400) }; + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + await registry.StartAsync("1", _ws, agent, await StartRemoteTaskAsync(agent), settings, CancellationToken.None); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty, timeoutMs: 5_000); // well before the remote's 10 s answer + Assert.Contains("gave no result", Assert.Single(delivery.Delivered).Text); + } + + [Fact] + public async Task Watch_AgentRemoved_EndsWatch() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("never").WithDelay(TimeSpan.FromSeconds(10))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = WatchSources.Poll, PollDelays = [TimeSpan.FromMilliseconds(100)] }; + registry.UseDelivery(delivery); + var settings = new A2AClientSettings(); + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + await registry.StartAsync("1", _ws, agent, await StartRemoteTaskAsync(agent), settings, CancellationToken.None); + File.WriteAllText(AgentsFile.PathFor(_ws), "[]"); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); + Assert.Contains("no longer in agents.json", Assert.Single(delivery.Delivered).Text); + Assert.Empty(registry.List("1")); + } + + [Fact] + public async Task Watch_Undelivered_KeptThenFlushed() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Later")); + WriteAgents(); + var delivery = new RecordingDelivery(_ws) { Accept = false }; + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = WatchSources.Poll, PollDelays = [TimeSpan.FromMilliseconds(100)] }; + registry.UseDelivery(delivery); + var settings = new A2AClientSettings(); + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + await registry.StartAsync("1", _ws, agent, await StartRemoteTaskAsync(agent), settings, CancellationToken.None); + await WaitUntilAsync(() => PendingFile.Read(_ws).Any(r => r.Undelivered is not null)); + delivery.Accept = true; + await registry.FlushUndeliveredAsync("1", _ws, CancellationToken.None); + Assert.Contains("Later", Assert.Single(delivery.Delivered).Text); + Assert.Empty(PendingFile.Read(_ws)); + } + + [Fact] + public async Task Watch_Restore_FromPendingFile() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Done before restart")); + 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)); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { PollDelays = [TimeSpan.FromMilliseconds(100)] }; + registry.UseDelivery(delivery); + Assert.Equal(1, await registry.RestoreAsync(settings, CancellationToken.None)); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); + Assert.Contains("Done before restart", Assert.Single(delivery.Delivered).Text); + await WaitUntilAsync(() => PendingFile.Read(_ws).Count == 0); + } + + [Fact] + public async Task StopWatching_EndsTheWatchAndForgetsIt() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("never").WithDelay(TimeSpan.FromSeconds(10))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = WatchSources.Poll, PollDelays = [TimeSpan.FromMilliseconds(100)] }; + 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.StartAsync("1", _ws, agent, task, settings, CancellationToken.None); + Assert.Single(registry.List("1")); + Assert.True(await registry.StopWatchingAsync("1", _ws, task.Id)); + Assert.Empty(registry.List("1")); + Assert.Empty(PendingFile.Read(_ws)); + Assert.False(await registry.StopWatchingAsync("1", _ws, task.Id)); + Assert.Empty(delivery.Delivered); + } + + [Fact] + public void PollDelays_BackOffToFiveMinutes() + { + var delays = A2AWatchRegistry.DefaultPollDelays; + Assert.Equal(TimeSpan.FromSeconds(15), delays[0]); + Assert.Equal(TimeSpan.FromMinutes(5), delays[^1]); + Assert.Equal(TimeSpan.FromMinutes(5), A2AWatchRegistry.DelayAt(delays, 99)); + } + + public void Dispose() { try { Directory.Delete(_ws, true); } catch (IOException) { } } +} diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/PushToRegistryHandler.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/PushToRegistryHandler.cs new file mode 100644 index 0000000..41343b3 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/PushToRegistryHandler.cs @@ -0,0 +1,24 @@ +using System.Net; +using System.Text.Json; +using Volcanic.AgenticCore.Extensions.A2A; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A.Support; + +/// Stands in for our push receiver endpoint: POST …/a2a/push/{watchId} → . +public sealed class PushToRegistryHandler(A2AWatchRegistry? registry) : HttpMessageHandler +{ + private int _received; + + public int Received => Volatile.Read(ref _received); + + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + Interlocked.Increment(ref _received); + if (registry is null) return new HttpResponseMessage(HttpStatusCode.OK); + var watchId = request.RequestUri!.Segments[^1]; + var token = request.Headers.TryGetValues("X-A2A-Notification-Token", out var v) ? v.Single() : null; + using var doc = JsonDocument.Parse(await request.Content!.ReadAsStringAsync(cancellationToken)); + var ok = await registry.OnPushAsync(watchId, token, doc.RootElement.Clone(), cancellationToken); + return new HttpResponseMessage(ok ? HttpStatusCode.OK : HttpStatusCode.NotFound); + } +}