admin/VolcanicAgenticCore
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(a2a-client): sturdier watches — transient agents.json errors, bounded remote calls, single delivery

Browse Source
This commit is contained in:
EugeneTes
2026-10-08 13:15:40 +00:00
parent 8ba99b1477
commit 24e2ad9877
3 changed files with 431 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

272
extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs
View File

@@ -15,6 +15,12 @@ namespace Volcanic.AgenticCore.Extensions.A2A;
/// subscribing (streaming agents), else by polling; it delivers the outcome through <see cref="IA2AResultDelivery"/> and
/// is persisted in the workspace's <c>pending.json</c> until delivered.
/// </summary>
/// <remarks>
/// Delivery is at least once: a result is delivered, then removed from <c>pending.json</c>, so a crash in between
/// delivers it again after the restart. Within a process, a per-workspace lock serialises finishing, flushing, restoring
/// and stopping, so one result is delivered once and a cancelled watch's record is never written back.
/// A push only means "check now": its body is never trusted.
/// </remarks>
public sealed class A2AWatchRegistry : IAsyncDisposable
{
public static A2AWatchRegistry Shared { get; } = new(TimeProvider.System,
@@ -23,10 +29,17 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
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)];
/// <summary>Longest single subscription; the stream is then reopened (a pause, not a failure).</summary>
private static readonly TimeSpan MaxSubscription = TimeSpan.FromDays(1);
private static readonly TimeSpan DeleteTimeout = TimeSpan.FromSeconds(10);
/// <summary>After this many failed checks in a row, a subscribe watch polls instead.</summary>
private const int SubscribeFailuresBeforePoll = 3;
private readonly HttpMessageHandler _handler;
private readonly ConcurrentDictionary<string, Watch> _watches = new(StringComparer.Ordinal);
private readonly ConcurrentDictionary<string, SemaphoreSlim> _workspaceLocks = new(StringComparer.Ordinal);
private volatile IA2AResultDelivery _delivery;
public A2AWatchRegistry(TimeProvider time, HttpMessageHandler handler, ILogger? logger = null)
@@ -43,14 +56,18 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
internal string? SourceOverride { get; init; }
internal TimeSpan[] PollDelays { get; init; } = DefaultPollDelays;
internal TimeSpan[] ReconnectDelays { get; init; } = [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(15)];
internal TimeSpan SafetyPollInterval { get; init; } = TimeSpan.FromMinutes(10);
/// <summary>Bound of one remote call (get, create/delete push config), never past the watch deadline.</summary>
internal TimeSpan CallTimeout { get; init; } = TimeSpan.FromSeconds(60);
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)];
internal static TimeSpan DelayAt(TimeSpan[] delays, int attempt) => delays[Math.Clamp(attempt, 0, delays.Length - 1)];
public IReadOnlyList<WatchRecord> List(string target) =>
_watches.Values.Where(w => w.Record.Target == target).Select(w => w.Record).ToList();
@@ -60,31 +77,45 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
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));
var watch = new Watch(new WatchRecord(watchId, target, agent.Entry.Name, task.Id, task.ContextId, null, Time.GetUtcNow(), source), workspace);
// Reserved before the push config exists, so an early push finds its watch (it wakes it once the loop runs).
_watches[watchId] = watch;
try
{
if (SourceOverride is null && settings.PushBaseUrl is { Length: > 0 } pushBase && agent.SupportsPush)
{
var token = Convert.ToHexString(RandomNumberGenerator.GetBytes(32));
watch.Record = watch.Record with { PushToken = token };
try
{
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
timeout.CancelAfter(CallTimeout);
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;
}, timeout.Token).ConfigureAwait(false);
watch.Record = watch.Record with { Source = WatchSources.Push };
}
catch (Exception ex) when (ex is A2AException or HttpRequestException)
catch (Exception ex) when (ex is not OutOfMemoryException && !cancellationToken.IsCancellationRequested)
{
Logger.LogInformation("A2A: {Agent} refused push for task {TaskId} ({Error}); watching by {Source}", agent.Entry.Name, task.Id, ex.Message, source);
token = null;
watch.Record = watch.Record with { PushToken = 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;
using (await LockAsync(workspace).ConfigureAwait(false))
PendingFile.Upsert(workspace, watch.Record);
}
catch
{
_watches.TryRemove(new(watchId, watch));
watch.DisposeStop();
throw;
}
watch.Loop = Task.Run(() => RunAsync(watch, settings));
return watch.Record;
}
/// <summary>
@@ -108,9 +139,20 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
{
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))
using (await LockAsync(workspace, cancellationToken).ConfigureAwait(false))
{
if (Arm(record, workspace, settings)) count++;
foreach (var stored in PendingFile.Read(workspace).Where(r => r.Target == target && r.Undelivered is null))
{
var record = stored;
// No push receiver now: nobody would wake a push watch, so follow the task another way.
if (record.Source == WatchSources.Push && string.IsNullOrEmpty(settings.PushBaseUrl))
record = record with { Source = WatchSources.Subscribe };
var watch = new Watch(record, workspace);
if (!_watches.TryAdd(record.WatchId, watch)) { watch.DisposeStop(); continue; }
if (!ReferenceEquals(record, stored)) PendingFile.Upsert(workspace, record);
watch.Loop = Task.Run(() => RunAsync(watch, settings));
count++;
}
}
}
return count;
@@ -118,66 +160,81 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
/// <summary>Deliver results that were kept because the target had no live conversation.</summary>
public async Task FlushUndeliveredAsync(string target, string workspace, CancellationToken cancellationToken)
{
using (await LockAsync(workspace, cancellationToken).ConfigureAwait(false))
{
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);
}
}
/// <summary>Stop watching a task (after <c>cancel</c>); true when a watch existed.</summary>
public async Task<bool> 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)
Watch? watch;
using (await LockAsync(workspace).ConfigureAwait(false))
{
_watches.TryRemove(watch.Record.WatchId, out _);
await watch.StopAsync().ConfigureAwait(false);
}
watch = _watches.Values.FirstOrDefault(w => w.Record.Target == target && w.Record.TaskId == taskId);
if (watch is not null && !_watches.TryRemove(new(watch.Record.WatchId, watch))) watch = null;
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));
if (watch is null) return false;
await watch.StopAsync().ConfigureAwait(false);
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;
RemoteAgent? agent = null;
var failures = 0;
var polls = 0;
try
{
while (!ct.IsCancellationRequested)
{
var record = watch.Record;
var deadline = DeadlineOf(record.StartedAt, settings.WatchTimeout);
TimeSpan wait;
try
{
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);
$"\"{record.Agent}\" gave no result within {settings.WatchTimeout}. Check it with action=status, agent={record.Agent}, taskId={record.TaskId}.", agent).ConfigureAwait(false);
return;
}
if (AgentsFile.Find(watch.Workspace, record.Agent) is not { } entry)
var agents = AgentsFile.Read(watch.Workspace);
if (agents.Error is { } error) throw new IOException(error); // transient: the user may be editing it
if (agents.Agents.FirstOrDefault(a => a.Name == 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
if (agent is null || agent.Entry != entry)
{
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);
agent?.Dispose();
agent = null;
agent = await Connector(settings).ConnectAsync(entry, ct).ConfigureAwait(false);
}
if (record.Source == WatchSources.Subscribe && !agent.Streaming)
record = await SetSourceAsync(watch, WatchSources.Poll).ConfigureAwait(false);
AgentTask? task;
var pause = false;
if (record.Source == WatchSources.Subscribe)
{
(task, pause) = await SubscribeAsync(agent, record.TaskId, deadline, ct).ConfigureAwait(false);
if ((task is null || !A2AResults.IsSettled(task.Status.State)) && !pause)
throw new IOException("the task stream ended before the task settled");
}
else
{
task = await GetTaskAsync(agent, record.TaskId, deadline, ct).ConfigureAwait(false);
}
if (task is not null && A2AResults.IsSettled(task.Status.State))
{
var saved = A2AResults.SaveFiles(watch.Workspace, record.Agent, task, settings.MaxFileBytes);
@@ -185,10 +242,10 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
return;
}
failures = 0;
wait = source switch
wait = record.Source switch
{
WatchSources.Push => SafetyPollInterval,
WatchSources.Subscribe => DelayAt(ReconnectDelays, 0),
WatchSources.Subscribe => TimeSpan.Zero, // our own cap or the deadline ended the stream
_ => DelayAt(PollDelays, polls++),
};
}
@@ -196,12 +253,18 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
{
return;
}
catch (Exception ex)
catch (Exception ex) when (ex is not OutOfMemoryException)
{
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++);
Logger.LogInformation("A2A: checking task {TaskId} of {Agent} failed ({Error})", record.TaskId, record.Agent, ex.Message);
agent?.Dispose();
agent = null;
if (record.Source == WatchSources.Subscribe && failures >= SubscribeFailuresBeforePoll)
{
Logger.LogInformation("A2A: task {TaskId} of {Agent}: subscribing failed {Count} times; polling instead", record.TaskId, record.Agent, failures);
record = await SetSourceAsync(watch, WatchSources.Poll).ConfigureAwait(false);
}
wait = record.Source == WatchSources.Subscribe ? DelayAt(ReconnectDelays, failures - 1) : DelayAt(PollDelays, polls++);
}
await WaitAsync(watch, Min(wait, deadline - Time.GetUtcNow()), ct).ConfigureAwait(false);
}
@@ -209,37 +272,62 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
}
catch (Exception ex)
catch (Exception ex) when (ex is not OutOfMemoryException)
{
Logger.LogWarning(ex, "A2A: watch {WatchId} crashed", record.WatchId);
Logger.LogWarning(ex, "A2A: watch {WatchId} crashed", watch.Record.WatchId);
}
finally
{
agent?.Dispose();
watch.DisposeStop();
}
}
private static DateTimeOffset DeadlineOf(DateTimeOffset start, TimeSpan timeout) =>
timeout >= DateTimeOffset.MaxValue - start ? DateTimeOffset.MaxValue : start + timeout;
private TimeSpan CallBound(DateTimeOffset deadline)
{
var remaining = deadline - Time.GetUtcNow();
return remaining < TimeSpan.FromSeconds(1) ? TimeSpan.FromSeconds(1) : Min(CallTimeout, remaining);
}
private async Task<AgentTask> GetTaskAsync(RemoteAgent agent, string taskId, DateTimeOffset deadline, CancellationToken ct)
{
using var bounded = CancellationTokenSource.CreateLinkedTokenSource(ct);
bounded.CancelAfter(CallBound(deadline));
return await agent.Client.GetTaskAsync(new GetTaskRequest { Id = taskId }, bounded.Token).ConfigureAwait(false);
}
/// <summary>
/// Follows the task's stream until it settles, the stream ends, or the watch deadline passes; then reads the task.
/// Follows the task's stream until it settles or ends, then reads the task. <c>Paused</c>: the stream was ended by
/// us (the deadline or <see cref="MaxSubscription"/>), which is not a failure.
/// </summary>
private async Task<AgentTask?> SubscribeAsync(RemoteAgent agent, string taskId, DateTimeOffset deadline, CancellationToken ct)
private async Task<(AgentTask? Task, bool Paused)> SubscribeAsync(RemoteAgent agent, string taskId, DateTimeOffset deadline, CancellationToken ct)
{
AgentTask? current = null;
var paused = false;
using (var until = CancellationTokenSource.CreateLinkedTokenSource(ct))
{
var remaining = deadline - Time.GetUtcNow();
using var until = CancellationTokenSource.CreateLinkedTokenSource(ct);
if (remaining > TimeSpan.Zero) until.CancelAfter(remaining);
else await until.CancelAsync().ConfigureAwait(false);
until.CancelAfter(remaining <= TimeSpan.Zero ? TimeSpan.Zero : Min(remaining, MaxSubscription));
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;
if (current is not null && A2AResults.IsSettled(current.Status.State)) return (current, false);
}
}
catch (A2AException) // e.g. the task is already terminal: read it instead
{
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested) // the deadline: the loop reports it
catch (OperationCanceledException) when (!ct.IsCancellationRequested && until.IsCancellationRequested)
{
paused = true;
}
return await agent.Client.GetTaskAsync(new GetTaskRequest { Id = taskId }, ct).ConfigureAwait(false);
}
return (await GetTaskAsync(agent, taskId, deadline, ct).ConfigureAwait(false), paused);
}
/// <summary>Waits <paramref name="wait"/>, or less when a push wakes the watch.</summary>
@@ -255,23 +343,40 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
ct.ThrowIfCancellationRequested();
}
private async Task<WatchRecord> SetSourceAsync(Watch watch, string source)
{
using (await LockAsync(watch.Workspace).ConfigureAwait(false))
{
watch.Record = watch.Record with { Source = source };
if (_watches.TryGetValue(watch.Record.WatchId, out var live) && ReferenceEquals(live, watch))
PendingFile.Upsert(watch.Workspace, watch.Record);
}
return watch.Record;
}
/// <summary>Delivers (or keeps undelivered) under the workspace lock, then drops the push config best-effort.</summary>
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)
var record = watch.Record;
using (await LockAsync(watch.Workspace).ConfigureAwait(false))
{
if (!_watches.TryRemove(new(record.WatchId, watch))) return; // stopped (cancel or shutdown) meanwhile
if (await TryDeliverAsync(record.Target, text, CancellationToken.None).ConfigureAwait(false))
PendingFile.Remove(watch.Workspace, record.WatchId);
else
PendingFile.Upsert(watch.Workspace, record with { Undelivered = text });
}
if (agent is null || record.PushToken is null) return;
try
{
using var timeout = new CancellationTokenSource(DeleteTimeout, Time);
await agent.Client.DeleteTaskPushNotificationConfigAsync(new DeleteTaskPushNotificationConfigRequest
{ TaskId = watch.Record.TaskId, Id = watch.Record.WatchId }, CancellationToken.None).ConfigureAwait(false);
{ TaskId = record.TaskId, Id = record.WatchId }, timeout.Token).ConfigureAwait(false);
}
catch (Exception ex) when (ex is A2AException or HttpRequestException) { }
catch (Exception ex) when (ex is not OutOfMemoryException)
{
Logger.LogInformation("A2A: removing the push config of task {TaskId} at {Agent} failed ({Error})", record.TaskId, record.Agent, ex.Message);
}
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<bool> TryDeliverAsync(string target, string text, CancellationToken ct)
@@ -280,27 +385,47 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
{
return await _delivery.DeliverAsync(target, text, ct).ConfigureAwait(false);
}
catch (Exception ex) when (ex is not OperationCanceledException)
catch (Exception ex) when (ex is not OperationCanceledException and not OutOfMemoryException)
{
Logger.LogWarning(ex, "A2A: delivering a result to {Target} failed", target);
return false;
}
}
private async Task<IDisposable> LockAsync(string workspace, CancellationToken cancellationToken = default)
{
var gate = _workspaceLocks.GetOrAdd(Path.GetFullPath(workspace), _ => new SemaphoreSlim(1, 1));
await gate.WaitAsync(cancellationToken).ConfigureAwait(false);
return new Releaser(gate);
}
private static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b;
/// <summary>Stops every watch (the records stay in pending.json for the next restore).</summary>
public async Task StopAsync()
{
foreach (var watch in _watches.Values) await watch.StopAsync().ConfigureAwait(false);
_watches.Clear();
var watches = _watches.ToArray();
foreach (var pair in watches) _watches.TryRemove(pair);
await Task.WhenAll(watches.Select(p => p.Value.StopAsync())).ConfigureAwait(false);
}
public async ValueTask DisposeAsync() => await StopAsync().ConfigureAwait(false);
private sealed class Releaser(SemaphoreSlim gate) : IDisposable
{
private int _released;
public void Dispose()
{
if (Interlocked.Exchange(ref _released, 1) == 0) gate.Release();
}
}
private sealed class Watch(WatchRecord record, string workspace)
{
public WatchRecord Record => record;
private volatile WatchRecord _record = record;
private int _disposed;
public WatchRecord Record { get => _record; set => _record = value; }
public string Workspace => workspace;
public CancellationTokenSource Stop { get; } = new();
public Channel<bool> Wake { get; } = Channel.CreateBounded<bool>(new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.DropOldest });
@@ -308,8 +433,15 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
public async Task StopAsync()
{
await Stop.CancelAsync().ConfigureAwait(false);
try { await Stop.CancelAsync().ConfigureAwait(false); }
catch (ObjectDisposedException) { } // the loop already ended
try { await Loop.ConfigureAwait(false); } catch (OperationCanceledException) { }
}
/// <summary>Called when the loop ends (its token's only user), or when the watch never started.</summary>
public void DisposeStop()
{
if (Interlocked.Exchange(ref _disposed, 1) == 0) Stop.Dispose();
}
}
}

