feat(a2a-client): watch long remote tasks by push, subscribe or poll and deliver the results
This commit is contained in:
EugeneTes
@@ -0,0 +1,49 @@
|
|||||||
|
using System.Collections.Concurrent;
|
||||||
|
using Volcanic.AgenticCore.Abstractions.Extensions;
|
||||||
|
|
||||||
|
namespace Volcanic.AgenticCore.Extensions.A2A;
|
||||||
|
|
||||||
|
/// <summary>Where finished remote tasks go. Targets are host ids (Telegram: chat ids).</summary>
|
||||||
|
public interface IA2AResultDelivery
|
||||||
|
{
|
||||||
|
/// <summary>Run <paramref name="text"/> as a follow-up prompt in the target's conversation. False: not deliverable now.</summary>
|
||||||
|
Task<bool> DeliverAsync(string target, string text, CancellationToken cancellationToken);
|
||||||
|
|
||||||
|
/// <summary>The target's workspace. A path computation: must not create or wake a session.</summary>
|
||||||
|
string? WorkspaceFor(string target);
|
||||||
|
|
||||||
|
/// <summary>Targets whose workspace may hold a pending.json (restore at startup).</summary>
|
||||||
|
IEnumerable<string> KnownTargets();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Default delivery: the target's live session, if any (registered on session start, removed on shutdown).</summary>
|
||||||
|
public sealed class LiveSessionDelivery : IA2AResultDelivery
|
||||||
|
{
|
||||||
|
private readonly ConcurrentDictionary<string, (IExtensionApi Api, string Workspace)> _live = new(StringComparer.Ordinal);
|
||||||
|
|
||||||
|
public void Attach(string target, string workspace, IExtensionApi api) => _live[target] = (api, workspace);
|
||||||
|
|
||||||
|
/// <summary>Detaches <paramref name="api"/> only if it is still the target's live session (a newer one may have attached).</summary>
|
||||||
|
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<bool> 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<string> KnownTargets() => _live.Keys.ToList();
|
||||||
|
}
|
||||||
@@ -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;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Process-wide (<see cref="Shared"/>): 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 <see cref="IA2AResultDelivery"/> and
|
||||||
|
/// is persisted in the workspace's <c>pending.json</c> until delivered.
|
||||||
|
/// </summary>
|
||||||
|
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<string, Watch> _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<WatchRecord> List(string target) =>
|
||||||
|
_watches.Values.Where(w => w.Record.Target == target).Select(w => w.Record).ToList();
|
||||||
|
|
||||||
|
/// <summary>Hand a still-running remote task to a watch.</summary>
|
||||||
|
internal async Task<WatchRecord> 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;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Push receiver: a known watch with the right token wakes up and checks its task (the body is never trusted).
|
||||||
|
/// False → 404.
|
||||||
|
/// </summary>
|
||||||
|
public Task<bool> 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);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Re-arm every persisted watch of the delivery's known targets; delivers results kept undelivered. Returns the count re-armed.</summary>
|
||||||
|
public async Task<int> 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;
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Deliver results that were kept because the target had no live conversation.</summary>
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <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)
|
||||||
|
{
|
||||||
|
_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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Follows the task's stream until it settles, the stream ends, or the watch deadline passes; then reads the task.
|
||||||
|
/// </summary>
|
||||||
|
private async Task<AgentTask?> 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);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>Waits <paramref name="wait"/>, or less when a push wakes the watch.</summary>
|
||||||
|
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<bool> 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;
|
||||||
|
|
||||||
|
/// <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();
|
||||||
|
}
|
||||||
|
|
||||||
|
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<bool> Wake { get; } = Channel.CreateBounded<bool>(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) { }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
228
tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs
Normal file
228
tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs
Normal file
@@ -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<bool> 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<string> 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<AgentTask> 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<bool> 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) { } }
|
||||||
|
}
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
using System.Net;
|
||||||
|
using System.Text.Json;
|
||||||
|
using Volcanic.AgenticCore.Extensions.A2A;
|
||||||
|
|
||||||
|
namespace Volcanic.AgenticCore.Extensions.Tests.A2A.Support;
|
||||||
|
|
||||||
|
/// <summary>Stands in for our push receiver endpoint: <c>POST …/a2a/push/{watchId}</c> → <see cref="A2AWatchRegistry.OnPushAsync"/>.</summary>
|
||||||
|
public sealed class PushToRegistryHandler(A2AWatchRegistry? registry) : HttpMessageHandler
|
||||||
|
{
|
||||||
|
private int _received;
|
||||||
|
|
||||||
|
public int Received => Volatile.Read(ref _received);
|
||||||
|
|
||||||
|
protected override async Task<HttpResponseMessage> 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);
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user