diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs index 11b9f91..614c10c 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs @@ -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) diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs index 9ce938d..c08eada 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs @@ -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(); 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 { ["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 { ["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 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); } - /// Wait up to WaitSeconds for the task to settle: subscribe (streaming agents) or poll every second. + /// One remote call, bounded by the registry's call timeout. + private CancellationTokenSource Bounded(CancellationToken ct) + { + var bounded = CancellationTokenSource.CreateLinkedTokenSource(ct); + bounded.CancelAfter(registry.CallTimeout); + return bounded; + } + + private async Task 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); + + /// + /// 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). + /// private async Task 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 CallAsync(RemoteAgentEntry entry, CancellationToken ct, Func> 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) + /// + /// Reads the attachments: regular workspace files (symlinks resolved and confined), together at most + /// , each read through one opened stream so a file growing meanwhile is + /// still capped. + /// + private async Task> FilePartsAsync(string workspace, IReadOnlyList 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(); + 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; } + /// + /// 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. + /// + 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); + /// The real path of a workspace file: inside the workspace lexically and after every symlink is resolved. 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; - - /// Exception texts must never echo a key. - private static string Clean(string text, RemoteAgentEntry entry) => - entry.Key is { Length: > 0 } key ? text.Replace(key, "[key]", StringComparison.Ordinal) : text; } diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs index 5a51584..3cc5560 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs @@ -49,6 +49,8 @@ public sealed class A2AWatchRegistry : IAsyncDisposable /// /// (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. /// private readonly ConcurrentDictionary _flushRequested = new(StringComparer.Ordinal); private volatile IA2AResultDelivery _delivery; @@ -73,6 +75,9 @@ public sealed class A2AWatchRegistry : IAsyncDisposable /// Bound of one remote call (get, create/delete push config), never past the watch deadline. internal TimeSpan CallTimeout { get; init; } = TimeSpan.FromSeconds(60); + /// How long after a session start its flush of kept results waits (see ). + 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 List(string target) => _watches.Values.Where(w => w.Record.Target == target).Select(w => w.Record).ToList(); - /// Hand a still-running remote task to a watch. + /// Hand a still-running remote task to a watch. A task the target already watches keeps its watch. internal async Task 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 } } - /// Stop watching a task (after cancel); true when a watch existed. - public async Task StopWatchingAsync(string target, string workspace, string taskId) + /// Stop watching a task (after cancel); the stopped watch's record, null when none was running. + public async Task 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; + } + + /// + /// Resumes a watch stopped by whose cancel didn't go through (same id, push token and + /// start). No-op when the task is watched again already. + /// + 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)); } } diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs index 7255463..aee3b24 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs @@ -2,6 +2,13 @@ using A2A; namespace Volcanic.AgenticCore.Extensions.A2A; +internal static class RemoteAgentEntryExtensions +{ + /// Texts shown to the model or logged (exception messages) must never echo the agent's key. + public static string Scrub(this RemoteAgentEntry entry, string text) => + entry.Key is { Length: > 0 } key ? text.Replace(key, "[key]", StringComparison.Ordinal) : text; +} + /// A connected remote agent: its (extended) card and a client that sends the key. internal sealed class RemoteAgent(RemoteAgentEntry entry, AgentCard card, IA2AClient client, HttpClient http) : IDisposable { diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AHopsTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AHopsTests.cs index b0b990a..436715e 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AHopsTests.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AHopsTests.cs @@ -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); diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs index 7a3859e..e73533c 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs @@ -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() { diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs index 006e213..4d2dd57 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs @@ -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; }; diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs index a9b3b62..32349bd 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs @@ -20,11 +20,17 @@ public sealed class A2AClientHost : IAsyncDisposable private readonly ServiceProvider _services; private readonly List _sessions = []; - public A2AClientHost(RemoteAgentServer remote, A2AClientSettings? settings = null, params IExtension[] extra) + /// Wraps the handler that reaches the remote (e.g. a ). + public A2AClientHost(RemoteAgentServer remote, A2AClientSettings? settings = null, + Func? 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 => { diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/RemoteRewriter.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/RemoteRewriter.cs new file mode 100644 index 0000000..8f37da3 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/RemoteRewriter.cs @@ -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; + +/// +/// 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 SendMessage itself with a plain message instead of a task. +/// +public sealed class RemoteRewriter(HttpMessageHandler inner) : DelegatingHandler(inner) +{ + private int _subscribes; + + /// Rewrites "streaming":true to false in JSON responses (the public and the extended card). + public bool NoStreaming { get; init; } + + /// When set, SendMessage is answered with a message carrying this text (the remote never sees it). + public string? MessageReply { get; init; } + + /// When true, every request fails as if the remote were down. + public volatile bool Down; + + public int Subscribes => Volatile.Read(ref _subscribes); + + protected override async Task 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; + } +}