diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs
index 8b30c74..0a9d108 100644
--- a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs
+++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs
@@ -15,6 +15,12 @@ namespace Volcanic.AgenticCore.Extensions.A2A;
/// subscribing (streaming agents), else by polling; it delivers the outcome through and
/// is persisted in the workspace's pending.json until delivered.
///
+///
+/// Delivery is at least once: a result is delivered, then removed from pending.json, 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.
+///
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)];
+ /// Longest single subscription; the stream is then reopened (a pause, not a failure).
+ private static readonly TimeSpan MaxSubscription = TimeSpan.FromDays(1);
+
+ private static readonly TimeSpan DeleteTimeout = TimeSpan.FromSeconds(10);
+
+ /// After this many failed checks in a row, a subscribe watch polls instead.
+ private const int SubscribeFailuresBeforePoll = 3;
private readonly HttpMessageHandler _handler;
private readonly ConcurrentDictionary _watches = new(StringComparer.Ordinal);
+ private readonly ConcurrentDictionary _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);
+ /// Bound of one remote call (get, create/delete push config), never past the watch deadline.
+ 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 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)
+ 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
{
- token = Convert.ToHexString(RandomNumberGenerator.GetBytes(32));
- try
+ if (SourceOverride is null && settings.PushBaseUrl is { Length: > 0 } pushBase && agent.SupportsPush)
{
- await agent.Client.CreateTaskPushNotificationConfigAsync(new CreateTaskPushNotificationConfigRequest
+ var token = Convert.ToHexString(RandomNumberGenerator.GetBytes(32));
+ watch.Record = watch.Record with { PushToken = token };
+ try
{
- 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;
+ 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 },
+ }, timeout.Token).ConfigureAwait(false);
+ watch.Record = watch.Record with { Source = WatchSources.Push };
+ }
+ 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);
+ watch.Record = watch.Record with { PushToken = null };
+ }
}
+ using (await LockAsync(workspace).ConfigureAwait(false))
+ PendingFile.Upsert(workspace, watch.Record);
}
- 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;
+ catch
+ {
+ _watches.TryRemove(new(watchId, watch));
+ watch.DisposeStop();
+ throw;
+ }
+ watch.Loop = Task.Run(() => RunAsync(watch, settings));
+ return watch.Record;
}
///
@@ -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;
@@ -119,65 +161,80 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
/// 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);
+ 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);
+ }
}
/// 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)
+ 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);
}
- 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)
{
- 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;
- }
-
+ var record = watch.Record;
+ var deadline = DeadlineOf(record.StartedAt, settings.WatchTimeout);
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 (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}.", agent).ConfigureAwait(false);
+ return;
+ }
+ 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;
+ }
+ if (agent is null || agent.Entry != entry)
+ {
+ 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 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);
+ }
+
///
- /// 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. Paused: the stream was ended by
+ /// us (the deadline or ), which is not a failure.
///
- private async Task 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 remaining = deadline - Time.GetUtcNow();
- using var until = CancellationTokenSource.CreateLinkedTokenSource(ct);
- if (remaining > TimeSpan.Zero) until.CancelAfter(remaining);
- else await until.CancelAsync().ConfigureAwait(false);
- try
+ var paused = false;
+ using (var until = CancellationTokenSource.CreateLinkedTokenSource(ct))
{
- await foreach (var ev in agent.Client.SubscribeToTaskAsync(new SubscribeToTaskRequest { Id = taskId }, until.Token).ConfigureAwait(false))
+ var remaining = deadline - Time.GetUtcNow();
+ until.CancelAfter(remaining <= TimeSpan.Zero ? TimeSpan.Zero : Min(remaining, MaxSubscription));
+ try
{
- current = TaskProjection.Apply(current, ev);
- if (current is not null && A2AResults.IsSettled(current.Status.State)) return current;
+ 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, false);
+ }
+ }
+ catch (A2AException) // e.g. the task is already terminal: read it instead
+ {
+ }
+ catch (OperationCanceledException) when (!ct.IsCancellationRequested && until.IsCancellationRequested)
+ {
+ paused = true;
}
}
- 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);
+ return (await GetTaskAsync(agent, taskId, deadline, ct).ConfigureAwait(false), paused);
}
/// Waits , or less when a push wakes the watch.
@@ -255,23 +343,40 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
ct.ThrowIfCancellationRequested();
}
+ private async Task 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;
+ }
+
+ /// Delivers (or keeps undelivered) under the workspace lock, then drops the push config best-effort.
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))
{
- 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 (!_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 = record.TaskId, Id = record.WatchId }, timeout.Token).ConfigureAwait(false);
+ }
+ 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 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 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;
/// 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();
+ 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 Wake { get; } = Channel.CreateBounded(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) { }
}
+
+ /// Called when the loop ends (its token's only user), or when the watch never started.
+ public void DisposeStop()
+ {
+ if (Interlocked.Exchange(ref _disposed, 1) == 0) Stop.Dispose();
+ }
}
}
diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs
index 65e19a6..7255463 100644
--- a/extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs
+++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs
@@ -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);
}
diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs
index 1f05d18..1984018 100644
--- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs
+++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs
@@ -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 KnownTargets() => ["1"];
}
+ /// Delivery that blocks until and then answers .
+ private sealed class GatedDelivery(string workspace) : IA2AResultDelivery
+ {
+ private readonly TaskCompletionSource _gate = new(TaskCreationOptions.RunContinuationsAsynchronously);
+ private int _entered;
+ public ConcurrentQueue Delivered { get; } = new();
+ public bool Accept { get; init; } = true;
+ public int Entered => Volatile.Read(ref _entered);
+ public void Release() => _gate.TrySetResult();
+ public async Task 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 KnownTargets() => ["1"];
+ }
+
+ private sealed class ListLogger : ILogger
+ {
+ public ConcurrentQueue Lines { get; } = new();
+ public IDisposable? BeginScope(TState state) where TState : notnull => null;
+ public bool IsEnabled(LogLevel logLevel) => true;
+ public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) =>
+ Lines.Enqueue(formatter(state, exception));
+ }
+
+ /// Over the remote: counts card fetches and, when , ends every subscribe at once with no events.
+ 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 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()
{