4
extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs
View File

@@ -58,7 +58,9 @@ internal sealed class RemoteAgentConnector(HttpMessageHandler handler, TimeSpan
card = extended;
client = A2AClientFactory.Create(card, http);
}
catch (Exception ex) when (ex is A2AException or HttpRequestException) { } // keep the public card
// Refused, unreachable or too slow (the card timeout, not the caller's cancel): keep the public card.
catch (Exception ex) when (ex is A2AException or HttpRequestException ||
(ex is OperationCanceledException && !cancellationToken.IsCancellationRequested)) { }
}
return new RemoteAgent(entry, card, client, http);
}

195
tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs
View File

@@ -1,6 +1,7 @@
using System.Collections.Concurrent;
using System.Text.Json;
using A2A;
using Microsoft.Extensions.Logging;
using Volcanic.AgenticCore.Extensions.A2A;
using Volcanic.AgenticCore.Extensions.Tests.A2A.Support;
using Volcanic.AgenticCore.Testing;
@@ -25,6 +26,64 @@ public sealed class A2AWatchTests : IDisposable
public IEnumerable<string> KnownTargets() => ["1"];
}
/// <summary>Delivery that blocks until <see cref="Release"/> and then answers <see cref="Accept"/>.</summary>
private sealed class GatedDelivery(string workspace) : IA2AResultDelivery
{
private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
private int _entered;
public ConcurrentQueue<string> Delivered { get; } = new();
public bool Accept { get; init; } = true;
public int Entered => Volatile.Read(ref _entered);
public void Release() => _gate.TrySetResult();
public async Task<bool> DeliverAsync(string target, string text, CancellationToken cancellationToken)
{
Interlocked.Increment(ref _entered);
await _gate.Task;
if (Accept) Delivered.Enqueue(text);
return Accept;
}
public string? WorkspaceFor(string target) => workspace;
public IEnumerable<string> KnownTargets() => ["1"];
}
private sealed class ListLogger : ILogger
{
public ConcurrentQueue<string> Lines { get; } = new();
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func<TState, Exception?, string> formatter) =>
Lines.Enqueue(formatter(state, exception));
}
/// <summary>Over the remote: counts card fetches and, when <see cref="EndSubscribes"/>, ends every subscribe at once with no events.</summary>
private sealed class InterceptingHandler(HttpMessageHandler inner) : DelegatingHandler(inner)
{
private int _cards, _subscribes;
public bool EndSubscribes { get; init; }
public int Cards => Volatile.Read(ref _cards);
public int Subscribes => Volatile.Read(ref _subscribes);
protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
{
if (request.Method == HttpMethod.Get && request.RequestUri!.AbsolutePath.EndsWith("agent-card.json", StringComparison.Ordinal))
Interlocked.Increment(ref _cards);
if (request.Content is not null)
{
var body = await request.Content.ReadAsStringAsync(cancellationToken);
if (body.Contains("\"SubscribeToTask\"", StringComparison.Ordinal))
{
Interlocked.Increment(ref _subscribes);
if (EndSubscribes)
return new HttpResponseMessage(System.Net.HttpStatusCode.OK)
{
Content = new StringContent("", System.Text.Encoding.UTF8, "text/event-stream"),
};
}
}
return await base.SendAsync(request, cancellationToken);
}
}
private void WriteAgents(string key = RemoteAgentServer.Key)
{
Directory.CreateDirectory(Path.GetDirectoryName(AgentsFile.PathFor(_ws))!);
@@ -94,8 +153,14 @@ public sealed class A2AWatchTests : IDisposable
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
// Removed best-effort right after the result was delivered.
var deadline = DateTime.UtcNow.AddSeconds(15);
while ((await agent.Client.ListTaskPushNotificationConfigAsync(new ListTaskPushNotificationConfigRequest { TaskId = task.Id })).Configs?
.Any(c => c.Id == record.WatchId) == true)
{
Assert.True(DateTime.UtcNow < deadline, "the push config was not removed");
await Task.Delay(20);
}
}
[Fact]
@@ -215,6 +280,132 @@ public sealed class A2AWatchTests : IDisposable
Assert.Empty(delivery.Delivered);
}
[Fact]
public async Task Watch_MalformedAgentsFile_KeepsWatching()
{
await using var remote = await RemoteAgentServer.StartAsync();
remote.Faux.Enqueue(FauxTurn.Say("Survived").WithDelay(TimeSpan.FromMilliseconds(300)));
WriteAgents();
var delivery = new RecordingDelivery(_ws);
var log = new ListLogger();
await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler, log) { 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);
File.WriteAllText(AgentsFile.PathFor(_ws), "{ not json");
await registry.StartAsync("1", _ws, agent, task, settings, CancellationToken.None);
await WaitUntilAsync(() => log.Lines.Count(l => l.Contains("agents.json", StringComparison.Ordinal)) >= 3);
Assert.Empty(delivery.Delivered);
Assert.Single(registry.List("1"));
WriteAgents();
await WaitUntilAsync(() => !delivery.Delivered.IsEmpty);
Assert.Contains("Survived", Assert.Single(delivery.Delivered).Text);
}
[Fact]
public async Task Watch_SubscribeEndingUnsettled_FallsBackToPoll()
{
await using var remote = await RemoteAgentServer.StartAsync();
remote.Faux.Enqueue(FauxTurn.Say("Polled at last").WithDelay(TimeSpan.FromMilliseconds(500)));
WriteAgents();
var delivery = new RecordingDelivery(_ws);
var http = new InterceptingHandler(remote.Handler) { EndSubscribes = true };
await using var registry = new A2AWatchRegistry(TimeProvider.System, http)
{
SourceOverride = WatchSources.Subscribe, PollDelays = [TimeSpan.FromMilliseconds(50)], ReconnectDelays = [TimeSpan.FromMilliseconds(10)],
};
registry.UseDelivery(delivery);
var settings = new A2AClientSettings();
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);
await WaitUntilAsync(() => PendingFile.Read(_ws).Any(r => r.WatchId == record.WatchId && r.Source == WatchSources.Poll) || !delivery.Delivered.IsEmpty);
await WaitUntilAsync(() => !delivery.Delivered.IsEmpty);
Assert.Contains("Polled at last", Assert.Single(delivery.Delivered).Text);
Assert.Equal(3, http.Subscribes); // three failed subscribes, then polling only
}
[Fact]
public async Task Watch_ReusesTheConnectionBetweenChecks()
{
await using var remote = await RemoteAgentServer.StartAsync();
remote.Faux.Enqueue(FauxTurn.Say("Done").WithDelay(TimeSpan.FromMilliseconds(600)));
WriteAgents();
var delivery = new RecordingDelivery(_ws);
var http = new InterceptingHandler(remote.Handler);
await using var registry = new A2AWatchRegistry(TimeProvider.System, http) { SourceOverride = WatchSources.Poll, PollDelays = [TimeSpan.FromMilliseconds(30)] };
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 before = http.Cards;
await registry.StartAsync("1", _ws, agent, task, settings, CancellationToken.None);
await WaitUntilAsync(() => !delivery.Delivered.IsEmpty);
Assert.Equal(1, http.Cards - before); // one connect for the whole watch, not one per poll
}
[Fact]
public async Task Flush_Concurrent_DeliversOnce()
{
PendingFile.Upsert(_ws, new WatchRecord("w1", "1", "remote", "t1", "c1", null, DateTimeOffset.UtcNow, WatchSources.Poll, Undelivered: "the result"));
var delivery = new GatedDelivery(_ws);
await using var registry = new A2AWatchRegistry(TimeProvider.System, new HttpClientHandler());
registry.UseDelivery(delivery);
var first = registry.FlushUndeliveredAsync("1", _ws, CancellationToken.None);
var second = registry.FlushUndeliveredAsync("1", _ws, CancellationToken.None);
var restore = registry.RestoreAsync(new A2AClientSettings(), CancellationToken.None);
await WaitUntilAsync(() => delivery.Entered >= 1);
delivery.Release();
await Task.WhenAll(first, second, restore);
Assert.Single(delivery.Delivered);
Assert.Equal(1, delivery.Entered);
Assert.Empty(PendingFile.Read(_ws));
}
[Fact]
public async Task StopWatching_RacingFinish_DoesNotResurrectTheRecord()
{
await using var remote = await RemoteAgentServer.StartAsync();
remote.Faux.Enqueue(FauxTurn.Say("Too late"));
WriteAgents();
var delivery = new GatedDelivery(_ws) { Accept = false };
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.StartAsync("1", _ws, agent, task, settings, CancellationToken.None);
await WaitUntilAsync(() => delivery.Entered >= 1); // the watch is finishing: delivering (and about to keep it undelivered)
var stop = registry.StopWatchingAsync("1", _ws, task.Id);
delivery.Release();
await stop;
Assert.Empty(PendingFile.Read(_ws));
Assert.Empty(registry.List("1"));
}
[Fact]
public async Task Restore_PushWatchWithoutPushUrl_WatchesBySubscribe()
{
await using var remote = await RemoteAgentServer.StartAsync();
remote.Faux.Enqueue(FauxTurn.Say("Followed").WithDelay(TimeSpan.FromMilliseconds(800)));
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, "tok", DateTimeOffset.UtcNow, WatchSources.Push));
var delivery = new RecordingDelivery(_ws);
await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SafetyPollInterval = TimeSpan.FromMinutes(10) };
registry.UseDelivery(delivery);
Assert.Equal(1, await registry.RestoreAsync(settings, CancellationToken.None));
Assert.Equal(WatchSources.Subscribe, Assert.Single(registry.List("1")).Source);
await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); // a 10-minute safety poll would never get here in time
Assert.Contains("Followed", Assert.Single(delivery.Delivered).Text);
}
[Fact]
public void PollDelays_BackOffToFiveMinutes()
{