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

fix(a2a-client): no orphaned remote tasks, bounded tool calls, safer file reads

Browse Source
This commit is contained in:
EugeneTes
2026-10-08 14:07:21 +00:00
parent 173b7dfc18
commit 0d2583ab01
9 changed files with 423 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs
View File

@@ -35,11 +35,14 @@ public sealed class A2AClientExtension(A2AClientSettings settings, string target
{
var workspace = ctx.Cwd;
_registry.Live.Attach(target, workspace, api);
// Not inline: delivering starts a run, which mustn't happen inside session creation.
// Not inline and not at once: delivering starts a run, which mustn't happen inside session creation, and
// the host subscribes to the session only once creation returned, so the flush waits a moment. (Telegram,
// Task 8, delivers through its ChatRegistry instead, which waits until the chat's session is created.)
_ = Task.Run(async () =>
{
try
{
await Task.Delay(_registry.SessionStartFlushDelay, _registry.Time).ConfigureAwait(false);
await _registry.FlushUndeliveredAsync(target, workspace, CancellationToken.None).ConfigureAwait(false);
}
catch (Exception ex) when (ex is not OutOfMemoryException)

226
extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs
View File

@@ -78,7 +78,7 @@ internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatc
}
catch (Exception ex) when (ex is not OutOfMemoryException && (ex is not OperationCanceledException || !ct.IsCancellationRequested))
{
sb.Append($"\n- {entry.Name}: unreachable ({Clean(ex.Message, entry)})");
sb.Append($"\n- {entry.Name}: unreachable ({entry.Scrub(ex.Message)})");
}
}
foreach (var problem in file.Problems) sb.Append($"\n(skipped: {problem})");
@@ -92,44 +92,61 @@ internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatc
throw new ArgumentException("send needs a message or files.");
var parts = new List<Part>();
if (!string.IsNullOrWhiteSpace(args.Message)) parts.Add(Part.FromText(args.Message));
foreach (var path in args.Files ?? []) parts.Add(FilePart(workspace, path));
parts.AddRange(await FilePartsAsync(workspace, args.Files ?? [], ct).ConfigureAwait(false));
return await CallAsync(entry, ct, async agent =>
{
var response = await agent.Client.SendMessageAsync(new SendMessageRequest
{
Message = new A2AMessage
SendMessageResponse response;
using (var bounded = Bounded(ct))
response = await agent.Client.SendMessageAsync(new SendMessageRequest
{
Role = Role.User,
MessageId = Guid.NewGuid().ToString("N"),
Parts = parts,
ContextId = string.IsNullOrWhiteSpace(args.ContextId) ? null : args.ContextId,
TaskId = string.IsNullOrWhiteSpace(args.TaskId) ? null : args.TaskId,
},
Configuration = new SendMessageConfiguration { ReturnImmediately = true },
Metadata = new Dictionary<string, JsonElement> { ["volcanic.hops"] = JsonSerializer.SerializeToElement(inboundHops() + 1) },
}, ct).ConfigureAwait(false);
Message = new A2AMessage
{
Role = Role.User,
MessageId = Guid.NewGuid().ToString("N"),
Parts = parts,
ContextId = string.IsNullOrWhiteSpace(args.ContextId) ? null : args.ContextId,
TaskId = string.IsNullOrWhiteSpace(args.TaskId) ? null : args.TaskId,
},
Configuration = new SendMessageConfiguration { ReturnImmediately = true },
Metadata = new Dictionary<string, JsonElement> { ["volcanic.hops"] = JsonSerializer.SerializeToElement(inboundHops() + 1) },
}, bounded.Token).ConfigureAwait(false);
if (response.Task is not { } task)
{
var message = response.Message;
return $"Agent \"{entry.Name}\" replied (context {message?.ContextId}):\n{A2AResults.TextOf(message)}";
}
task = await WaitAsync(agent, task, ct).ConfigureAwait(false);
// The remote task exists from here on: whatever happens to this call, it is never left unwatched.
try
{
task = await WaitAsync(agent, task, ct).ConfigureAwait(false);
}
catch (Exception ex) when (ex is not OutOfMemoryException)
{
// Aborted, or the remote stopped answering while we waited: a watch takes over (its result still arrives).
await registry.StartAsync(target, workspace, agent, task, settings, CancellationToken.None).ConfigureAwait(false);
if (ct.IsCancellationRequested) throw;
return A2AResults.Describe(entry.Name, task, []) + "\n" + StillRunning;
}
if (A2AResults.IsSettled(task.Status.State)) return DescribeSettled(workspace, entry.Name, task);
await registry.StartAsync(target, workspace, agent, task, settings, ct).ConfigureAwait(false);
return A2AResults.Describe(entry.Name, task, []) +
"\nIt is still running: the result will be delivered to this conversation when it is ready. You can also check it with action=status.";
// Not the caller's token: an abort now must not undo a watch half set up.
await registry.StartAsync(target, workspace, agent, task, settings, CancellationToken.None).ConfigureAwait(false);
return A2AResults.Describe(entry.Name, task, []) + "\n" + StillRunning;
}).ConfigureAwait(false);
}
private const string StillRunning =
"It is still running: the result will be delivered to this conversation when it is ready. You can also check it with action=status.";
private async Task<string> StatusAsync(string workspace, A2AArgs args, CancellationToken ct)
{
var entry = Entry(workspace, args.Agent);
var taskId = Required(args.TaskId, "taskId");
return await CallAsync(entry, ct, async agent =>
{
var task = await agent.Client.GetTaskAsync(new GetTaskRequest { Id = taskId }, ct).ConfigureAwait(false);
var task = await GetTaskAsync(agent, taskId, ct).ConfigureAwait(false);
return A2AResults.IsSettled(task.Status.State) ? DescribeSettled(workspace, entry.Name, task) : A2AResults.Describe(entry.Name, task, []);
}).ConfigureAwait(false);
}
@@ -139,22 +156,34 @@ internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatc
var entry = Entry(workspace, args.Agent);
var taskId = Required(args.TaskId, "taskId");
// First, so a "canceled" outcome is never delivered as a follow-up.
await registry.StopWatchingAsync(target, workspace, taskId).ConfigureAwait(false);
return await CallAsync(entry, ct, async agent =>
var stopped = await registry.StopWatchingAsync(target, workspace, taskId).ConfigureAwait(false);
try
{
try
return await CallAsync(entry, ct, async agent =>
{
var task = await agent.Client.CancelTaskAsync(new CancelTaskRequest { Id = taskId }, ct).ConfigureAwait(false);
return A2AResults.Describe(entry.Name, task, []);
}
catch (A2AException ex)
{
// Typically: already finished. Its watch is gone, so hand over the outcome here.
var task = await agent.Client.GetTaskAsync(new GetTaskRequest { Id = taskId }, ct).ConfigureAwait(false);
return $"Could not cancel ({Clean(ex.Message, entry)}). " +
(A2AResults.IsSettled(task.Status.State) ? DescribeSettled(workspace, entry.Name, task) : A2AResults.Describe(entry.Name, task, []));
}
}).ConfigureAwait(false);
try
{
AgentTask task;
using (var bounded = Bounded(ct))
task = await agent.Client.CancelTaskAsync(new CancelTaskRequest { Id = taskId }, bounded.Token).ConfigureAwait(false);
return A2AResults.Describe(entry.Name, task, []);
}
catch (A2AException ex)
{
// Typically: already finished. Its watch is gone, so hand over the outcome here.
var task = await GetTaskAsync(agent, taskId, ct).ConfigureAwait(false);
return $"Could not cancel ({entry.Scrub(ex.Message)}). " +
(A2AResults.IsSettled(task.Status.State) ? DescribeSettled(workspace, entry.Name, task) : A2AResults.Describe(entry.Name, task, []));
}
}).ConfigureAwait(false);
}
catch (Exception ex) when (stopped is not null && ex is not OutOfMemoryException)
{
// Unreachable, too slow or aborted: the task may still run, so its watch goes on.
await registry.RearmAsync(workspace, stopped, settings).ConfigureAwait(false);
if (ex is InvalidOperationException) throw new InvalidOperationException(ex.Message + " The task is still watched.", ex);
throw;
}
}
private string DescribeSettled(string workspace, string agent, AgentTask task)
@@ -163,36 +192,62 @@ internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatc
return A2AResults.Describe(agent, task, saved.Saved, saved.Skipped);
}
/// <summary>Wait up to WaitSeconds for the task to settle: subscribe (streaming agents) or poll every second.</summary>
/// <summary>One remote call, bounded by the registry's call timeout.</summary>
private CancellationTokenSource Bounded(CancellationToken ct)
{
var bounded = CancellationTokenSource.CreateLinkedTokenSource(ct);
bounded.CancelAfter(registry.CallTimeout);
return bounded;
}
private async Task<AgentTask> GetTaskAsync(RemoteAgent agent, string taskId, CancellationToken ct)
{
using var bounded = Bounded(ct);
return await agent.Client.GetTaskAsync(new GetTaskRequest { Id = taskId }, bounded.Token).ConfigureAwait(false);
}
private static readonly TimeSpan PollInterval = TimeSpan.FromSeconds(1);
/// <summary>
/// Wait up to WaitSeconds for the task to settle: subscribe (streaming agents) or poll every second; a stream that
/// fails or ends unsettled falls back to polling. Throws on the caller's cancel (the caller then hands over to a watch).
/// </summary>
private async Task<AgentTask> WaitAsync(RemoteAgent agent, AgentTask task, CancellationToken ct)
{
if (A2AResults.IsSettled(task.Status.State) || settings.WaitSeconds <= 0) return task;
using var wait = CancellationTokenSource.CreateLinkedTokenSource(ct);
wait.CancelAfter(TimeSpan.FromSeconds(settings.WaitSeconds));
try
var streaming = agent.Streaming;
var pause = !streaming; // polling: the task was just read
while (true)
{
if (agent.Streaming)
try
{
AgentTask current = task;
await foreach (var ev in agent.Client.SubscribeToTaskAsync(new SubscribeToTaskRequest { Id = task.Id }, wait.Token).ConfigureAwait(false))
if (streaming)
{
current = TaskProjection.Apply(current, ev) ?? current;
if (A2AResults.IsSettled(current.Status.State)) return current;
AgentTask current = task;
await foreach (var ev in agent.Client.SubscribeToTaskAsync(new SubscribeToTaskRequest { Id = task.Id }, wait.Token).ConfigureAwait(false))
{
current = TaskProjection.Apply(current, ev) ?? current;
if (A2AResults.IsSettled(current.Status.State)) return current;
}
streaming = false; // the stream ended unsettled: poll for the rest of the wait
pause = false;
continue;
}
if (pause) await Task.Delay(PollInterval, registry.Time, wait.Token).ConfigureAwait(false);
pause = true;
var latest = await agent.Client.GetTaskAsync(new GetTaskRequest { Id = task.Id }, wait.Token).ConfigureAwait(false);
if (A2AResults.IsSettled(latest.Status.State)) return latest;
}
else
catch (OperationCanceledException) when (!ct.IsCancellationRequested) { break; } // waited long enough
catch (Exception ex) when (streaming && ex is A2AException or HttpRequestException or IOException)
{
while (true)
{
await Task.Delay(TimeSpan.FromSeconds(1), registry.Time, wait.Token).ConfigureAwait(false);
var latest = await agent.Client.GetTaskAsync(new GetTaskRequest { Id = task.Id }, wait.Token).ConfigureAwait(false);
if (A2AResults.IsSettled(latest.Status.State)) return latest;
}
streaming = false; // e.g. subscribing to a task that just finished, or a refused stream: poll at once
pause = false;
}
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested) { } // waited long enough
catch (A2AException) { } // e.g. subscribe on a task that just finished
return await agent.Client.GetTaskAsync(new GetTaskRequest { Id = task.Id }, ct).ConfigureAwait(false);
return await GetTaskAsync(agent, task.Id, ct).ConfigureAwait(false);
}
private async Task<string> CallAsync(RemoteAgentEntry entry, CancellationToken ct, Func<RemoteAgent, Task<string>> call)
@@ -208,11 +263,11 @@ internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatc
}
catch (HttpRequestException ex)
{
throw new InvalidOperationException($"Could not reach agent \"{entry.Name}\": {Clean(ex.Message, entry)}");
throw new InvalidOperationException($"Could not reach agent \"{entry.Name}\": {entry.Scrub(ex.Message)}");
}
catch (A2AException ex)
{
throw new InvalidOperationException($"Agent \"{entry.Name}\" refused: {Clean(ex.Message, entry)} ({ex.ErrorCode}).");
throw new InvalidOperationException($"Agent \"{entry.Name}\" refused: {entry.Scrub(ex.Message)} ({ex.ErrorCode}).");
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{
@@ -229,15 +284,66 @@ internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatc
?? throw new ArgumentException($"Unknown agent \"{name}\". Configured: {(file.Agents.Count == 0 ? "none (see .volcanic/a2a/agents.json)" : string.Join(", ", file.Agents.Select(a => a.Name)))}.");
}
private Part FilePart(string workspace, string path)
/// <summary>
/// Reads the attachments: regular workspace files (symlinks resolved and confined), together at most
/// <see cref="A2AClientSettings.MaxFileBytes"/>, each read through one opened stream so a file growing meanwhile is
/// still capped.
/// </summary>
private async Task<List<Part>> FilePartsAsync(string workspace, IReadOnlyList<string> paths, CancellationToken ct)
{
var full = ConfinedPath(workspace, path);
var info = new FileInfo(full);
if (!info.Exists) throw new ArgumentException($"{path} does not exist.");
if (info.Length > settings.MaxFileBytes) throw new ArgumentException($"{path} is larger than {settings.MaxFileBytes} bytes.");
return Part.FromRaw(File.ReadAllBytes(full), MediaType(full), Path.GetFileName(full));
var parts = new List<Part>();
var remaining = settings.MaxFileBytes;
foreach (var path in paths)
{
var full = ConfinedPath(workspace, path);
if (Directory.Exists(full) || !File.Exists(full)) throw new ArgumentException($"{path} does not exist or is not a file.");
if (!IsRegularFile(full)) throw new ArgumentException($"{path} is not a regular file.");
byte[] bytes;
await using (var stream = new FileStream(full, new FileStreamOptions
{ Mode = FileMode.Open, Access = FileAccess.Read, Share = FileShare.Read, Options = FileOptions.Asynchronous }))
{
using var copy = new MemoryStream();
var chunk = new byte[81920];
int n;
while ((n = await stream.ReadAsync(chunk, ct).ConfigureAwait(false)) > 0)
{
if (copy.Length + n > remaining)
throw new ArgumentException(remaining == settings.MaxFileBytes
? $"{path} is larger than {settings.MaxFileBytes} bytes."
: $"the attached files together are larger than {settings.MaxFileBytes} bytes.");
copy.Write(chunk, 0, n);
}
bytes = copy.ToArray();
}
remaining -= bytes.Length;
parts.Add(Part.FromRaw(bytes, MediaType(full), Path.GetFileName(full)));
}
return parts;
}
/// <summary>
/// False for FIFOs, sockets and devices, which could block or never end. Unix: the runtime's own stat (the managed
/// API reports every non-directory as a file); elsewhere every existing file counts as regular.
/// </summary>
private static bool IsRegularFile(string path)
{
if (OperatingSystem.IsWindows()) return true;
try
{
var status = new byte[512]; // FileStatus is ~120 bytes; Mode is the second int32
if (SystemNativeStat(path, status) != 0) return false;
const int typeMask = 0xF000, regular = 0x8000;
return (BitConverter.ToInt32(status, 4) & typeMask) == regular;
}
catch (Exception ex) when (ex is DllNotFoundException or EntryPointNotFoundException)
{
return true;
}
}
[System.Runtime.InteropServices.DllImport("libSystem.Native", EntryPoint = "SystemNative_Stat")]
private static extern int SystemNativeStat([System.Runtime.InteropServices.MarshalAs(System.Runtime.InteropServices.UnmanagedType.LPUTF8Str)] string path, byte[] status);
/// <summary>The real path of a workspace file: inside the workspace lexically and after every symlink is resolved.</summary>
internal static string ConfinedPath(string workspace, string path)
{
@@ -285,8 +391,4 @@ internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatc
private static string Required(string? value, string name) =>
string.IsNullOrWhiteSpace(value) ? throw new ArgumentException($"{name} is required.") : value;
/// <summary>Exception texts must never echo a key.</summary>
private static string Clean(string text, RemoteAgentEntry entry) =>
entry.Key is { Length: > 0 } key ? text.Replace(key, "[key]", StringComparison.Ordinal) : text;
}

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

@@ -49,6 +49,8 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
/// <summary>
/// (workspace, target) pairs where a flush skipped a record because it was claimed: if that delivery fails, the flush
/// runs again (else the result would wait for the next session start). Set and cleared under the workspace lock.
/// Keyed by target as well as workspace because a flush delivers one target's records only, so the retry must flush
/// the target whose record was skipped.
/// </summary>
private readonly ConcurrentDictionary<string, byte> _flushRequested = new(StringComparer.Ordinal);
private volatile IA2AResultDelivery _delivery;
@@ -73,6 +75,9 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
/// <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);
/// <summary>How long after a session start its flush of kept results waits (see <see cref="A2AClientExtension"/>).</summary>
internal TimeSpan SessionStartFlushDelay { get; init; } = TimeSpan.FromSeconds(1);
public void UseDelivery(IA2AResultDelivery delivery) => _delivery = delivery;
internal RemoteAgentConnector Connector(A2AClientSettings settings) =>
@@ -83,15 +88,23 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
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>
/// <summary>Hand a still-running remote task to a watch. A task the target already watches keeps its 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");
var source = SourceOverride ?? (agent.Streaming ? WatchSources.Subscribe : WatchSources.Poll);
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;
using (await LockAsync(workspace, cancellationToken).ConfigureAwait(false))
{
if (_watches.Values.FirstOrDefault(w => w.Record.Target == target && w.Record.TaskId == task.Id) is { } existing)
{
watch.DisposeStop();
return existing.Record;
}
// 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)
@@ -112,7 +125,7 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
}
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);
Logger.LogInformation("A2A: {Agent} refused push for task {TaskId} ({Error}); watching by {Source}", agent.Entry.Name, task.Id, agent.Entry.Scrub(ex.Message), source);
watch.Record = watch.Record with { PushToken = null };
}
}
@@ -214,8 +227,8 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
}
}
/// <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)
/// <summary>Stop watching a task (after <c>cancel</c>); the stopped watch's record, null when none was running.</summary>
public async Task<WatchRecord?> StopWatchingAsync(string target, string workspace, string taskId)
{
Watch? watch;
using (await LockAsync(workspace).ConfigureAwait(false))
@@ -225,15 +238,32 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
foreach (var record in PendingFile.Read(workspace).Where(r => r.Target == target && r.TaskId == taskId))
PendingFile.Remove(workspace, record.WatchId);
}
if (watch is null) return false;
if (watch is null) return null;
await watch.StopAsync().ConfigureAwait(false);
return true;
return watch.Record;
}
/// <summary>
/// Resumes a watch stopped by <see cref="StopWatchingAsync"/> whose cancel didn't go through (same id, push token and
/// start). No-op when the task is watched again already.
/// </summary>
internal async Task RearmAsync(string workspace, WatchRecord record, A2AClientSettings settings)
{
using (await LockAsync(workspace).ConfigureAwait(false))
{
if (_watches.Values.Any(w => w.Record.Target == record.Target && w.Record.TaskId == record.TaskId)) return;
var watch = new Watch(record with { Undelivered = null }, workspace);
if (!_watches.TryAdd(record.WatchId, watch)) { watch.DisposeStop(); return; }
PendingFile.Upsert(workspace, watch.Record);
watch.Loop = Task.Run(() => RunAsync(watch, settings));
}
}
private async Task RunAsync(Watch watch, A2AClientSettings settings)
{
var ct = watch.Stop.Token;
RemoteAgent? agent = null;
RemoteAgentEntry? lastEntry = null;
var failures = 0;
var polls = 0;
try
@@ -253,7 +283,9 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
}
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)
var entry = agents.Agents.FirstOrDefault(a => a.Name == record.Agent);
if (entry is not null) lastEntry = entry;
if (entry is null)
{
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);
@@ -301,7 +333,8 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
catch (Exception ex) when (ex is not OutOfMemoryException)
{
failures++;
Logger.LogInformation("A2A: checking task {TaskId} of {Agent} failed ({Error})", record.TaskId, record.Agent, ex.Message);
Logger.LogInformation("A2A: checking task {TaskId} of {Agent} failed ({Error})", record.TaskId, record.Agent,
lastEntry is null ? ex.Message : lastEntry.Scrub(ex.Message));
agent?.Dispose();
agent = null;
if (record.Source == WatchSources.Subscribe && failures >= SubscribeFailuresBeforePoll)
@@ -447,7 +480,7 @@ public sealed class A2AWatchRegistry : IAsyncDisposable
}
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);
Logger.LogInformation("A2A: removing the push config of task {TaskId} at {Agent} failed ({Error})", record.TaskId, record.Agent, agent.Entry.Scrub(ex.Message));
}
}

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

