From 173b7dfc18c2dceefd0b52ce97e7efbc9c9733c1 Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Thu, 8 Oct 2026 13:49:06 +0000 Subject: [PATCH] feat(a2a-client): the a2a tool with agents, send, status and cancel, and per-session wiring --- .../A2AClientExtension.cs | 59 ++++ .../A2ATool.cs | 292 ++++++++++++++++++ .../A2A/A2AHopsTests.cs | 49 +++ .../A2A/A2AToolTests.cs | 252 +++++++++++++++ .../A2A/Support/A2AClientHost.cs | 108 +++++++ 5 files changed, 760 insertions(+) create mode 100644 extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs create mode 100644 extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs create mode 100644 tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AHopsTests.cs create mode 100644 tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs create mode 100644 tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs new file mode 100644 index 0000000..11b9f91 --- /dev/null +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs @@ -0,0 +1,59 @@ +using System.Text.Json; +using Microsoft.Extensions.Logging; +using Volcanic.AgenticCore.Abstractions.Extensions; +using Volcanic.AgenticCore.Abstractions.Extensions.Events; + +namespace Volcanic.AgenticCore.Extensions.A2A; + +/// +/// Per session (the host adds it, e.g. per Telegram chat; never host-wide): the a2a tool bound to the host's +/// , the hop counter from a2a.inbound, and the live-session delivery hook that also +/// flushes results kept while the target had no live conversation. +/// +public sealed class A2AClientExtension(A2AClientSettings settings, string target, A2AWatchRegistry? registry = null) : IExtension +{ + public const string ExtensionName = "a2a-client"; + + /// Emitted by the A2A server's per-session extension before each inbound prompt: {"hops": n, "caller": "…"}. + public const string InboundChannel = "a2a.inbound"; + + private readonly A2AWatchRegistry _registry = registry ?? A2AWatchRegistry.Shared; + private volatile int _hops; + + public string Name => ExtensionName; + + public ValueTask ActivateAsync(IExtensionApi api, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(api); + api.RegisterTool(new A2ATool(settings, target, _registry, () => _hops)); + api.Events.On(InboundChannel, data => + { + if (data is JsonElement { ValueKind: JsonValueKind.Object } e && e.TryGetProperty("hops", out var h) && h.TryGetInt32(out var n)) + _hops = n; + }); + api.On((_, ctx, _) => + { + var workspace = ctx.Cwd; + _registry.Live.Attach(target, workspace, api); + // Not inline: delivering starts a run, which mustn't happen inside session creation. + _ = Task.Run(async () => + { + try + { + await _registry.FlushUndeliveredAsync(target, workspace, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OutOfMemoryException) + { + api.Logger.LogWarning(ex, "A2A: flushing undelivered results for {Target} failed", target); + } + }); + return ValueTask.CompletedTask; + }); + api.On((_, _, _) => + { + _registry.Live.Detach(target, api); + return ValueTask.CompletedTask; + }); + return ValueTask.CompletedTask; + } +} diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs new file mode 100644 index 0000000..9ce938d --- /dev/null +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs @@ -0,0 +1,292 @@ +using System.ComponentModel; +using System.Net; +using System.Text; +using System.Text.Json; +using A2A; +using Volcanic.AgenticCore.Abstractions.Tools; +using A2AMessage = global::A2A.Message; + +namespace Volcanic.AgenticCore.Extensions.A2A; + +public sealed record A2AArgs( + [property: Description("\"agents\" (list remote agents), \"send\" (message an agent), \"status\" (check a task) or \"cancel\" (cancel a task)")] string Action, + [property: Description("Agent name from .volcanic/a2a/agents.json (send, status, cancel)")] string? Agent = null, + [property: Description("Text to send (send)")] string? Message = null, + [property: Description("Continue an earlier conversation with this agent: the context id from a previous result (send)")] string? ContextId = null, + [property: Description("A task id: answer the agent's question (send), or the task to check or cancel (status, cancel)")] string? TaskId = null, + [property: Description("Workspace files to attach (send)")] string[]? Files = null); + +/// The a2a tool: remote agents from agents.json. Errors are thrown with readable, key-free messages. +internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatchRegistry registry, Func inboundHops) : Tool +{ + /// Symlink hops followed when confining a path (a loop fails earlier in the OS). + private const int MaxLinkDepth = 40; + + public override string Name => "a2a"; + + public override string Description => + "Talk to other AI agents over the A2A protocol. Agents are configured in .volcanic/a2a/agents.json " + + "([{\"name\", \"url\", \"key\"}]). Use action=agents to see them and their skills, action=send to delegate work."; + + public override string? PromptSnippet => "a2a: delegate to other AI agents listed in .volcanic/a2a/agents.json"; + + public override IReadOnlyList PromptGuidelines => + [ + "When a remote agent asks a question (input-required), answer with action=send and its taskId; ask the user first if you don't know the answer.", + "Long remote tasks are watched for you: their result arrives later as a message starting with [A2A result from …].", + ]; + + protected override async Task ExecuteAsync(A2AArgs args, ToolInvocation invocation, CancellationToken cancellationToken) + { + var workspace = invocation.Context.Cwd; + return args.Action?.Trim().ToLowerInvariant() switch + { + "agents" => ToolResult.Text(await ListAsync(workspace, cancellationToken).ConfigureAwait(false)), + "send" => ToolResult.Text(await SendAsync(workspace, args, cancellationToken).ConfigureAwait(false)), + "status" => ToolResult.Text(await StatusAsync(workspace, args, cancellationToken).ConfigureAwait(false)), + "cancel" => ToolResult.Text(await CancelAsync(workspace, args, cancellationToken).ConfigureAwait(false)), + _ => throw new ArgumentException("action must be agents, send, status or cancel."), + }; + } + + private async Task ListAsync(string workspace, CancellationToken ct) + { + var file = AgentsFile.Read(workspace); + if (file.Error is not null) return file.Error; + if (file.Agents.Count == 0) + { + var none = "No remote agents configured. Add them to .volcanic/a2a/agents.json: [{\"name\": \"travel-bot\", \"url\": \"https://…\", \"key\": \"…\"}]"; + return file.Problems.Count == 0 ? none : none + string.Concat(file.Problems.Select(p => $"\n(skipped: {p})")); + } + var sb = new StringBuilder("Remote agents:"); + foreach (var entry in file.Agents) + { + try + { + using var agent = await registry.Connector(settings).ConnectAsync(entry, ct).ConfigureAwait(false); + var card = agent.Card; + sb.Append($"\n- {entry.Name}: {entry.Description ?? card.Description} ({card.Name})"); + var caps = new List(); + if (agent.Streaming) caps.Add("streaming"); + if (agent.SupportsPush) caps.Add("push"); + if (caps.Count > 0) sb.Append($" [{string.Join(", ", caps)}]"); + foreach (var skill in card.Skills ?? []) + { + sb.Append($"\n - skill {skill.Id}: {skill.Name} — {skill.Description}"); + if (skill.Examples is { Count: > 0 } examples) sb.Append($" (e.g. {string.Join("; ", examples.Take(3))})"); + } + } + catch (Exception ex) when (ex is not OutOfMemoryException && (ex is not OperationCanceledException || !ct.IsCancellationRequested)) + { + sb.Append($"\n- {entry.Name}: unreachable ({Clean(ex.Message, entry)})"); + } + } + foreach (var problem in file.Problems) sb.Append($"\n(skipped: {problem})"); + return sb.ToString(); + } + + private async Task SendAsync(string workspace, A2AArgs args, CancellationToken ct) + { + var entry = Entry(workspace, args.Agent); + if (string.IsNullOrWhiteSpace(args.Message) && args.Files is not { Length: > 0 }) + 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)); + + return await CallAsync(entry, ct, async agent => + { + var response = await agent.Client.SendMessageAsync(new SendMessageRequest + { + 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) }, + }, ct).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); + 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."; + }).ConfigureAwait(false); + } + + 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); + return A2AResults.IsSettled(task.Status.State) ? DescribeSettled(workspace, entry.Name, task) : A2AResults.Describe(entry.Name, task, []); + }).ConfigureAwait(false); + } + + private async Task CancelAsync(string workspace, A2AArgs args, CancellationToken ct) + { + 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 => + { + try + { + 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); + } + + private string DescribeSettled(string workspace, string agent, AgentTask task) + { + var saved = A2AResults.SaveFiles(workspace, agent, task, settings.MaxFileBytes); + 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. + 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 + { + if (agent.Streaming) + { + 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; + } + } + else + { + 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; + } + } + } + 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); + } + + private async Task CallAsync(RemoteAgentEntry entry, CancellationToken ct, Func> call) + { + try + { + using var agent = await registry.Connector(settings).ConnectAsync(entry, ct).ConfigureAwait(false); + return await call(agent).ConfigureAwait(false); + } + catch (HttpRequestException ex) when (ex.StatusCode is HttpStatusCode.Unauthorized or HttpStatusCode.Forbidden) + { + throw new InvalidOperationException($"Agent \"{entry.Name}\" rejected the key ({(int)ex.StatusCode!}). Check its key in .volcanic/a2a/agents.json."); + } + catch (HttpRequestException ex) + { + throw new InvalidOperationException($"Could not reach agent \"{entry.Name}\": {Clean(ex.Message, entry)}"); + } + catch (A2AException ex) + { + throw new InvalidOperationException($"Agent \"{entry.Name}\" refused: {Clean(ex.Message, entry)} ({ex.ErrorCode})."); + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested) + { + throw new InvalidOperationException($"Agent \"{entry.Name}\" did not answer in time."); + } + } + + private static RemoteAgentEntry Entry(string workspace, string? name) + { + var file = AgentsFile.Read(workspace); + if (file.Error is not null) throw new InvalidOperationException(file.Error); + if (string.IsNullOrWhiteSpace(name)) throw new ArgumentException("agent is required."); + return file.Agents.FirstOrDefault(a => a.Name == name) + ?? 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) + { + 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)); + } + + /// 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) + { + if (string.IsNullOrWhiteSpace(path)) throw new ArgumentException("a file path is empty."); + var root = Path.GetFullPath(workspace); + var full = Path.GetFullPath(Path.Combine(root, path)); + if (!IsInside(full, root)) throw new ArgumentException($"{path} is outside the workspace."); + if (!IsInside(RealPath(full, 0), RealPath(root, 0))) throw new ArgumentException($"{path} points outside the workspace."); + return full; + } + + private static bool IsInside(string path, string root) + { + var prefix = root.TrimEnd(Path.DirectorySeparatorChar) + Path.DirectorySeparatorChar; + return path.StartsWith(prefix, StringComparison.Ordinal); + } + + /// Resolves symlinks in every existing component of an absolute path. + private static string RealPath(string full, int depth) + { + if (depth > MaxLinkDepth) throw new ArgumentException("too many symbolic links."); + var current = Path.GetPathRoot(full)!; + foreach (var segment in full[current.Length..].Split(Path.DirectorySeparatorChar, StringSplitOptions.RemoveEmptyEntries)) + { + var next = Path.Combine(current, segment); + FileSystemInfo info = Directory.Exists(next) ? new DirectoryInfo(next) : new FileInfo(next); + // A link's target may itself sit under links: resolve it again from the top. + if (info.LinkTarget is not null && info.ResolveLinkTarget(true) is { } resolved) next = RealPath(resolved.FullName, depth + 1); + current = next; + } + return current; + } + + private static string MediaType(string path) => Path.GetExtension(path).ToLowerInvariant() switch + { + ".png" => "image/png", + ".jpg" or ".jpeg" => "image/jpeg", + ".pdf" => "application/pdf", + ".txt" => "text/plain", + ".md" => "text/markdown", + ".json" => "application/json", + ".csv" => "text/csv", + _ => "application/octet-stream", + }; + + 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/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AHopsTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AHopsTests.cs new file mode 100644 index 0000000..b0b990a --- /dev/null +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AHopsTests.cs @@ -0,0 +1,49 @@ +using Volcanic.AgenticCore.Abstractions.Extensions; +using Volcanic.AgenticCore.Abstractions.Extensions.Events; +using Volcanic.AgenticCore.Abstractions.Json; +using Volcanic.AgenticCore.Extensions.A2A; +using Volcanic.AgenticCore.Extensions.Tests.A2A.Support; +using Volcanic.AgenticCore.Testing; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A; + +public class A2AHopsTests +{ + /// Plays the server's per-session extension: emits a2a.inbound like an inbound A2A run would. + private sealed class InboundEmitter(int hops) : IExtension + { + public string Name => "inbound-emitter"; + public ValueTask ActivateAsync(IExtensionApi api, CancellationToken cancellationToken) + { + api.On((_, _, _) => + { + api.Events.Emit(A2AClientExtension.InboundChannel, VolcanicJson.ToElement(new { hops, caller = "upstream" })); + return ValueTask.CompletedTask; + }); + return ValueTask.CompletedTask; + } + } + + [Fact] + public async Task Hops_Incremented_WithoutInbound() + { + await using var remote = await RemoteAgentServer.StartAsync(o => o.MaxHops = 1); + remote.Faux.Enqueue(FauxTurn.Say("fine")); + await using var host = new A2AClientHost(remote); + var session = await host.CreateSessionAsync(); + var (text, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "hi" }); + Assert.Contains("completed", text); // sent with hops 1, allowed at MaxHops 1 + } + + [Fact] + 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)); + var session = await host.CreateSessionAsync(); + var (text, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "hi" }); + Assert.Contains("failed", text); + Assert.Contains("hop limit", text); + Assert.Empty(remote.Faux.Requests); + } +} diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs new file mode 100644 index 0000000..7a3859e --- /dev/null +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs @@ -0,0 +1,252 @@ +using Volcanic.AgenticCore.Extensions.A2A; +using Volcanic.AgenticCore.Extensions.Tests.A2A.Support; +using Volcanic.AgenticCore.Testing; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A; + +public class A2AToolTests +{ + private static string Field(string text, string label) => + System.Text.RegularExpressions.Regex.Match(text, label + @" ([A-Za-z0-9_-]+)").Groups[1].Value; + + private static async Task WaitUntilAsync(Func 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); + } + } + + [Fact] + public async Task Agents_ListsCardsAndSkills() + { + await using var remote = await RemoteAgentServer.StartAsync(); + File.WriteAllText(Path.Combine(remote.Workspace, ".volcanic", "a2a", "card.json"), + """{ "name": "Remote travel", "description": "Plans trips", "skills": [ { "id": "trips", "name": "Trips", "description": "Plans trips", "tags": ["travel"], "examples": ["Find a flight"] } ] }"""); + await using var host = new A2AClientHost(remote); + var session = await host.CreateSessionAsync(); + var (text, isError) = await host.CallToolAsync(session, new { action = "agents" }); + Assert.False(isError); + Assert.Contains("remote", text); + Assert.Contains("Plans trips", text); + Assert.Contains("trips", text); + Assert.Contains("Find a flight", text); + Assert.DoesNotContain(RemoteAgentServer.Key, text); + } + + [Fact] + public async Task Agents_UnreachableAgent_Listed() + { + await using var remote = await RemoteAgentServer.StartAsync(); + await using var host = new A2AClientHost(remote, new A2AClientSettings { Enabled = true, CardTimeoutSeconds = 2 }); + host.WriteAgents("""[ { "name": "ghost", "url": "http://nowhere.invalid/missing/card.json", "key": "ghost-key-0123456789" } ]"""); + var session = await host.CreateSessionAsync(); + var (text, isError) = await host.CallToolAsync(session, new { action = "agents" }); + Assert.False(isError); + Assert.Contains("ghost", text); + Assert.Contains("unreachable", text); + Assert.DoesNotContain("ghost-key-0123456789", text); + } + + [Fact] + public async Task Send_CompletesWithinWait_ReturnsAnswerAndIds() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Paris is lovely")); + await using var host = new A2AClientHost(remote); + var session = await host.CreateSessionAsync(); + var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "Where should I go?" }); + Assert.False(isError); + Assert.Contains("completed", text); + Assert.Contains("Paris is lovely", text); + Assert.NotEmpty(Field(text, "task")); + Assert.NotEmpty(Field(text, "context")); + Assert.Empty(host.Registry.List(A2AClientHost.Target)); + } + + [Fact] + public async Task Send_ContinuesContext() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("first"), FauxTurn.Say("second")); + await using var host = new A2AClientHost(remote); + var session = await host.CreateSessionAsync(); + var (first, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "one" }); + var (second, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "two", contextId = Field(first, "context") }); + Assert.Contains("second", second); + Assert.Equal(Field(first, "context"), Field(second, "context")); + Assert.Contains("first", string.Join("\n", remote.Faux.Requests.Last().Messages.Select(m => m.Text))); + } + + [Fact] + public async Task Send_InputRequired_ReturnsQuestion_AndAnswerContinues() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Tool("request_input", new { question = "Which date?" }), FauxTurn.Say("Booked Friday")); + await using var host = new A2AClientHost(remote); + var session = await host.CreateSessionAsync(); + var (question, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "Book a table" }); + Assert.Contains("input-required", question); + Assert.Contains("Which date?", question); + var (answer, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "Friday", taskId = Field(question, "task") }); + Assert.Contains("completed", answer); + Assert.Contains("Booked Friday", answer); + } + + [Fact] + public async Task Send_SlowTask_StartsWatch_AndResultArrivesAsFollowUp() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Slow answer").WithDelay(TimeSpan.FromMilliseconds(1500))); + await using var host = new A2AClientHost(remote, new A2AClientSettings { Enabled = true, WaitSeconds = 0 }); + var session = await host.CreateSessionAsync(); + // The follow-up run gets the faux default answer. + var (text, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "take your time" }); + Assert.True(text.Contains("working") || text.Contains("submitted"), text); + Assert.Contains("delivered", text); + Assert.Single(host.Registry.List(A2AClientHost.Target)); + var delivered = await host.WaitForUserMessageAsync(session, "[A2A result from \"remote\""); + Assert.Contains("Slow answer", delivered); + Assert.Contains(host.Faux.Requests, r => r.Messages.Any(m => m.Text.Contains("Slow answer"))); + Assert.Empty(host.Registry.List(A2AClientHost.Target)); + await WaitUntilAsync(() => PendingFile.Read(host.Workspace).Count == 0); // removed right after the hand-over + } + + [Fact] + public async Task SessionStart_FlushesUndeliveredResult_AsFollowUp() + { + await using var remote = await RemoteAgentServer.StartAsync(); + await using var host = new A2AClientHost(remote); + PendingFile.Upsert(host.Workspace, new WatchRecord("w1", A2AClientHost.Target, "remote", "t1", "c1", null, DateTimeOffset.UtcNow, + 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.StartsWith("[A2A result from \"remote\"", delivered); + await WaitUntilAsync(() => PendingFile.Read(host.Workspace).Count == 0); + } + + [Fact] + public async Task Send_FilesArgument_SentAsParts_AndFileArtifactSaved() + { + await using var remote = await RemoteAgentServer.StartAsync(); + File.WriteAllText(Path.Combine(remote.Workspace, "reply.txt"), "remote file"); + remote.Faux.Enqueue(FauxTurn.Tool("send", new { path = "reply.txt" }), FauxTurn.Say("sent it")); + await using var host = new A2AClientHost(remote); + File.WriteAllText(Path.Combine(host.Workspace, "notes.txt"), "my notes"); + var session = await host.CreateSessionAsync(); + var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "here", files = new[] { "notes.txt" } }); + Assert.False(isError, text); + Assert.Contains(remote.Faux.Requests.First().Messages, m => m.Text.Contains("Files: inbox/a2a-client-")); + Assert.Contains("inbox/a2a/remote/reply.txt", text); + Assert.Equal("remote file", File.ReadAllText(Path.Combine(host.Workspace, "inbox", "a2a", "remote", "reply.txt"))); + var inbound = Assert.Single(Directory.GetFiles(Path.Combine(remote.Workspace, "inbox"), "a2a-client-*")); + Assert.Equal("my notes", File.ReadAllText(inbound)); + } + + [Fact] + public async Task Send_FileOutsideWorkspace_Error() + { + await using var remote = await RemoteAgentServer.StartAsync(); + await using var host = new A2AClientHost(remote); + var session = await host.CreateSessionAsync(); + var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "x", files = new[] { "../../etc/passwd" } }); + Assert.True(isError); + Assert.Contains("workspace", text); + Assert.Empty(remote.Faux.Requests); + } + + [Theory] + [InlineData(false)] + [InlineData(true)] + public async Task Send_SymlinkOutsideWorkspace_Error(bool directoryLink) + { + await using var remote = await RemoteAgentServer.StartAsync(); + await using var host = new A2AClientHost(remote); + var outside = Path.Combine(Path.GetTempPath(), "a2a-outside", Guid.NewGuid().ToString("N")); + Directory.CreateDirectory(outside); + try + { + File.WriteAllText(Path.Combine(outside, "secret.txt"), "secret"); + string path; + if (directoryLink) + { + Directory.CreateSymbolicLink(Path.Combine(host.Workspace, "linked"), outside); + path = "linked/secret.txt"; + } + else + { + File.CreateSymbolicLink(Path.Combine(host.Workspace, "secret.txt"), Path.Combine(outside, "secret.txt")); + path = "secret.txt"; + } + var session = await host.CreateSessionAsync(); + var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "x", files = new[] { path } }); + Assert.True(isError); + Assert.Contains("outside the workspace", text); + Assert.Empty(remote.Faux.Requests); + } + finally + { + Directory.Delete(outside, true); + } + } + + [Fact] + public async Task Send_UnknownAgent_Error() + { + await using var remote = await RemoteAgentServer.StartAsync(); + await using var host = new A2AClientHost(remote); + var session = await host.CreateSessionAsync(); + var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "nobody", message = "hi" }); + Assert.True(isError); + Assert.Contains("nobody", text); + Assert.Contains("remote", text); // lists the configured ones + } + + [Fact] + public async Task Send_401_ReadableError_AndResultNeverContainsKey() + { + await using var remote = await RemoteAgentServer.StartAsync(); + await using var host = new A2AClientHost(remote); + host.WriteAgents($$"""[ { "name": "remote", "url": "{{RemoteAgentServer.BaseUrl}}", "key": "wrong-key-abcdefghij" } ]"""); + var session = await host.CreateSessionAsync(); + var (text, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "hi" }); + Assert.True(isError); + Assert.Contains("rejected", text); + Assert.DoesNotContain("wrong-key-abcdefghij", text); + } + + [Fact] + public async Task Status_ReturnsTask_AndCancel_CancelsRemoteAndDropsWatch() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("slow").WithDelay(TimeSpan.FromSeconds(10))); + await using var host = new A2AClientHost(remote, new A2AClientSettings { Enabled = true, WaitSeconds = 0 }); + var session = await host.CreateSessionAsync(); + var (started, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "go" }); + 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); + var (cancelled, isError) = await host.CallToolAsync(session, new { action = "cancel", agent = "remote", taskId }); + Assert.False(isError, cancelled); + Assert.Contains("canceled", cancelled); + Assert.Empty(host.Registry.List(A2AClientHost.Target)); + Assert.Empty(PendingFile.Read(host.Workspace)); + Assert.DoesNotContain(host.UserTexts(), t => t.Contains("[A2A result", StringComparison.Ordinal)); + } + + [Fact] + public async Task NoAgentsConfigured_ExplainsFile() + { + await using var remote = await RemoteAgentServer.StartAsync(); + await using var host = new A2AClientHost(remote); + File.Delete(AgentsFile.PathFor(host.Workspace)); + var session = await host.CreateSessionAsync(); + var (text, isError) = await host.CallToolAsync(session, new { action = "agents" }); + Assert.False(isError); + Assert.Contains(".volcanic/a2a/agents.json", text); + } +} diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs new file mode 100644 index 0000000..a9b3b62 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs @@ -0,0 +1,108 @@ +using Microsoft.Extensions.DependencyInjection; +using Volcanic.AgenticCore.Abstractions.Events; +using Volcanic.AgenticCore.Abstractions.Extensions; +using Volcanic.AgenticCore.Abstractions.Messages; +using Volcanic.AgenticCore.Extensions.A2A; +using Volcanic.AgenticCore.Features.Agent; +using Volcanic.AgenticCore.Features.Extensions; +using Volcanic.AgenticCore.Features.Sessions; +using Volcanic.AgenticCore.Hosting; +using Volcanic.AgenticCore.Testing; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A.Support; + +/// Our assistant: a core host whose sessions get (target "chat-1"). +public sealed class A2AClientHost : IAsyncDisposable +{ + public const string Target = "chat-1"; + + private readonly string _root = Path.Combine(Path.GetTempPath(), "a2a-client", Guid.NewGuid().ToString("N")); + private readonly ServiceProvider _services; + private readonly List _sessions = []; + + public A2AClientHost(RemoteAgentServer remote, A2AClientSettings? settings = 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)] }; + var services = new ServiceCollection(); + services.AddAgenticCore(o => + { + o.AgentDir = Path.Combine(_root, "agent"); + o.DiscoverExtensionAssemblies = false; + o.DefaultModel = FauxProvider.ModelKey; + o.SessionStoreFactory = _ => new InMemorySessionStore(); + o.ProcessEnvironment.HardenHostProcess = false; + o.Retry = new RetrySettings { Enabled = false }; + o.Extensions.Add(ExtensionRegistration.FromInstance(Faux.Extension)); + foreach (var e in extra) o.Extensions.Add(ExtensionRegistration.FromInstance(e)); + }); + _services = services.BuildServiceProvider(); + WriteAgents($$"""[ { "name": "remote", "url": "{{RemoteAgentServer.BaseUrl}}", "key": "{{RemoteAgentServer.Key}}" } ]"""); + } + + public FauxProvider Faux { get; } = new(); + public A2AClientSettings Settings { get; } + public A2AWatchRegistry Registry { get; } + public string Workspace => Path.Combine(_root, "ws"); + public List Events { get; } = []; + + public void WriteAgents(string json) + { + Directory.CreateDirectory(Path.GetDirectoryName(AgentsFile.PathFor(Workspace))!); + File.WriteAllText(AgentsFile.PathFor(Workspace), json); + } + + public async Task CreateSessionAsync() + { + var session = await _services.GetRequiredService().CreateAsync(new AgentSessionOptions + { + Cwd = Workspace, + ExtraExtensions = [ExtensionRegistration.FromInstance(new A2AClientExtension(Settings, Target, Registry))], + }); + session.Subscribe(e => { lock (Events) Events.Add(e); }); + lock (_sessions) _sessions.Add(session); + return session; + } + + /// Has the local model call the a2a tool once, then say "ok"; returns the tool result text and error flag. + public async Task<(string Text, bool IsError)> CallToolAsync(AgentSession session, object args) + { + Faux.Enqueue(FauxTurn.Tool("a2a", args), FauxTurn.Say("ok")); + await session.PromptAsync("use the remote agent"); + AgentSessionEvent.ToolExecutionEnd end; + lock (Events) end = Events.OfType().Last(e => e.ToolName == "a2a"); + return (end.Result.TextContent, end.IsError); + } + + /// Texts of the user messages the sessions ran so far (prompts and follow-ups), from the session events. + public IReadOnlyList UserTexts() + { + lock (Events) + return Events.OfType().Where(e => e.Message is UserMessage).Select(e => e.Message.Text).ToList(); + } + + /// Waits until a user message containing ran, then until that run settled. + public async Task WaitForUserMessageAsync(AgentSession session, string text, int timeoutMs = 15_000) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (true) + { + if (UserTexts().FirstOrDefault(t => t.Contains(text, StringComparison.Ordinal)) is { } found) + { + await session.WaitForIdleAsync(); + return found; + } + if (DateTime.UtcNow > deadline) throw new TimeoutException($"no user message containing \"{text}\""); + await Task.Delay(20); + } + } + + public async ValueTask DisposeAsync() + { + await Registry.DisposeAsync(); + foreach (var session in _sessions) await session.DisposeAsync(); + await _services.DisposeAsync(); + try { Directory.Delete(_root, true); } catch (IOException) { } catch (UnauthorizedAccessException) { } + } +}