@@ -2,6 +2,13 @@ using A2A;
namespace Volcanic.AgenticCore.Extensions.A2A;
internal static class RemoteAgentEntryExtensions
{
/// <summary>Texts shown to the model or logged (exception messages) must never echo the agent's key.</summary>
public static string Scrub(this RemoteAgentEntry entry, string text) =>
entry.Key is { Length: > 0 } key ? text.Replace(key, "[key]", StringComparison.Ordinal) : text;
}
/// <summary>A connected remote agent: its (extended) card and a client that sends the key.</summary>
internal sealed class RemoteAgent(RemoteAgentEntry entry, AgentCard card, IA2AClient client, HttpClient http) : IDisposable
{

2
tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AHopsTests.cs
View File

@@ -39,7 +39,7 @@ public class A2AHopsTests
public async Task Hops_FromInboundEvent()
{
await using var remote = await RemoteAgentServer.StartAsync(o => o.MaxHops = 3);
await using var host = new A2AClientHost(remote, null, new InboundEmitter(3));
await using var host = new A2AClientHost(remote, null, null, new InboundEmitter(3));
var session = await host.CreateSessionAsync();
var (text, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "hi" });
Assert.Contains("failed", text);

117
tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs
View File

@@ -123,7 +123,10 @@ public class A2AToolTests
WatchSources.Poll, "[A2A result from \"remote\", task t1 (completed)]\nKept for later"));
host.Faux.Enqueue(FauxTurn.Say("got it"));
var session = await host.CreateSessionAsync(); // returns before the flush's run (it runs off the creation path)
var delivered = await host.WaitForUserMessageAsync(session, "Kept for later");
Assert.DoesNotContain(session.Messages, m => m.Text.Contains("Kept for later")); // not inside creation
// Read from the transcript, not the host's event subscription, so an early run can't be missed.
await WaitUntilAsync(() => session.Messages.Any(m => m is Abstractions.Messages.UserMessage && m.Text.Contains("Kept for later")));
var delivered = session.Messages.First(m => m.Text.Contains("Kept for later")).Text;
Assert.StartsWith("[A2A result from \"remote\"", delivered);
await WaitUntilAsync(() => PendingFile.Read(host.Workspace).Count == 0);
}
@@ -229,7 +232,7 @@ public class A2AToolTests
var taskId = Field(started, "task");
Assert.Single(host.Registry.List(A2AClientHost.Target));
var (status, _) = await host.CallToolAsync(session, new { action = "status", agent = "remote", taskId });
Assert.Contains("working", status);
Assert.True(status.Contains("working") || status.Contains("submitted"), status);
var (cancelled, isError) = await host.CallToolAsync(session, new { action = "cancel", agent = "remote", taskId });
Assert.False(isError, cancelled);
Assert.Contains("canceled", cancelled);
@@ -238,6 +241,116 @@ public class A2AToolTests
Assert.DoesNotContain(host.UserTexts(), t => t.Contains("[A2A result", StringComparison.Ordinal));
}
[Fact]
public async Task Send_AbortedWhileWaiting_TaskWatched_AndResultArrives()
{
await using var remote = await RemoteAgentServer.StartAsync();
remote.Faux.Enqueue(FauxTurn.Say("Late answer").WithDelay(TimeSpan.FromMilliseconds(1500)));
RemoteRewriter? rewriter = null;
await using var host = new A2AClientHost(remote, handler: inner => rewriter = new RemoteRewriter(inner));
var session = await host.CreateSessionAsync();
host.Faux.Enqueue(FauxTurn.Tool("a2a", new { action = "send", agent = "remote", message = "take your time" }), FauxTurn.Say("ok"));
var run = session.PromptAsync("use the remote agent");
await WaitUntilAsync(() => rewriter!.Subscribes > 0); // the task exists and the tool waits on its stream
await session.AbortAsync();
try { await run; } catch (OperationCanceledException) { }
Assert.Single(host.Registry.List(A2AClientHost.Target));
var delivered = await host.WaitForUserMessageAsync(session, "[A2A result from \"remote\"");
Assert.Contains("Late answer", delivered);
}
[Fact]
public async Task Send_NonStreamingAgent_PollsWithinWait()
{
await using var remote = await RemoteAgentServer.StartAsync();
remote.Faux.Enqueue(FauxTurn.Say("Polled answer").WithDelay(TimeSpan.FromMilliseconds(1200)));
RemoteRewriter? rewriter = null;
await using var host = new A2AClientHost(remote, handler: inner => rewriter = new RemoteRewriter(inner) { NoStreaming = true });
var session = await host.CreateSessionAsync();
var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "go" });
Assert.False(isError, text);
Assert.Contains("completed", text);
Assert.Contains("Polled answer", text);
Assert.Equal(0, rewriter!.Subscribes);
Assert.Empty(host.Registry.List(A2AClientHost.Target));
}
[Fact]
public async Task Send_MessageReply_ReturnedAsText()
{
await using var remote = await RemoteAgentServer.StartAsync();
await using var host = new A2AClientHost(remote, handler: inner => new RemoteRewriter(inner) { MessageReply = "Just a message" });
var session = await host.CreateSessionAsync();
var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "hi" });
Assert.False(isError, text);
Assert.Contains("replied (context ctx-message)", text);
Assert.Contains("Just a message", text);
Assert.Empty(host.Registry.List(A2AClientHost.Target));
}
[Fact]
public async Task Send_FileOverMaxBytes_Error()
{
await using var remote = await RemoteAgentServer.StartAsync();
await using var host = new A2AClientHost(remote, new A2AClientSettings { Enabled = true, MaxFileBytes = 10 });
File.WriteAllText(Path.Combine(host.Workspace, "big.txt"), new string('x', 11));
var session = await host.CreateSessionAsync();
var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "x", files = new[] { "big.txt" } });
Assert.True(isError);
Assert.Contains("larger than 10 bytes", text);
Assert.Empty(remote.Faux.Requests);
}
[Fact]
public async Task Send_FilesTogetherOverMaxBytes_Error()
{
await using var remote = await RemoteAgentServer.StartAsync();
await using var host = new A2AClientHost(remote, new A2AClientSettings { Enabled = true, MaxFileBytes = 10 });
File.WriteAllText(Path.Combine(host.Workspace, "a.txt"), new string('a', 6));
File.WriteAllText(Path.Combine(host.Workspace, "b.txt"), new string('b', 6));
var session = await host.CreateSessionAsync();
var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "x", files = new[] { "a.txt", "b.txt" } });
Assert.True(isError);
Assert.Contains("together are larger than 10 bytes", text);
Assert.Empty(remote.Faux.Requests);
}
[Fact]
public async Task Send_Fifo_Rejected()
{
if (OperatingSystem.IsWindows()) return;
await using var remote = await RemoteAgentServer.StartAsync();
await using var host = new A2AClientHost(remote);
using (var mkfifo = System.Diagnostics.Process.Start("mkfifo", Path.Combine(host.Workspace, "pipe")))
await mkfifo.WaitForExitAsync();
var session = await host.CreateSessionAsync();
var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "x", files = new[] { "pipe" } });
Assert.True(isError);
Assert.Contains("not a regular file", text);
Assert.Empty(remote.Faux.Requests);
}
[Fact]
public async Task Cancel_RemoteDown_KeepsWatching_AndResultStillArrives()
{
await using var remote = await RemoteAgentServer.StartAsync();
remote.Faux.Enqueue(FauxTurn.Say("Survived").WithDelay(TimeSpan.FromMilliseconds(1500)));
RemoteRewriter? rewriter = null;
await using var host = new A2AClientHost(remote, new A2AClientSettings { Enabled = true, WaitSeconds = 0 }, inner => rewriter = new RemoteRewriter(inner));
var session = await host.CreateSessionAsync();
var (started, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "go" });
var watch = Assert.Single(host.Registry.List(A2AClientHost.Target));
rewriter!.Down = true;
var (cancelled, isError) = await host.CallToolAsync(session, new { action = "cancel", agent = "remote", taskId = Field(started, "task") });
Assert.True(isError);
Assert.Contains("still watched", cancelled);
Assert.Equal(watch.WatchId, Assert.Single(host.Registry.List(A2AClientHost.Target)).WatchId);
Assert.Contains(PendingFile.Read(host.Workspace), r => r.WatchId == watch.WatchId);
rewriter.Down = false;
var delivered = await host.WaitForUserMessageAsync(session, "[A2A result from \"remote\"");
Assert.Contains("Survived", delivered);
}
[Fact]
public async Task NoAgentsConfigured_ExplainsFile()
{

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

@@ -316,10 +316,10 @@ public sealed class A2AWatchTests : IDisposable
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.NotNull(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.Null(await registry.StopWatchingAsync("1", _ws, task.Id));
Assert.Empty(delivery.Delivered);
}
@@ -444,6 +444,24 @@ public sealed class A2AWatchTests : IDisposable
await WaitUntilAsync(() => PendingFile.Read(_ws).Count == 0);
}
[Fact]
public async Task Start_SameTaskTwice_KeepsOneWatch()
{
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) { SourceOverride = WatchSources.Poll, PollDelays = [TimeSpan.FromMilliseconds(100)] };
registry.UseDelivery(new RecordingDelivery(_ws));
var settings = new A2AClientSettings();
using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None);
var task = await StartRemoteTaskAsync(agent);
var first = await registry.StartAsync("1", _ws, agent, task, settings, CancellationToken.None);
var second = await registry.StartAsync("1", _ws, agent, task, settings, CancellationToken.None);
Assert.Equal(first.WatchId, second.WatchId);
Assert.Single(registry.List("1"));
Assert.Single(PendingFile.Read(_ws));
}
[Fact]
public async Task StopWatching_RacingFinish_DoesNotResurrectTheRecord()
{
@@ -505,7 +523,7 @@ public sealed class A2AWatchTests : IDisposable
// The delivered result's "turn" cancels the slow task's watch and starts a new watch in the same workspace.
delivery.OnDeliver = async () =>
{
Assert.True(await registry.StopWatchingAsync("1", _ws, slow.Id));
Assert.NotNull(await registry.StopWatchingAsync("1", _ws, slow.Id));
await registry.StartAsync("1", _ws, agent, slow, settings, CancellationToken.None);
turnDone = true;
};

10
tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs
View File

@@ -20,11 +20,17 @@ public sealed class A2AClientHost : IAsyncDisposable
private readonly ServiceProvider _services;
private readonly List<AgentSession> _sessions = [];
public A2AClientHost(RemoteAgentServer remote, A2AClientSettings? settings = null, params IExtension[] extra)
/// <param name="handler">Wraps the handler that reaches the remote (e.g. a <see cref="RemoteRewriter"/>).</param>
public A2AClientHost(RemoteAgentServer remote, A2AClientSettings? settings = null,
Func<HttpMessageHandler, HttpMessageHandler>? handler = null, params IExtension[] extra)
{
Directory.CreateDirectory(Workspace);
Settings = settings ?? new A2AClientSettings { Enabled = true, WaitSeconds = 10 };
Registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { PollDelays = [TimeSpan.FromMilliseconds(100)] };
Registry = new A2AWatchRegistry(TimeProvider.System, handler is null ? remote.Handler : handler(remote.Handler))
{
PollDelays = [TimeSpan.FromMilliseconds(100)],
ReconnectDelays = [TimeSpan.FromMilliseconds(100)],
};
var services = new ServiceCollection();
services.AddAgenticCore(o =>
{

59
tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/RemoteRewriter.cs Normal file
View File

@@ -0,0 +1,59 @@
using System.Net;
using System.Text;
using System.Text.Json;
using System.Text.Json.Nodes;
using A2A;
using A2AMessage = global::A2A.Message;
namespace Volcanic.AgenticCore.Extensions.Tests.A2A.Support;
/// <summary>
/// Between our client and the remote: counts subscribes and card fetches, can hide the card's streaming capability
/// (so the client polls), and can answer <c>SendMessage</c> itself with a plain message instead of a task.
/// </summary>
public sealed class RemoteRewriter(HttpMessageHandler inner) : DelegatingHandler(inner)
{
private int _subscribes;
/// <summary>Rewrites <c>"streaming":true</c> to false in JSON responses (the public and the extended card).</summary>
public bool NoStreaming { get; init; }
/// <summary>When set, <c>SendMessage</c> is answered with a message carrying this text (the remote never sees it).</summary>
public string? MessageReply { get; init; }
/// <summary>When true, every request fails as if the remote were down.</summary>
public volatile bool Down;
public int Subscribes => Volatile.Read(ref _subscribes);
protected override async Task<HttpResponseMessage> SendAsync(HttpRequestMessage request, CancellationToken cancellationToken)
{
if (Down) throw new HttpRequestException("remote is down");
string? body = request.Content is null ? null : await request.Content.ReadAsStringAsync(cancellationToken);
if (body is not null && body.Contains("\"SubscribeToTask\"", StringComparison.Ordinal)) Interlocked.Increment(ref _subscribes);
if (MessageReply is not null && body is not null && body.Contains("\"SendMessage\"", StringComparison.Ordinal))
{
var id = JsonNode.Parse(body)!["id"]?.DeepClone();
var result = JsonSerializer.SerializeToNode(new SendMessageResponse
{
Message = new A2AMessage
{
Role = Role.Agent,
MessageId = Guid.NewGuid().ToString("N"),
ContextId = "ctx-message",
Parts = [Part.FromText(MessageReply)],
},
}, A2AJsonUtilities.DefaultOptions);
var rpc = new JsonObject { ["jsonrpc"] = "2.0", ["id"] = id, ["result"] = result };
return new HttpResponseMessage(HttpStatusCode.OK) { Content = new StringContent(rpc.ToJsonString(), Encoding.UTF8, "application/json") };
}
var response = await base.SendAsync(request, cancellationToken);
if (NoStreaming && response.Content.Headers.ContentType?.MediaType == "application/json")
{
var json = await response.Content.ReadAsStringAsync(cancellationToken);
if (json.Contains("\"streaming\":true", StringComparison.Ordinal))
response.Content = new StringContent(json.Replace("\"streaming\":true", "\"streaming\":false", StringComparison.Ordinal), Encoding.UTF8, "application/json");
}
return response;
}
}