From e70557725acc759583a22f48142e5f208462c0fc Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Thu, 8 Oct 2026 12:29:29 +0000 Subject: [PATCH] docs(plan): A2A client implementation plan; spec revisions --- .../plans/2026-10-08-a2a-client.md | 2504 +++++++++++++++++ .../specs/2026-10-07-a2a-client-design.md | 43 +- 2 files changed, 2539 insertions(+), 8 deletions(-) create mode 100644 docs/superpowers/plans/2026-10-08-a2a-client.md diff --git a/docs/superpowers/plans/2026-10-08-a2a-client.md b/docs/superpowers/plans/2026-10-08-a2a-client.md new file mode 100644 index 0000000..2da46ca --- /dev/null +++ b/docs/superpowers/plans/2026-10-08-a2a-client.md @@ -0,0 +1,2504 @@ +# A2A client Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Let the owner's assistant call other A2A agents: an `a2a` tool (`agents | send | status | cancel`) over a plain `agents.json` in the workspace, with long tasks handed to watches that deliver results later as follow-up prompts. Off by default behind `A2AClient__Enabled`. + +**Architecture:** A feature extension `extensions/Volcanic.AgenticCore.Extensions.A2A` (references only Abstractions + the official `A2A` 1.0.0-preview2 client package). A process-wide `A2AWatchRegistry` owns one HTTP handler, the watches (subscribe / poll / push) and `pending.json` persistence; it delivers through `IA2AResultDelivery` (default `LiveSessionDelivery`; Telegram: `TelegramA2AResultDelivery`). The Telegram host adds `A2AClientExtension` per chat session (never to A2A server sessions) when `A2AClient:Enabled` is on, restores watches at startup and, when the server is on too, routes the server's push receiver into the registry. + +**Tech Stack:** .NET 10, `A2A` 1.0.0-preview2, xunit 2, `Microsoft.AspNetCore.TestHost` (tests run against the in-process `Volcanic.AgenticCore.A2A.Server`), `Microsoft.Extensions.TimeProvider.Testing`, the core `FauxProvider`. + +**Spec:** `docs/superpowers/specs/2026-10-07-a2a-client-design.md`, including its "Revisions (2026-10-08)" section, which wins over older sections. + +## Facts the engineer needs + +- **Namespace gotcha (bit us in the server):** our namespace `Volcanic.AgenticCore.Extensions.A2A` contains a segment `A2A`, so a *qualified* SDK reference `A2A.X` resolves to our namespace. Write `global::A2A.X` when qualifying. `Message`, `TaskStatus`, `Part` may collide with `Volcanic.AgenticCore.Abstractions.Messages.*` / `System.Threading.Tasks.TaskStatus`: alias them, e.g. `using A2AMessage = global::A2A.Message;`. +- SDK client (verified by reflection on the installed DLL): + - `new A2ACardResolver(Uri baseUrl, HttpClient? httpClient = null, string agentCardPath = "/.well-known/agent-card.json", ILogger? logger = null).GetAgentCardAsync(ct)` + - `A2AClientFactory.Create(AgentCard card, HttpClient? http = null, A2AClientOptions? options = null)` → `IA2AClient` (picks the binding from `card.SupportedInterfaces`). + - `IA2AClient`: `SendMessageAsync(SendMessageRequest)` → `SendMessageResponse { Task | Message }`, `GetTaskAsync(new GetTaskRequest { Id })`, `CancelTaskAsync(new CancelTaskRequest { Id })`, `SubscribeToTaskAsync(new SubscribeToTaskRequest { Id })` → `IAsyncEnumerable`, `CreateTaskPushNotificationConfigAsync(new CreateTaskPushNotificationConfigRequest { TaskId, Config = new PushNotificationConfig { Url, Token } })`, `DeleteTaskPushNotificationConfigAsync(new DeleteTaskPushNotificationConfigRequest { TaskId, Id })`, `GetExtendedAgentCardAsync(new GetExtendedAgentCardRequest())`. + - `TaskProjection.Apply(AgentTask? current, StreamResponse ev)` → `AgentTask?` (public static). + - `TaskStateExtensions.IsTerminal(state)`; states `Submitted, Working, Completed, Failed, Canceled, InputRequired, Rejected, AuthRequired`. + - `AgentCard.Capabilities { Streaming, PushNotifications, ExtendedAgentCard }` (nullable bools), `SecuritySchemes: Dictionary` with `ApiKeySecurityScheme { Name, Location }` / `HttpAuthSecurityScheme { Scheme }`. + - `Part.FromText / FromRaw(bytes, mediaType, filename)`, `part.ContentCase` (`Text, Raw, Url, Data`), `Message.Metadata` / `SendMessageRequest.Metadata`: `Dictionary?`, `SendMessageConfiguration { ReturnImmediately }`. + - Errors: JSON-RPC errors → `A2AException { ErrorCode }`; HTTP failures (e.g. 401) → `HttpRequestException { StatusCode }`. +- Our server (in this repo) is the remote in tests: `src/Volcanic.AgenticCore.A2A.Server` — `AddA2AServer`, `MapA2AServer`, `WorkspaceCallerDirectory`, `A2AWorkspace`, `CallersFile`, `A2AHostOptions { OnPush, AllowPrivatePushUrls, MaxHops }`, `HttpA2APushSender(HttpClient, TimeProvider, ILogger)` (internal; the server project grants `InternalsVisibleTo` only to its own tests and Telegram tests — Task 1 adds the extension tests). Its card advertises `streaming` and `pushNotifications`, `X-API-Key` auth, and it honours `returnImmediately`. A context id must match `^[A-Za-z0-9_-]{1,100}$`. +- Core: `IExtension`, `IExtensionApi` (`RegisterTool`, `On(ExtensionHandler)`, `Events.On(channel, Action)`, `SendUserMessage(text, images, MessageDelivery.FollowUp)`), `SessionStartEvent`, `SessionShutdownEvent`, `Tool` (throw → error result), `ToolInvocation.Context.Cwd`. Test hosts: copy the shape of `tests/Volcanic.AgenticCore.Extensions.Tests/Mcp/Support/McpTestHost.cs`. + +## File structure + +``` +extensions/Volcanic.AgenticCore.Extensions.A2A/ + Volcanic.AgenticCore.Extensions.A2A.csproj + A2AClientSettings.cs settings (section A2AClient) + AgentsFile.cs /.volcanic/a2a/agents.json + RemoteAgents.cs RemoteAgentConnector + RemoteAgent (card, auth header, IA2AClient) + A2AResults.cs state names, text extraction, artifact saving, result texts + PendingFile.cs WatchRecord + /.volcanic/a2a/pending.json + A2AResultDelivery.cs IA2AResultDelivery + LiveSessionDelivery + A2AWatchRegistry.cs watches, push, restore, delivery, waiting helper + A2ATool.cs the a2a tool + A2AClientExtension.cs per-session wiring (tool, hops, session start/shutdown) +tests/Volcanic.AgenticCore.Extensions.Tests/A2A/ + Support/RemoteAgentServer.cs, Support/A2AClientHost.cs, Support/PushToRegistryHandler.cs + AgentsFileTests.cs, A2AResultsTests.cs, PendingFileTests.cs, RemoteAgentsTests.cs, + A2AToolTests.cs, A2AWatchTests.cs, A2AHopsTests.cs +hosts/Volcanic.AgenticCore.Telegram/ + Chats/ChatFollowUps.cs (extracted), A2A/TelegramA2AResultDelivery.cs, A2A/A2AWatchHostedService.cs, + A2A/TelegramA2AClientSetup.cs; ChatRegistry.cs, ChatCommands.cs, Program.cs, TelegramJobDelivery.cs modified +``` + +--- + +### Task 1: Project, settings, test wiring + +**Files:** +- Create: `extensions/Volcanic.AgenticCore.Extensions.A2A/Volcanic.AgenticCore.Extensions.A2A.csproj` +- Create: `extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientSettings.cs` +- Modify: `VolcanicAgenticCore.slnx` (add under `/extensions/`) +- Modify: `tests/Volcanic.AgenticCore.Extensions.Tests/Volcanic.AgenticCore.Extensions.Tests.csproj` +- Modify: `src/Volcanic.AgenticCore.A2A.Server/Volcanic.AgenticCore.A2A.Server.csproj` (InternalsVisibleTo the extension tests) +- Test: `tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AClientSettingsTests.cs` + +- [ ] **Step 1: Project file** + +```xml + + + A2A client extension: the a2a tool, remote agents in agents.json, watches for long tasks. + + + + + + + + + + + +``` + +- [ ] **Step 2: Settings** + +```csharp +namespace Volcanic.AgenticCore.Extensions.A2A; + +/// Section A2AClient (env A2AClient__…). Named so it doesn't clash with the SDK's A2AClientOptions. +public sealed class A2AClientSettings +{ + public const string SectionName = "A2AClient"; + + /// Off by default: no a2a tool, no watches. Independent of the server's A2A:Enabled. + public bool Enabled { get; set; } + + /// How long send waits for a result before a watch takes over. 0 hands over at once. + public int WaitSeconds { get; set; } = 60; + + public TimeSpan WatchTimeout { get; set; } = TimeSpan.FromDays(1); + public int CardTimeoutSeconds { get; set; } = 10; + public long MaxFileBytes { get; set; } = 10 * 1024 * 1024; + + /// + /// Our public base URL for push webhooks ({PushBaseUrl}/a2a/push/{watchId}). Set by the host from the server's + /// A2A:PublicUrl only while the server (and so the push receiver) is on; null = subscribe or poll. + /// + public string? PushBaseUrl { get; set; } +} +``` + +- [ ] **Step 3: Test project and server internals** + +In `tests/Volcanic.AgenticCore.Extensions.Tests/Volcanic.AgenticCore.Extensions.Tests.csproj` add: + +```xml + + + + +``` + +(`FrameworkReference` goes in its own `` or next to the package references; both are fine.) In the server csproj add ``. Add the extension project to `VolcanicAgenticCore.slnx` under `/extensions/`. + +- [ ] **Step 4: Smoke test** + +`tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AClientSettingsTests.cs`: + +```csharp +using Volcanic.AgenticCore.Extensions.A2A; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A; + +public class A2AClientSettingsTests +{ + [Fact] + public void Defaults_OffWithSpecValues() + { + var s = new A2AClientSettings(); + Assert.False(s.Enabled); + Assert.Equal(60, s.WaitSeconds); + Assert.Equal(TimeSpan.FromDays(1), s.WatchTimeout); + Assert.Equal(10, s.CardTimeoutSeconds); + Assert.Equal(10 * 1024 * 1024, s.MaxFileBytes); + Assert.Null(s.PushBaseUrl); + Assert.Equal("A2AClient", A2AClientSettings.SectionName); + } +} +``` + +- [ ] **Step 5: Build and test** + +Run: `dotnet build && dotnet test tests/Volcanic.AgenticCore.Extensions.Tests --filter "FullyQualifiedName~A2AClientSettingsTests"` +Expected: 0 warnings; 1 passed. (If NuGet reports NU1510 or a version conflict for packages that `Microsoft.AspNetCore.App` already provides — e.g. `Microsoft.Extensions.DependencyInjection` / `Logging` — remove those explicit references from the test csproj; warnings are not errors in test projects, but keep the output clean.) + +- [ ] **Step 6: Commit** + +```bash +git add extensions/Volcanic.AgenticCore.Extensions.A2A tests/Volcanic.AgenticCore.Extensions.Tests VolcanicAgenticCore.slnx src/Volcanic.AgenticCore.A2A.Server/Volcanic.AgenticCore.A2A.Server.csproj +git commit -m "feat(a2a-client): scaffold the A2A client extension behind A2AClient__Enabled" +``` + +--- + +### Task 2: agents.json + +**Files:** +- Create: `extensions/Volcanic.AgenticCore.Extensions.A2A/AgentsFile.cs` +- Test: `tests/Volcanic.AgenticCore.Extensions.Tests/A2A/AgentsFileTests.cs` + +- [ ] **Step 1: Failing tests** + +```csharp +using Volcanic.AgenticCore.Extensions.A2A; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A; + +public sealed class AgentsFileTests : IDisposable +{ + private readonly string _ws = Path.Combine(Path.GetTempPath(), "a2a-agents", Guid.NewGuid().ToString("N")); + + private void Write(string json) + { + Directory.CreateDirectory(Path.GetDirectoryName(AgentsFile.PathFor(_ws))!); + File.WriteAllText(AgentsFile.PathFor(_ws), json); + } + + [Fact] + public void Missing_NoAgentsNoError() + { + var r = AgentsFile.Read(_ws); + Assert.Empty(r.Agents); + Assert.Null(r.Error); + } + + [Fact] + public void Agents_MalformedFile_ReportsFile() + { + Write("{ nope"); + var r = AgentsFile.Read(_ws); + Assert.Empty(r.Agents); + Assert.Contains("agents.json", r.Error); + } + + [Fact] + public void Read_ValidatesEntries_AndNeverEchoesKeys() + { + Write(""" + [ { "name": "travel-bot", "url": "https://travel.example/", "key": "secret-1", "description": "Trips" }, + { "name": "Bad Name", "url": "https://x.example/", "key": "secret-2" }, + { "name": "no-url" }, + { "name": "ftp", "url": "ftp://x.example/" }, + { "name": "travel-bot", "url": "https://dup.example/" } ] + """); + var r = AgentsFile.Read(_ws); + var agent = Assert.Single(r.Agents); + Assert.Equal(new RemoteAgentEntry("travel-bot", "https://travel.example/", "secret-1", "Trips"), agent); + Assert.Equal(4, r.Problems.Count); + Assert.DoesNotContain(r.Problems, p => p.Contains("secret")); + } + + [Fact] + public void Agents_FileEdited_AppliesToNextCall() + { + Write("""[ { "name": "a", "url": "https://a.example/" } ]"""); + Assert.NotNull(AgentsFile.Find(_ws, "a")); + Write("[]"); + Assert.Null(AgentsFile.Find(_ws, "a")); + } + + public void Dispose() { try { Directory.Delete(_ws, true); } catch (IOException) { } } +} +``` + +- [ ] **Step 2: Run to see it fail** + +Run: `dotnet test tests/Volcanic.AgenticCore.Extensions.Tests --filter "FullyQualifiedName~AgentsFileTests"` +Expected: build error (`AgentsFile` missing). + +- [ ] **Step 3: Implement** + +```csharp +using System.Text.Json; +using System.Text.RegularExpressions; + +namespace Volcanic.AgenticCore.Extensions.A2A; + +public sealed record RemoteAgentEntry(string Name, string Url, string? Key, string? Description); + +/// Skipped entries, described without keys. +public sealed record AgentsFileResult(IReadOnlyList Agents, IReadOnlyList Problems, string? Error); + +/// +/// <workspace>/.volcanic/a2a/agents.json: remote agents the assistant may call. A plain file the agent and the +/// user edit; read on every call (no cache). +/// +public static partial class AgentsFile +{ + [GeneratedRegex("^[a-z0-9-]{1,40}$", RegexOptions.CultureInvariant)] + private static partial Regex NamePattern(); + + public static string PathFor(string workspace) => Path.Combine(workspace, ".volcanic", "a2a", "agents.json"); + + public static RemoteAgentEntry? Find(string workspace, string name) => + Read(workspace).Agents.FirstOrDefault(a => a.Name == name); + + public static AgentsFileResult Read(string workspace) + { + var path = PathFor(workspace); + if (!File.Exists(path)) return new([], [], null); + JsonElement root; + try + { + using var doc = JsonDocument.Parse(File.ReadAllText(path)); + root = doc.RootElement.Clone(); + } + catch (Exception ex) when (ex is JsonException or IOException or UnauthorizedAccessException) + { + return new([], [], $".volcanic/a2a/agents.json could not be read: {ex.GetType().Name}."); + } + if (root.ValueKind != JsonValueKind.Array) + return new([], [], ".volcanic/a2a/agents.json must be a JSON array of agents."); + + var agents = new List(); + var problems = new List(); + var seen = new HashSet(StringComparer.Ordinal); + var index = 0; + foreach (var item in root.EnumerateArray()) + { + index++; + if (item.ValueKind != JsonValueKind.Object) { problems.Add($"entry {index} is not an object"); continue; } + var name = Str(item, "name"); + var url = Str(item, "url"); + if (name is null || !NamePattern().IsMatch(name)) { problems.Add($"entry {index}: name must match [a-z0-9-]{{1,40}}"); continue; } + if (url is null || !Uri.TryCreate(url, UriKind.Absolute, out var uri) || (uri.Scheme != Uri.UriSchemeHttps && uri.Scheme != Uri.UriSchemeHttp)) + { problems.Add($"{name}: url must be an absolute http(s) URL"); continue; } + if (!seen.Add(name)) { problems.Add($"{name}: duplicate name"); continue; } + agents.Add(new RemoteAgentEntry(name, url, Str(item, "key"), Str(item, "description"))); + } + return new(agents, problems, null); + } + + private static string? Str(JsonElement e, string property) => + e.TryGetProperty(property, out var v) && v.ValueKind == JsonValueKind.String && v.GetString() is { Length: > 0 } s ? s : null; +} +``` + +- [ ] **Step 4: Run tests** — Expected: 4 passed. + +- [ ] **Step 5: Commit** + +```bash +git add extensions/Volcanic.AgenticCore.Extensions.A2A/AgentsFile.cs tests/Volcanic.AgenticCore.Extensions.Tests/A2A/AgentsFileTests.cs +git commit -m "feat(a2a-client): read remote agents from agents.json on every call" +``` + +--- + +### Task 3: Results — state names, text, artifacts + +**Files:** +- Create: `extensions/Volcanic.AgenticCore.Extensions.A2A/A2AResults.cs` +- Test: `tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AResultsTests.cs` + +- [ ] **Step 1: Failing tests** + +```csharp +using System.Text; +using A2A; +using Volcanic.AgenticCore.Extensions.A2A; +using A2AMessage = global::A2A.Message; +using A2ATaskStatus = global::A2A.TaskStatus; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A; + +public sealed class A2AResultsTests : IDisposable +{ + private readonly string _ws = Path.Combine(Path.GetTempPath(), "a2a-results", Guid.NewGuid().ToString("N")); + + private static AgentTask Task_(TaskState state, string? statusText = null, params Artifact[] artifacts) => new() + { + Id = "task-123456789", ContextId = "ctx-1", + Status = new A2ATaskStatus + { + State = state, + Message = statusText is null ? null : new A2AMessage { Role = Role.Agent, MessageId = "m", Parts = [Part.FromText(statusText)] }, + }, + Artifacts = [.. artifacts], + }; + + [Fact] + public void FinalText_PrefersResponseArtifact() + { + var task = Task_(TaskState.Completed, null, + new Artifact { ArtifactId = "other", Parts = [Part.FromText("side")] }, + new Artifact { ArtifactId = "response", Parts = [Part.FromText("Booked "), Part.FromText("Friday")] }); + Assert.Equal("Booked Friday", A2AResults.FinalText(task)); + } + + [Fact] + public void FinalText_FallsBackToTextArtifactsThenStatus() + { + Assert.Equal("side", A2AResults.FinalText(Task_(TaskState.Completed, null, new Artifact { ArtifactId = "x", Parts = [Part.FromText("side")] }))); + Assert.Equal("why", A2AResults.FinalText(Task_(TaskState.Failed, "why"))); + } + + [Fact] + public void Send_FileArtifact_SavedToInbox() + { + var task = Task_(TaskState.Completed, null, new Artifact + { + ArtifactId = "f1", Name = "plan.pdf", Parts = [Part.FromRaw(Encoding.UTF8.GetBytes("pdf!"), "application/pdf", "../../plan.pdf")], + }); + var files = A2AResults.SaveFiles(_ws, "travel-bot", task, 1024); + var file = Assert.Single(files); + Assert.Equal("inbox/a2a/travel-bot/plan.pdf", file); + Assert.Equal("pdf!", File.ReadAllText(Path.Combine(_ws, file))); + } + + [Fact] + public void Describe_InputRequired_HasQuestionAndHint() + { + var text = A2AResults.Describe("travel-bot", Task_(TaskState.InputRequired, "Which date?"), []); + Assert.Contains("input-required", text); + Assert.Contains("Which date?", text); + Assert.Contains("action=send", text); + Assert.Contains("task-123456789", text); + Assert.Contains("ctx-1", text); + } + + [Fact] + public void DeliveryText_HasHeader() + { + var text = A2AResults.DeliveryText("travel-bot", Task_(TaskState.Completed, null, + new Artifact { ArtifactId = "response", Parts = [Part.FromText("Done")] }), ["inbox/a2a/travel-bot/x.pdf"]); + Assert.StartsWith("[A2A result from \"travel-bot\", task task-123456789 (completed)]", text); + Assert.Contains("Done", text); + Assert.Contains("Files: inbox/a2a/travel-bot/x.pdf", text); + } + + public void Dispose() { try { Directory.Delete(_ws, true); } catch (IOException) { } } +} +``` + +- [ ] **Step 2: Run to see it fail** — build error. + +- [ ] **Step 3: Implement** + +```csharp +using System.Text; +using A2A; +using A2AMessage = global::A2A.Message; + +namespace Volcanic.AgenticCore.Extensions.A2A; + +/// How remote tasks read for the model: state names, the answer text, saved files, result and delivery texts. +internal static class A2AResults +{ + public const string ResponseArtifactId = "response"; + + /// Terminal, or waiting for us (input/auth required): nothing more happens without us. + public static bool IsSettled(TaskState state) => + state.IsTerminal() || state is TaskState.InputRequired or TaskState.AuthRequired; + + public static string StateName(TaskState state) => state switch + { + TaskState.Submitted => "submitted", + TaskState.Working => "working", + TaskState.Completed => "completed", + TaskState.Failed => "failed", + TaskState.Canceled => "canceled", + TaskState.InputRequired => "input-required", + TaskState.Rejected => "rejected", + TaskState.AuthRequired => "auth-required", + _ => "unknown", + }; + + public static string TextOf(IEnumerable? parts) => + string.Concat((parts ?? []).Where(p => p.ContentCase == PartContentCase.Text).Select(p => p.Text)); + + public static string TextOf(A2AMessage? message) => TextOf(message?.Parts); + + /// The answer: artifact response, else every text artifact part, else the status message. + public static string? FinalText(AgentTask task) + { + var artifacts = task.Artifacts ?? []; + var response = artifacts.Where(a => a.ArtifactId == ResponseArtifactId).Select(a => TextOf(a.Parts)).FirstOrDefault(t => t.Length > 0); + if (response is not null) return response; + var text = string.Join("\n", artifacts.Select(a => TextOf(a.Parts)).Where(t => t.Length > 0)); + if (text.Length > 0) return text; + var status = TextOf(task.Status.Message); + return status.Length > 0 ? status : null; + } + + /// Saves file parts (inline bytes) under inbox/a2a/<agent>/; returns workspace-relative paths. + public static IReadOnlyList SaveFiles(string workspace, string agent, AgentTask task, long maxBytes) + { + var saved = new List(); + foreach (var artifact in task.Artifacts ?? []) + foreach (var part in artifact.Parts ?? []) + { + if (part.ContentCase != PartContentCase.Raw || part.Raw is not { } bytes || bytes.Length > maxBytes) continue; + var name = Sanitize(part.Filename ?? artifact.Name ?? artifact.ArtifactId); + var relative = $"inbox/a2a/{agent}/{name}"; + var full = Path.Combine(workspace, relative); + for (var n = 2; File.Exists(full); n++) + { + relative = $"inbox/a2a/{agent}/{Path.GetFileNameWithoutExtension(name)}-{n}{Path.GetExtension(name)}"; + full = Path.Combine(workspace, relative); + } + Directory.CreateDirectory(Path.GetDirectoryName(full)!); + File.WriteAllBytes(full, bytes); + saved.Add(relative); + } + return saved; + } + + /// Links the remote agent returned instead of bytes (not downloaded). + public static IReadOnlyList Links(AgentTask task) => + (task.Artifacts ?? []).SelectMany(a => a.Parts ?? []).Where(p => p.ContentCase == PartContentCase.Url && p.Url is not null) + .Select(p => p.Url!).ToList(); + + /// The tool result for a task. + public static string Describe(string agent, AgentTask task, IReadOnlyList files) + { + var sb = new StringBuilder(); + sb.Append($"Agent \"{agent}\": task {task.Id} (context {task.ContextId}) is {StateName(task.Status.State)}."); + AppendBody(sb, task, files); + return sb.ToString(); + } + + /// The follow-up prompt a watch delivers. + public static string DeliveryText(string agent, AgentTask task, IReadOnlyList files) + { + var kind = task.Status.State is TaskState.InputRequired or TaskState.AuthRequired ? "question" : "result"; + var sb = new StringBuilder($"[A2A {kind} from \"{agent}\", task {task.Id} ({StateName(task.Status.State)})]"); + sb.Append($"\nContext: {task.ContextId}"); + AppendBody(sb, task, files); + return sb.ToString(); + } + + private static void AppendBody(StringBuilder sb, AgentTask task, IReadOnlyList files) + { + if (FinalText(task) is { } text) sb.Append('\n').Append(text); + if (files.Count > 0) sb.Append("\nFiles: ").Append(string.Join(", ", files)); + var links = Links(task); + if (links.Count > 0) sb.Append("\nLinks: ").Append(string.Join(", ", links)); + if (task.Status.State is TaskState.InputRequired or TaskState.AuthRequired) + sb.Append($"\nThe agent is waiting for an answer: reply with action=send, agent, taskId={task.Id} and your answer."); + } + + private static string Sanitize(string? name) + { + var file = Path.GetFileName(name ?? ""); + if (string.IsNullOrWhiteSpace(file)) return "file"; + var invalid = Path.GetInvalidFileNameChars(); + var clean = new string(file.Select(c => invalid.Contains(c) || char.IsWhiteSpace(c) ? '_' : c).ToArray()); + return clean.Length > 80 ? clean[^80..] : clean; + } +} +``` + +- [ ] **Step 4: Run tests** — Expected: 5 passed. + +- [ ] **Step 5: Commit** + +```bash +git add extensions/Volcanic.AgenticCore.Extensions.A2A/A2AResults.cs tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AResultsTests.cs +git commit -m "feat(a2a-client): describe remote tasks and save their files to the inbox" +``` + +--- + +### Task 4: pending.json + +**Files:** +- Create: `extensions/Volcanic.AgenticCore.Extensions.A2A/PendingFile.cs` +- Test: `tests/Volcanic.AgenticCore.Extensions.Tests/A2A/PendingFileTests.cs` + +- [ ] **Step 1: Failing tests** + +```csharp +using Volcanic.AgenticCore.Extensions.A2A; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A; + +public sealed class PendingFileTests : IDisposable +{ + private readonly string _ws = Path.Combine(Path.GetTempPath(), "a2a-pending", Guid.NewGuid().ToString("N")); + + private static WatchRecord Record(string id, string target = "1") => + new(id, target, "travel-bot", "task-" + id, "ctx", null, DateTimeOffset.UnixEpoch, WatchSources.Poll); + + [Fact] + public void AddUpdateRemove_RoundTrips() + { + PendingFile.Upsert(_ws, Record("a")); + PendingFile.Upsert(_ws, Record("b")); + PendingFile.Upsert(_ws, Record("a") with { Undelivered = "text" }); + var all = PendingFile.Read(_ws); + Assert.Equal(2, all.Count); + Assert.Equal("text", all.Single(r => r.WatchId == "a").Undelivered); + PendingFile.Remove(_ws, "a"); + Assert.Equal("b", Assert.Single(PendingFile.Read(_ws)).WatchId); + } + + [Fact] + public void Read_MissingOrMalformed_Empty() + { + Assert.Empty(PendingFile.Read(_ws)); + Directory.CreateDirectory(Path.GetDirectoryName(PendingFile.PathFor(_ws))!); + File.WriteAllText(PendingFile.PathFor(_ws), "{ broken"); + Assert.Empty(PendingFile.Read(_ws)); + } + + public void Dispose() { try { Directory.Delete(_ws, true); } catch (IOException) { } } +} +``` + +- [ ] **Step 2: Run to see it fail** — build error. + +- [ ] **Step 3: Implement** + +```csharp +using System.Collections.Concurrent; +using System.Text.Json; + +namespace Volcanic.AgenticCore.Extensions.A2A; + +public static class WatchSources +{ + public const string Push = "push"; + public const string Subscribe = "subscribe"; + public const string Poll = "poll"; +} + +/// The host's delivery target (Telegram: the chat id). +/// Our secret for this watch's push webhook; null when not using push. +/// A finished result that couldn't be delivered yet (sent on the target's next session start). +public sealed record WatchRecord( + string WatchId, string Target, string Agent, string TaskId, string ContextId, string? PushToken, + DateTimeOffset StartedAt, string Source, string? Undelivered = null); + +/// <workspace>/.volcanic/a2a/pending.json: the watches of a workspace, so they survive restarts. +public static class PendingFile +{ + private static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.Web) { WriteIndented = true }; + private static readonly ConcurrentDictionary Locks = new(StringComparer.Ordinal); + + public static string PathFor(string workspace) => Path.Combine(workspace, ".volcanic", "a2a", "pending.json"); + + public static IReadOnlyList Read(string workspace) + { + lock (LockFor(workspace)) return ReadUnlocked(workspace); + } + + public static void Upsert(string workspace, WatchRecord record) => Change(workspace, list => + { + list.RemoveAll(r => r.WatchId == record.WatchId); + list.Add(record); + }); + + public static void Remove(string workspace, string watchId) => Change(workspace, list => list.RemoveAll(r => r.WatchId == watchId)); + + private static void Change(string workspace, Action> change) + { + lock (LockFor(workspace)) + { + var list = ReadUnlocked(workspace).ToList(); + change(list); + var path = PathFor(workspace); + Directory.CreateDirectory(Path.GetDirectoryName(path)!); + var temp = path + "." + Guid.NewGuid().ToString("N") + ".tmp"; + File.WriteAllText(temp, JsonSerializer.Serialize(list, Json)); + File.Move(temp, path, overwrite: true); + } + } + + private static List ReadUnlocked(string workspace) + { + var path = PathFor(workspace); + try + { + return File.Exists(path) ? JsonSerializer.Deserialize>(File.ReadAllText(path), Json) ?? [] : []; + } + catch (Exception ex) when (ex is JsonException or IOException or UnauthorizedAccessException) + { + return []; + } + } + + private static object LockFor(string workspace) => Locks.GetOrAdd(Path.GetFullPath(workspace), _ => new object()); +} +``` + +- [ ] **Step 4: Run tests** — Expected: 2 passed. + +- [ ] **Step 5: Commit** + +```bash +git add extensions/Volcanic.AgenticCore.Extensions.A2A/PendingFile.cs tests/Volcanic.AgenticCore.Extensions.Tests/A2A/PendingFileTests.cs +git commit -m "feat(a2a-client): persist watches in pending.json" +``` + +--- + +### Task 5: Connecting to remote agents (+ the in-process remote for tests) + +**Files:** +- Create: `extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs` +- Create: `tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/RemoteAgentServer.cs` +- Test: `tests/Volcanic.AgenticCore.Extensions.Tests/A2A/RemoteAgentsTests.cs` + +- [ ] **Step 1: The remote server harness** + +`tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/RemoteAgentServer.cs` — our own A2A server, in process, as "the remote agent": + +```csharp +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting; +using Microsoft.AspNetCore.TestHost; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Volcanic.AgenticCore.A2A.Server; +using Volcanic.AgenticCore.A2A.Server.Callers; +using Volcanic.AgenticCore.A2A.Server.Hosting; +using Volcanic.AgenticCore.A2A.Server.Push; +using Volcanic.AgenticCore.Features.Agent; +using Volcanic.AgenticCore.Features.Extensions; +using Volcanic.AgenticCore.Hosting; +using Volcanic.AgenticCore.Testing; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A.Support; + +/// +/// A remote A2A agent: our server library on a TestServer over one workspace, with its own FauxProvider as the remote +/// agent's brain. reaches it from an HttpClient; pushes go to . +/// +public sealed class RemoteAgentServer : IAsyncDisposable +{ + public const string Key = "remote-key-0123456789"; + public const string BaseUrl = "http://remote.test/"; + + private readonly string _root = Path.Combine(Path.GetTempPath(), "a2a-remote", Guid.NewGuid().ToString("N")); + + public FauxProvider Faux { get; } = new(); + public string Workspace => Path.Combine(_root, "ws"); + public WebApplication App { get; private set; } = null!; + public HttpMessageHandler Handler => App.GetTestServer().CreateHandler(); + + /// Where the remote's push sender posts. Default: answers 200 to everything. + public HttpMessageHandler PushHandler { get; set; } = new OkHandler(); + + public static async Task StartAsync(Action? configure = null) + { + var host = new RemoteAgentServer(); + Directory.CreateDirectory(Path.GetDirectoryName(CallersFile.PathFor(host.Workspace))!); + File.WriteAllText(CallersFile.PathFor(host.Workspace), $$"""[ { "name": "client", "key": "{{Key}}" } ]"""); + var builder = WebApplication.CreateBuilder(); + builder.WebHost.UseTestServer(); + builder.Services.AddAgenticCore(o => + { + o.AgentDir = Path.Combine(host._root, "agent"); + o.DiscoverExtensionAssemblies = false; + o.DefaultModel = FauxProvider.ModelKey; + o.Extensions.Add(ExtensionRegistration.FromInstance(host.Faux.Extension)); + o.ProcessEnvironment.HardenHostProcess = false; + o.Retry = new RetrySettings { Enabled = false }; + }); + builder.Services.AddSingleton(_ => new HttpA2APushSender( + new HttpClient(new DeferredHandler(() => host.PushHandler)), TimeProvider.System, NullLogger.Instance)); + builder.Services.AddSingleton(new WorkspaceCallerDirectory( + () => [new A2AWorkspace(host.Workspace, "remote", "remote")], NullLogger.Instance)); + builder.Services.AddA2AServer(o => + { + o.AllowPrivatePushUrls = true; + o.PublicUrl = BaseUrl.TrimEnd('/'); + configure?.Invoke(o); + }); + host.App = builder.Build(); + host.App.MapA2AServer(); + await host.App.StartAsync(); + return host; + } + + public async ValueTask DisposeAsync() + { + await App.StopAsync(); + await App.DisposeAsync(); + try { Directory.Delete(_root, true); } catch (IOException) { } catch (UnauthorizedAccessException) { } + } + + private sealed class OkHandler : HttpMessageHandler + { + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) => + Task.FromResult(new HttpResponseMessage(System.Net.HttpStatusCode.OK)); + } + + /// Resolves the target handler per request, so tests can swap after start. + private sealed class DeferredHandler(Func target) : DelegatingHandler + { + protected override Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) => + new HttpMessageInvoker(target(), disposeHandler: false).SendAsync(request, cancellationToken); + } +} +``` + +`tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/PushToRegistryHandler.cs` (create it in **Task 6**, it needs the registry): + +```csharp +using System.Net; +using System.Text.Json; +using Volcanic.AgenticCore.Extensions.A2A; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A.Support; + +/// Stands in for our push receiver endpoint: POST …/a2a/push/{watchId} → . +public sealed class PushToRegistryHandler(A2AWatchRegistry? registry) : HttpMessageHandler +{ + public int Received; + + protected override async Task SendAsync(HttpRequestMessage request, CancellationToken cancellationToken) + { + Interlocked.Increment(ref Received); + if (registry is null) return new HttpResponseMessage(HttpStatusCode.OK); + var watchId = request.RequestUri!.Segments[^1]; + var token = request.Headers.TryGetValues("X-A2A-Notification-Token", out var v) ? v.Single() : null; + using var doc = JsonDocument.Parse(await request.Content!.ReadAsStringAsync(cancellationToken)); + var ok = await registry.OnPushAsync(watchId, token, doc.RootElement.Clone(), cancellationToken); + return new HttpResponseMessage(ok ? HttpStatusCode.OK : HttpStatusCode.NotFound); + } +} +``` + +(Only `RemoteAgentServer.cs` is created in this task.) + +- [ ] **Step 2: Failing tests** + +`tests/Volcanic.AgenticCore.Extensions.Tests/A2A/RemoteAgentsTests.cs`: + +```csharp +using Volcanic.AgenticCore.Extensions.A2A; +using Volcanic.AgenticCore.Extensions.Tests.A2A.Support; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A; + +public class RemoteAgentsTests +{ + [Fact] + public async Task Connect_FetchesExtendedCardWithKey() + { + 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" } ] }"""); + var connector = new RemoteAgentConnector(remote.Handler, TimeSpan.FromSeconds(10)); + using var agent = await connector.ConnectAsync(new RemoteAgentEntry("travel", RemoteAgentServer.BaseUrl, RemoteAgentServer.Key, null), CancellationToken.None); + Assert.Equal("Remote travel", agent.Card.Name); + Assert.True(agent.Streaming); + Assert.True(agent.SupportsPush); + } + + [Fact] + public async Task Connect_WithoutKey_PublicCardOnly() + { + await using var remote = await RemoteAgentServer.StartAsync(); + var connector = new RemoteAgentConnector(remote.Handler, TimeSpan.FromSeconds(10)); + using var agent = await connector.ConnectAsync(new RemoteAgentEntry("travel", RemoteAgentServer.BaseUrl, null, null), CancellationToken.None); + Assert.Equal("Volcanic assistant", agent.Card.Name); + } + + [Fact] + public void CardUrl_BaseOrExplicitJson() + { + Assert.Equal("https://a.example/x/.well-known/agent-card.json", RemoteAgentConnector.CardUrl("https://a.example/x").ToString()); + Assert.Equal("https://a.example/card.json", RemoteAgentConnector.CardUrl("https://a.example/card.json").ToString()); + } +} +``` + +- [ ] **Step 3: Run to see it fail** — build error. + +- [ ] **Step 4: Implement** + +`extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs`: + +```csharp +using A2A; + +namespace Volcanic.AgenticCore.Extensions.A2A; + +/// 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 +{ + public RemoteAgentEntry Entry => entry; + public AgentCard Card => card; + public IA2AClient Client => client; + public bool Streaming => card.Capabilities?.Streaming == true; + public bool SupportsPush => card.Capabilities?.PushNotifications == true; + + public void Dispose() + { + (client as IDisposable)?.Dispose(); + http.Dispose(); + } +} + +/// Resolves an agent's card (public, then extended when we have a key) and builds a client over one shared handler. +internal sealed class RemoteAgentConnector(HttpMessageHandler handler, TimeSpan cardTimeout) +{ + public static Uri CardUrl(string url) => + url.EndsWith(".json", StringComparison.OrdinalIgnoreCase) + ? new Uri(url) + : new Uri(new Uri(url.TrimEnd('/') + "/"), ".well-known/agent-card.json"); + + public async Task ConnectAsync(RemoteAgentEntry entry, CancellationToken cancellationToken) + { + var cardUrl = CardUrl(entry.Url); + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + timeout.CancelAfter(cardTimeout); + AgentCard card; + using (var publicHttp = new HttpClient(handler, disposeHandler: false)) + card = await new A2ACardResolver(new Uri(cardUrl.GetLeftPart(UriPartial.Authority)), publicHttp, cardUrl.PathAndQuery) + .GetAgentCardAsync(timeout.Token).ConfigureAwait(false); + + var http = new HttpClient(handler, disposeHandler: false) { Timeout = Timeout.InfiniteTimeSpan }; + if (entry.Key is { Length: > 0 } key) + { + if (UsesBearer(card)) http.DefaultRequestHeaders.Authorization = new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", key); + else http.DefaultRequestHeaders.Add(ApiKeyHeader(card), key); + } + var client = A2AClientFactory.Create(card, http); + if (entry.Key is not null && card.Capabilities?.ExtendedAgentCard == true) + { + try + { + var extended = await client.GetExtendedAgentCardAsync(new GetExtendedAgentCardRequest(), timeout.Token).ConfigureAwait(false); + (client as IDisposable)?.Dispose(); + card = extended; + client = A2AClientFactory.Create(card, http); + } + catch (Exception ex) when (ex is A2AException or HttpRequestException) { } // keep the public card + } + return new RemoteAgent(entry, card, client, http); + } + + private static bool UsesBearer(AgentCard card) => + card.SecuritySchemes is { Count: > 0 } schemes && + schemes.Values.All(s => s.HttpAuthSecurityScheme is { Scheme: var scheme } && string.Equals(scheme, "bearer", StringComparison.OrdinalIgnoreCase)); + + private static string ApiKeyHeader(AgentCard card) => + card.SecuritySchemes?.Values.Select(s => s.ApiKeySecurityScheme) + .FirstOrDefault(k => k is not null && string.Equals(k.Location, "header", StringComparison.OrdinalIgnoreCase))?.Name + ?? "X-API-Key"; +} +``` + +(`A2ACardResolver` with an authority-only base and a path: check its URL combination in the decompiled SDK if a test fails; the goal is `cardUrl` exactly.) + +- [ ] **Step 5: Run tests** — Expected: 3 passed. + +- [ ] **Step 6: Commit** + +```bash +git add extensions/Volcanic.AgenticCore.Extensions.A2A/RemoteAgents.cs tests/Volcanic.AgenticCore.Extensions.Tests/A2A +git commit -m "feat(a2a-client): connect to remote agents through their cards with the configured key" +``` + +--- + +### Task 6: Delivery and the watch registry + +**Files:** +- Create: `extensions/Volcanic.AgenticCore.Extensions.A2A/A2AResultDelivery.cs` +- Create: `extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs` +- Create: `tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/PushToRegistryHandler.cs` (code in Task 5 Step 1) +- Test: `tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs` + +- [ ] **Step 1: Delivery** + +```csharp +using System.Collections.Concurrent; +using Volcanic.AgenticCore.Abstractions.Extensions; + +namespace Volcanic.AgenticCore.Extensions.A2A; + +/// Where finished remote tasks go. Targets are host ids (Telegram: chat ids). +public interface IA2AResultDelivery +{ + /// Run as a follow-up prompt in the target's conversation. False: not deliverable now. + Task DeliverAsync(string target, string text, CancellationToken cancellationToken); + + /// The target's workspace. A path computation: must not create or wake a session. + string? WorkspaceFor(string target); + + /// Targets whose workspace may hold a pending.json (restore at startup). + IEnumerable KnownTargets(); +} + +/// Default delivery: the target's live session, if any (registered on session start, removed on shutdown). +public sealed class LiveSessionDelivery : IA2AResultDelivery +{ + private readonly ConcurrentDictionary _live = new(StringComparer.Ordinal); + + public void Attach(string target, string workspace, IExtensionApi api) => _live[target] = (api, workspace); + + public void Detach(string target, IExtensionApi api) + { + if (_live.TryGetValue(target, out var entry) && ReferenceEquals(entry.Api, api)) _live.TryRemove(target, out _); + } + + public Task DeliverAsync(string target, string text, CancellationToken cancellationToken) + { + if (!_live.TryGetValue(target, out var entry)) return Task.FromResult(false); + try + { + entry.Api.SendUserMessage(text, null, MessageDelivery.FollowUp); + return Task.FromResult(true); + } + catch (Exception ex) when (ex is InvalidOperationException or ObjectDisposedException) + { + return Task.FromResult(false); + } + } + + public string? WorkspaceFor(string target) => _live.TryGetValue(target, out var entry) ? entry.Workspace : null; + public IEnumerable KnownTargets() => _live.Keys.ToList(); +} +``` + +- [ ] **Step 2: Failing tests** (the registry is exercised against the in-process remote) + +`tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs`: + +```csharp +using System.Collections.Concurrent; +using System.Text.Json; +using A2A; +using Volcanic.AgenticCore.Extensions.A2A; +using Volcanic.AgenticCore.Extensions.Tests.A2A.Support; +using Volcanic.AgenticCore.Testing; +using A2AMessage = global::A2A.Message; + +namespace Volcanic.AgenticCore.Extensions.Tests.A2A; + +public sealed class A2AWatchTests : IDisposable +{ + private readonly string _ws = Path.Combine(Path.GetTempPath(), "a2a-watch", Guid.NewGuid().ToString("N")); + + private sealed class RecordingDelivery(string workspace) : IA2AResultDelivery + { + public ConcurrentQueue<(string Target, string Text)> Delivered { get; } = new(); + public bool Accept { get; set; } = true; + public Task DeliverAsync(string target, string text, CancellationToken cancellationToken) + { + if (Accept) Delivered.Enqueue((target, text)); + return Task.FromResult(Accept); + } + public string? WorkspaceFor(string target) => workspace; + public IEnumerable KnownTargets() => ["1"]; + } + + private void WriteAgents(string key = RemoteAgentServer.Key) + { + Directory.CreateDirectory(Path.GetDirectoryName(AgentsFile.PathFor(_ws))!); + File.WriteAllText(AgentsFile.PathFor(_ws), $$"""[ { "name": "remote", "url": "{{RemoteAgentServer.BaseUrl}}", "key": "{{key}}" } ]"""); + } + + private static async Task StartRemoteTaskAsync(RemoteAgent agent, string text = "go") => + (await agent.Client.SendMessageAsync(new SendMessageRequest + { + Message = new A2AMessage { Role = Role.User, MessageId = Guid.NewGuid().ToString("N"), Parts = [Part.FromText(text)] }, + Configuration = new SendMessageConfiguration { ReturnImmediately = true }, + })).Task!; + + private static async Task WaitUntilAsync(Func condition, int timeoutMs = 10_000) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException("condition not met"); + await Task.Delay(20); + } + } + + [Theory] + [InlineData(WatchSources.Subscribe)] + [InlineData(WatchSources.Poll)] + public async Task Watch_DeliversResult(string source) + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Trip booked").WithDelay(TimeSpan.FromMilliseconds(500))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = source, PollDelays = [TimeSpan.FromMilliseconds(100)] }; + registry.UseDelivery(delivery); + var settings = new A2AClientSettings(); + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + var task = await StartRemoteTaskAsync(agent); + var record = await registry.StartAsync("1", _ws, agent, task, settings, CancellationToken.None); + Assert.Equal(source, record.Source); + Assert.Single(PendingFile.Read(_ws)); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); + var (target, text) = Assert.Single(delivery.Delivered); + Assert.Equal("1", target); + Assert.StartsWith($"[A2A result from \"remote\", task {task.Id} (completed)]", text); + Assert.Contains("Trip booked", text); + await WaitUntilAsync(() => PendingFile.Read(_ws).Count == 0); + } + + [Fact] + public async Task Watch_Push_DeliversFollowUp() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Pushed result").WithDelay(TimeSpan.FromMilliseconds(500))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SafetyPollInterval = TimeSpan.FromMinutes(10) }; + registry.UseDelivery(delivery); + var push = new PushToRegistryHandler(registry); + remote.PushHandler = push; + var settings = new A2AClientSettings { PushBaseUrl = "http://local.test" }; + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + var task = await StartRemoteTaskAsync(agent); + var record = await registry.StartAsync("1", _ws, agent, task, settings, CancellationToken.None); + Assert.Equal(WatchSources.Push, record.Source); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); + Assert.True(push.Received > 0); + Assert.Contains("Pushed result", Assert.Single(delivery.Delivered).Text); + } + + [Fact] + public async Task Watch_PushBadToken_Rejected() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("slow").WithDelay(TimeSpan.FromSeconds(5))); + WriteAgents(); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler); + registry.UseDelivery(new RecordingDelivery(_ws)); + var settings = new A2AClientSettings { PushBaseUrl = "http://local.test" }; + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + var record = await registry.StartAsync("1", _ws, agent, await StartRemoteTaskAsync(agent), settings, CancellationToken.None); + var body = JsonDocument.Parse("{}").RootElement; + Assert.False(await registry.OnPushAsync(record.WatchId, "wrong", body, CancellationToken.None)); + Assert.False(await registry.OnPushAsync("unknown", record.PushToken, body, CancellationToken.None)); + Assert.True(await registry.OnPushAsync(record.WatchId, record.PushToken, body, CancellationToken.None)); + } + + [Fact] + public async Task Watch_Timeout_DeliversNotice() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("never").WithDelay(TimeSpan.FromSeconds(10))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = WatchSources.Poll, PollDelays = [TimeSpan.FromMilliseconds(100)] }; + registry.UseDelivery(delivery); + var settings = new A2AClientSettings { WatchTimeout = TimeSpan.FromMilliseconds(400) }; + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + await registry.StartAsync("1", _ws, agent, await StartRemoteTaskAsync(agent), settings, CancellationToken.None); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); + Assert.Contains("gave no result", Assert.Single(delivery.Delivered).Text); + } + + [Fact] + public async Task Watch_AgentRemoved_EndsWatch() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("never").WithDelay(TimeSpan.FromSeconds(10))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = WatchSources.Poll, PollDelays = [TimeSpan.FromMilliseconds(100)] }; + registry.UseDelivery(delivery); + var settings = new A2AClientSettings(); + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + await registry.StartAsync("1", _ws, agent, await StartRemoteTaskAsync(agent), settings, CancellationToken.None); + File.WriteAllText(AgentsFile.PathFor(_ws), "[]"); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); + Assert.Contains("no longer in agents.json", Assert.Single(delivery.Delivered).Text); + Assert.Empty(registry.List("1")); + } + + [Fact] + public async Task Watch_Undelivered_KeptThenFlushed() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Later")); + WriteAgents(); + var delivery = new RecordingDelivery(_ws) { Accept = false }; + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { SourceOverride = WatchSources.Poll, PollDelays = [TimeSpan.FromMilliseconds(100)] }; + registry.UseDelivery(delivery); + var settings = new A2AClientSettings(); + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + await registry.StartAsync("1", _ws, agent, await StartRemoteTaskAsync(agent), settings, CancellationToken.None); + await WaitUntilAsync(() => PendingFile.Read(_ws).Any(r => r.Undelivered is not null)); + delivery.Accept = true; + await registry.FlushUndeliveredAsync("1", _ws, CancellationToken.None); + Assert.Contains("Later", Assert.Single(delivery.Delivered).Text); + Assert.Empty(PendingFile.Read(_ws)); + } + + [Fact] + public async Task Watch_Restore_FromPendingFile() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Done before restart")); + WriteAgents(); + var settings = new A2AClientSettings(); + AgentTask task; + await using (var first = new A2AWatchRegistry(TimeProvider.System, remote.Handler)) + { + using var agent = await first.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + task = await StartRemoteTaskAsync(agent); + } + PendingFile.Upsert(_ws, new WatchRecord("w1", "1", "remote", task.Id, task.ContextId, null, DateTimeOffset.UtcNow, WatchSources.Poll)); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) { PollDelays = [TimeSpan.FromMilliseconds(100)] }; + registry.UseDelivery(delivery); + Assert.Equal(1, await registry.RestoreAsync(settings, CancellationToken.None)); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); + Assert.Contains("Done before restart", Assert.Single(delivery.Delivered).Text); + } + + [Fact] + public void PollDelays_BackOffToFiveMinutes() + { + var delays = A2AWatchRegistry.DefaultPollDelays; + Assert.Equal(TimeSpan.FromSeconds(15), delays[0]); + Assert.Equal(TimeSpan.FromMinutes(5), delays[^1]); + Assert.Equal(TimeSpan.FromMinutes(5), A2AWatchRegistry.DelayAt(delays, 99)); + } + + public void Dispose() { try { Directory.Delete(_ws, true); } catch (IOException) { } } +} +``` + +- [ ] **Step 3: Run to see it fail** — build error. + +- [ ] **Step 4: Implement the registry** + +`extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs`: + +```csharp +using System.Collections.Concurrent; +using System.Security.Cryptography; +using System.Text; +using System.Text.Json; +using System.Threading.Channels; +using A2A; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; + +namespace Volcanic.AgenticCore.Extensions.A2A; + +/// +/// Process-wide (): one HTTP handler for every remote call, and the watches of long remote tasks. +/// A watch follows its task by push (when the card supports it and the host gave us a push base URL), else by +/// subscribing (streaming agents), else by polling; it delivers the outcome through and +/// is persisted in the workspace's pending.json until delivered. +/// +public sealed class A2AWatchRegistry : IAsyncDisposable +{ + public static A2AWatchRegistry Shared { get; } = new(TimeProvider.System, + new SocketsHttpHandler { PooledConnectionLifetime = TimeSpan.FromMinutes(5) }); + + public static readonly TimeSpan[] DefaultPollDelays = + [TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(30), TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(2), TimeSpan.FromMinutes(5)]; + + private static readonly TimeSpan[] ReconnectDelays = [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(15)]; + + private readonly HttpMessageHandler _handler; + private readonly ConcurrentDictionary _watches = new(StringComparer.Ordinal); + private IA2AResultDelivery _delivery; + + public A2AWatchRegistry(TimeProvider time, HttpMessageHandler handler, ILogger? logger = null) + { + Time = time; + _handler = handler; + Logger = logger ?? NullLogger.Instance; + _delivery = Live; + } + + public TimeProvider Time { get; } + public ILogger Logger { get; set; } + public LiveSessionDelivery Live { get; } = new(); + + internal string? SourceOverride { get; init; } + internal TimeSpan[] PollDelays { get; init; } = DefaultPollDelays; + internal TimeSpan SafetyPollInterval { get; init; } = TimeSpan.FromMinutes(10); + + public void UseDelivery(IA2AResultDelivery delivery) => _delivery = delivery; + + internal RemoteAgentConnector Connector(A2AClientSettings settings) => + new(_handler, TimeSpan.FromSeconds(settings.CardTimeoutSeconds)); + + internal static TimeSpan DelayAt(TimeSpan[] delays, int attempt) => delays[Math.Min(attempt, delays.Length - 1)]; + + public IReadOnlyList 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. + internal async Task StartAsync(string target, string workspace, RemoteAgent agent, AgentTask task, + A2AClientSettings settings, CancellationToken cancellationToken) + { + var watchId = Guid.NewGuid().ToString("N"); + string? token = null; + var source = SourceOverride ?? (agent.Streaming ? WatchSources.Subscribe : WatchSources.Poll); + if (SourceOverride is null && settings.PushBaseUrl is { Length: > 0 } pushBase && agent.SupportsPush) + { + token = Convert.ToHexString(RandomNumberGenerator.GetBytes(32)); + try + { + await agent.Client.CreateTaskPushNotificationConfigAsync(new CreateTaskPushNotificationConfigRequest + { + TaskId = task.Id, + ConfigId = watchId, + Config = new PushNotificationConfig { Id = watchId, Url = $"{pushBase.TrimEnd('/')}/a2a/push/{watchId}", Token = token }, + }, cancellationToken).ConfigureAwait(false); + source = WatchSources.Push; + } + catch (Exception ex) when (ex is A2AException or HttpRequestException) + { + Logger.LogInformation("A2A: {Agent} refused push for task {TaskId} ({Error}); watching by {Source}", agent.Entry.Name, task.Id, ex.Message, source); + token = null; + } + } + var record = new WatchRecord(watchId, target, agent.Entry.Name, task.Id, task.ContextId, token, Time.GetUtcNow(), source); + PendingFile.Upsert(workspace, record); + Arm(record, workspace, settings); + return record; + } + + /// Push receiver: a known watch with the right token wakes up and checks its task. False → 404. + public Task OnPushAsync(string watchId, string? token, JsonElement body, CancellationToken cancellationToken) + { + if (!_watches.TryGetValue(watchId, out var watch) || watch.Record.PushToken is not { } expected || token is null || + !CryptographicOperations.FixedTimeEquals(Encoding.UTF8.GetBytes(expected), Encoding.UTF8.GetBytes(token))) + return Task.FromResult(false); + watch.Wake.Writer.TryWrite(true); + return Task.FromResult(true); + } + + /// Re-arm every persisted watch of the delivery's known targets; delivers results kept undelivered. Returns the count re-armed. + public async Task RestoreAsync(A2AClientSettings settings, CancellationToken cancellationToken) + { + var count = 0; + foreach (var target in _delivery.KnownTargets().Distinct(StringComparer.Ordinal)) + { + if (_delivery.WorkspaceFor(target) is not { } workspace) continue; + await FlushUndeliveredAsync(target, workspace, cancellationToken).ConfigureAwait(false); + foreach (var record in PendingFile.Read(workspace).Where(r => r.Target == target && r.Undelivered is null)) + { + if (_watches.ContainsKey(record.WatchId)) continue; + Arm(record, workspace, settings); + count++; + } + } + return count; + } + + /// Deliver results that were kept because the target had no live conversation. + public async Task FlushUndeliveredAsync(string target, string workspace, CancellationToken cancellationToken) + { + foreach (var record in PendingFile.Read(workspace).Where(r => r.Target == target && r.Undelivered is not null)) + if (await TryDeliverAsync(target, record.Undelivered!, cancellationToken).ConfigureAwait(false)) + PendingFile.Remove(workspace, record.WatchId); + } + + /// Stop watching a task (after cancel); true when a watch existed. + public async Task StopWatchingAsync(string target, string workspace, string taskId) + { + var watch = _watches.Values.FirstOrDefault(w => w.Record.Target == target && w.Record.TaskId == taskId); + 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; + await watch.StopAsync().ConfigureAwait(false); + return true; + } + + private void Arm(WatchRecord record, string workspace, A2AClientSettings settings) + { + var watch = new Watch(record, workspace); + if (!_watches.TryAdd(record.WatchId, watch)) return; + watch.Loop = Task.Run(() => RunAsync(watch, settings)); + } + + private async Task RunAsync(Watch watch, A2AClientSettings settings) + { + var record = watch.Record; + var ct = watch.Stop.Token; + var deadline = record.StartedAt + settings.WatchTimeout; + var source = record.Source; + var failures = 0; + var polls = 0; + try + { + while (!ct.IsCancellationRequested) + { + if (Time.GetUtcNow() >= deadline) + { + await FinishAsync(watch, $"[A2A result from \"{record.Agent}\", task {record.TaskId} (no result)]\n" + + $"\"{record.Agent}\" gave no result within {settings.WatchTimeout}. Check it with action=status, agent={record.Agent}, taskId={record.TaskId}.", null).ConfigureAwait(false); + return; + } + if (AgentsFile.Find(watch.Workspace, record.Agent) is not { } entry) + { + await FinishAsync(watch, $"[A2A result from \"{record.Agent}\", task {record.TaskId} (stopped)]\n" + + $"\"{record.Agent}\" is no longer in agents.json, so its task is no longer watched.", null).ConfigureAwait(false); + return; + } + + AgentTask? task = null; + TimeSpan wait; + try + { + using var agent = await Connector(settings).ConnectAsync(entry, ct).ConfigureAwait(false); + task = source == WatchSources.Subscribe + ? await SubscribeAsync(agent, record.TaskId, deadline, ct).ConfigureAwait(false) + : await agent.Client.GetTaskAsync(new GetTaskRequest { Id = record.TaskId }, ct).ConfigureAwait(false); + if (task is not null && A2AResults.IsSettled(task.Status.State)) + { + var files = A2AResults.SaveFiles(watch.Workspace, record.Agent, task, settings.MaxFileBytes); + await FinishAsync(watch, A2AResults.DeliveryText(record.Agent, task, files), agent).ConfigureAwait(false); + return; + } + failures = 0; + wait = source switch + { + WatchSources.Push => SafetyPollInterval, + WatchSources.Subscribe => DelayAt(ReconnectDelays, 0), + _ => DelayAt(PollDelays, polls++), + }; + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + failures++; + Logger.LogInformation("A2A: watching task {TaskId} of {Agent} failed ({Error})", record.TaskId, record.Agent, ex.Message); + if (source == WatchSources.Subscribe && failures >= 3) source = WatchSources.Poll; + wait = source == WatchSources.Subscribe ? DelayAt(ReconnectDelays, failures) : DelayAt(PollDelays, polls++); + } + await WaitAsync(watch, Min(wait, deadline - Time.GetUtcNow()), ct).ConfigureAwait(false); + } + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + } + catch (Exception ex) + { + Logger.LogWarning(ex, "A2A: watch {WatchId} crashed", record.WatchId); + } + } + + private async Task SubscribeAsync(RemoteAgent agent, string taskId, DateTimeOffset deadline, CancellationToken ct) + { + AgentTask? current = null; + try + { + await foreach (var ev in agent.Client.SubscribeToTaskAsync(new SubscribeToTaskRequest { Id = taskId }, ct).ConfigureAwait(false)) + { + current = TaskProjection.Apply(current, ev); + if (current is not null && A2AResults.IsSettled(current.Status.State)) return current; + if (Time.GetUtcNow() >= deadline) break; + } + } + catch (A2AException) // e.g. the task is already terminal: read it instead + { + } + return await agent.Client.GetTaskAsync(new GetTaskRequest { Id = taskId }, ct).ConfigureAwait(false); + } + + private async Task WaitAsync(Watch watch, TimeSpan wait, CancellationToken ct) + { + if (wait <= TimeSpan.Zero) return; + using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); + var delay = Task.Delay(wait, Time, timeout.Token); + var woken = watch.Wake.Reader.WaitToReadAsync(timeout.Token).AsTask(); + await Task.WhenAny(delay, woken).ConfigureAwait(false); + await timeout.CancelAsync().ConfigureAwait(false); + while (watch.Wake.Reader.TryRead(out _)) { } + ct.ThrowIfCancellationRequested(); + } + + private async Task FinishAsync(Watch watch, string text, RemoteAgent? agent) + { + _watches.TryRemove(watch.Record.WatchId, out _); + if (agent is not null && watch.Record.PushToken is not null) + { + try + { + await agent.Client.DeleteTaskPushNotificationConfigAsync(new DeleteTaskPushNotificationConfigRequest + { TaskId = watch.Record.TaskId, Id = watch.Record.WatchId }, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) when (ex is A2AException or HttpRequestException) { } + } + if (await TryDeliverAsync(watch.Record.Target, text, CancellationToken.None).ConfigureAwait(false)) + PendingFile.Remove(watch.Workspace, watch.Record.WatchId); + else + PendingFile.Upsert(watch.Workspace, watch.Record with { Undelivered = text }); + } + + private async Task TryDeliverAsync(string target, string text, CancellationToken ct) + { + try + { + return await _delivery.DeliverAsync(target, text, ct).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + Logger.LogWarning(ex, "A2A: delivering a result to {Target} failed", target); + return false; + } + } + + private static TimeSpan Min(TimeSpan a, TimeSpan b) => a < b ? a : b; + + public async Task StopAsync() + { + foreach (var watch in _watches.Values) await watch.StopAsync().ConfigureAwait(false); + _watches.Clear(); + } + + public async ValueTask DisposeAsync() => await StopAsync().ConfigureAwait(false); + + private sealed class Watch(WatchRecord record, string workspace) + { + public WatchRecord Record => record; + public string Workspace => workspace; + public CancellationTokenSource Stop { get; } = new(); + public Channel Wake { get; } = Channel.CreateBounded(new BoundedChannelOptions(1) { FullMode = BoundedChannelFullMode.DropOldest }); + public Task Loop { get; set; } = Task.CompletedTask; + + public async Task StopAsync() + { + await Stop.CancelAsync().ConfigureAwait(false); + try { await Loop.ConfigureAwait(false); } catch (OperationCanceledException) { } + } + } +} +``` + +Notes for the engineer: +- `RemoteAgent` is `internal`, so `StartAsync`/`Connector` are internal too; the tool (Task 7) is in the same assembly and tests see internals. +- `WaitAsync` wakes early on a push; a push only means "check now" — its body is never trusted. +- The subscribe branch returns the latest task (via `GetTask`) when the stream ends without settling; the loop then reconnects after 1 s. + +- [ ] **Step 5: Run tests** + +Run: `dotnet test tests/Volcanic.AgenticCore.Extensions.Tests --filter "FullyQualifiedName~A2AWatchTests"` +Expected: 9 passed. Run it 3 more times to check for flakiness. + +- [ ] **Step 6: Commit** + +```bash +git add extensions/Volcanic.AgenticCore.Extensions.A2A tests/Volcanic.AgenticCore.Extensions.Tests/A2A +git commit -m "feat(a2a-client): watch long remote tasks by push, subscribe or poll and deliver the results" +``` + +--- + +### Task 7: The `a2a` tool and the extension + +**Files:** +- Create: `extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs` +- Create: `extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs` +- Create: `tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs` +- Test: `tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs`, `A2AHopsTests.cs` + +- [ ] **Step 1: The client test host** + +`tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/A2AClientHost.cs`: + +```csharp +using Microsoft.Extensions.DependencyInjection; +using Volcanic.AgenticCore.Abstractions.Events; +using Volcanic.AgenticCore.Abstractions.Extensions; +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); + } + + 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) { } + } +} +``` + +- [ ] **Step 2: Failing tool tests** + +`tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs`: + +```csharp +using System.Text; +using Volcanic.AgenticCore.Abstractions.Events; +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; + + [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", "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.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/" } ]"""); + var session = await host.CreateSessionAsync(); + var (text, _) = await host.CallToolAsync(session, new { action = "agents" }); + Assert.Contains("ghost", text); + Assert.Contains("unreachable", 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")); + } + + [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" }); + await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "two", contextId = Field(first, "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(); + var (text, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "take your time" }); + Assert.Contains("working", text); + Assert.Contains("delivered", text); + Assert.Single(host.Registry.List(A2AClientHost.Target)); + host.Faux.Enqueue(FauxTurn.Say("Thanks, noted")); + var deadline = DateTime.UtcNow.AddSeconds(15); + while (!host.Faux.Requests.Any(r => r.Messages.Any(m => m.Text.Contains("[A2A result from \"remote\""))) && DateTime.UtcNow < deadline) + await Task.Delay(50); + Assert.Contains(host.Faux.Requests, r => r.Messages.Any(m => m.Text.Contains("Slow answer"))); + } + + [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, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "here", files = new[] { "notes.txt" } }); + Assert.Contains(remote.Faux.Requests.Last().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"))); + } + + [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); + } + + [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"); + var (status, _) = await host.CallToolAsync(session, new { action = "status", agent = "remote", taskId }); + Assert.Contains("working", status); + var (cancelled, _) = await host.CallToolAsync(session, new { action = "cancel", agent = "remote", taskId }); + Assert.Contains("canceled", cancelled); + Assert.Empty(host.Registry.List(A2AClientHost.Target)); + Assert.Empty(PendingFile.Read(host.Workspace)); + } + + [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, _) = await host.CallToolAsync(session, new { action = "agents" }); + Assert.Contains(".volcanic/a2a/agents.json", text); + } +} +``` + +`tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AHopsTests.cs`: + +```csharp +using Volcanic.AgenticCore.Abstractions.Extensions; +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); + } +} +``` + +(`BeforeAgentStartEvent` runs at the start of every run, before the model call that triggers the tool, so the hop count is set by then. If its namespace differs, use `AgentStartEvent`.) + +- [ ] **Step 3: Run to see them fail** — build error. + +- [ ] **Step 4: Implement the tool** + +`extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs`: + +```csharp +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 +{ + 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) + return "No remote agents configured. Add them to .volcanic/a2a/agents.json: [{\"name\": \"travel-bot\", \"url\": \"https://…\", \"key\": \"…\"}]"; + 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 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, 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 A2AResults.Describe(entry.Name, task, A2AResults.SaveFiles(workspace, entry.Name, task, settings.MaxFileBytes)); + 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, async agent => + { + var task = await agent.Client.GetTaskAsync(new GetTaskRequest { Id = taskId }, ct).ConfigureAwait(false); + var files = A2AResults.IsSettled(task.Status.State) ? A2AResults.SaveFiles(workspace, entry.Name, task, settings.MaxFileBytes) : []; + return A2AResults.Describe(entry.Name, task, files); + }).ConfigureAwait(false); + } + + private async Task CancelAsync(string workspace, A2AArgs args, CancellationToken ct) + { + var entry = Entry(workspace, args.Agent); + var taskId = Required(args.TaskId, "taskId"); + await registry.StopWatchingAsync(target, workspace, taskId).ConfigureAwait(false); + return await CallAsync(entry, async agent => + { + var task = await agent.Client.CancelTaskAsync(new CancelTaskRequest { Id = taskId }, ct).ConfigureAwait(false); + return A2AResults.Describe(entry.Name, task, []); + }).ConfigureAwait(false); + } + + /// 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)) break; + } + } + 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, Func> call) + { + try + { + using var agent = await registry.Connector(settings).ConnectAsync(entry, CancellationToken.None).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})."); + } + } + + 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" : string.Join(", ", file.Agents.Select(a => a.Name)))}."); + } + + private Part FilePart(string workspace, string path) + { + var root = Path.GetFullPath(workspace).TrimEnd(Path.DirectorySeparatorChar) + Path.DirectorySeparatorChar; + var full = Path.GetFullPath(Path.Combine(workspace, path)); + if (!full.StartsWith(root, StringComparison.Ordinal)) throw new ArgumentException($"{path} is outside the workspace."); + var info = new FileInfo(full); + if (info.LinkTarget is not null && info.ResolveLinkTarget(true) is { } resolved && !resolved.FullName.StartsWith(root, StringComparison.Ordinal)) + throw new ArgumentException($"{path} points outside the workspace."); + 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)); + } + + 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; +} +``` + +(The tool's `WaitAsync` uses `CancelAfter` on real time; tests that need fake time only use `WaitSeconds = 0`.) + +- [ ] **Step 5: Implement the extension** + +`extensions/Volcanic.AgenticCore.Extensions.A2A/A2AClientExtension.cs`: + +```csharp +using System.Text.Json; +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 may start a run, which mustn't happen inside session creation. + _ = Task.Run(() => _registry.FlushUndeliveredAsync(target, workspace, CancellationToken.None)); + return ValueTask.CompletedTask; + }); + api.On((_, _, _) => + { + _registry.Live.Detach(target, api); + return ValueTask.CompletedTask; + }); + return ValueTask.CompletedTask; + } +} +``` + +- [ ] **Step 6: Run tests** + +Run: `dotnet test tests/Volcanic.AgenticCore.Extensions.Tests --filter "FullyQualifiedName~A2A"` +Expected: all A2A tests pass (13 tool/hops + earlier). Run 3 times for flakiness. If `Send_SlowTask_*` fails because the follow-up's `SendUserMessage` runs while the session is idle, check `MessageDelivery.FollowUp` semantics in `IExtension.cs` ("delivered when the agent has no more tool calls…"; idle → starts a run) — the core starts a run for follow-ups on an idle session. + +- [ ] **Step 7: Commit** + +```bash +git add extensions/Volcanic.AgenticCore.Extensions.A2A tests/Volcanic.AgenticCore.Extensions.Tests/A2A +git commit -m "feat(a2a-client): the a2a tool with agents, send, status and cancel, and per-session wiring" +``` + +--- + +### Task 8: Telegram host + +**Files:** +- Create: `hosts/Volcanic.AgenticCore.Telegram/Chats/ChatFollowUps.cs` +- Modify: `hosts/Volcanic.AgenticCore.Telegram/Scheduling/TelegramJobDelivery.cs` (use `ChatFollowUps`) +- Create: `hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AResultDelivery.cs` +- Create: `hosts/Volcanic.AgenticCore.Telegram/A2A/A2AWatchHostedService.cs` +- Create: `hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AClientSetup.cs` +- Modify: `hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs`, `Chats/ChatCommands.cs`, `Program.cs`, `appsettings.json`, `Volcanic.AgenticCore.Telegram.csproj` +- Modify: `tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs` +- Test: `tests/Volcanic.AgenticCore.Telegram.Tests/A2AClientHostTests.cs` + +- [ ] **Step 1: Failing tests** + +```csharp +using Microsoft.Extensions.Logging.Abstractions; +using Volcanic.AgenticCore.Extensions.A2A; +using Volcanic.AgenticCore.Telegram.A2A; +using Volcanic.AgenticCore.Telegram.Tests.Support; +using Volcanic.AgenticCore.Testing; + +namespace Volcanic.AgenticCore.Telegram.Tests; + +public class A2AClientHostTests +{ + [Fact] + public async Task ClientDisabled_NoA2aTool() + { + await using var host = new ChatHost(); + var chat = await host.ChatAsync(); + Assert.DoesNotContain(chat.Session.Tools, t => t.Name == "a2a"); + } + + [Fact] + public async Task ClientEnabled_ChatHasA2aTool() + { + await using var host = new ChatHost(a2aClient: new A2AClientSettings { Enabled = true }); + var chat = await host.ChatAsync(); + Assert.Contains(chat.Session.Tools, t => t.Name == "a2a"); + } + + [Fact] + public async Task ResultDelivery_GoesToChatWithTimestampHeader() + { + await using var host = new ChatHost(a2aClient: new A2AClientSettings { Enabled = true }); + host.Faux.Enqueue(FauxTurn.Say("Got the result")); + var delivery = new TelegramA2AResultDelivery(host.Chats, host.Settings, host.Options, TimeProvider.System, NullLogger.Instance); + Assert.True(await delivery.DeliverAsync("1001", "[A2A result from \"remote\", task t1 (completed)]\nDone", CancellationToken.None)); + var prompt = host.Faux.Requests.Last().Messages.Last().Text; + Assert.Contains("[A2A result from \"remote\"", prompt); + Assert.Matches(@"^\[.*\d{4}", prompt); // the local-time header comes first + Assert.Contains(host.Texts, t => t.Contains("Got the result")); + } + + [Fact] + public async Task ResultDelivery_UnknownOrNotAllowedTarget_False() + { + await using var host = new ChatHost(a2aClient: new A2AClientSettings { Enabled = true }); + var delivery = new TelegramA2AResultDelivery(host.Chats, host.Settings, host.Options, TimeProvider.System, NullLogger.Instance); + Assert.False(await delivery.DeliverAsync("999", "x", CancellationToken.None)); + Assert.False(await delivery.DeliverAsync("not-a-chat", "x", CancellationToken.None)); + Assert.Equal(["1001"], delivery.KnownTargets()); + } + + [Fact] + public async Task A2aCommand_ListsAgentsAndWatches() + { + await using var host = new ChatHost(a2aClient: new A2AClientSettings { Enabled = true }); + var ws = host.WorkspaceOf(); + Directory.CreateDirectory(Path.GetDirectoryName(AgentsFile.PathFor(ws))!); + File.WriteAllText(AgentsFile.PathFor(ws), """[ { "name": "travel-bot", "url": "https://travel.example/", "key": "agent-key-123456789" } ]"""); + await host.SendAsync("/a2a"); + await host.SettleAsync(); + var reply = host.LastReply(); + Assert.Contains("travel-bot", reply); + Assert.Contains("https://travel.example/", reply); + Assert.DoesNotContain("agent-key-123456789", reply); + Assert.Contains("Watches", reply); + } + + [Fact] + public async Task A2aCommand_BothOff_SaysHowToEnable() + { + await using var host = new ChatHost(); + await host.SendAsync("/a2a"); + await host.SettleAsync(); + var reply = host.LastReply(); + Assert.Contains("A2A is off", reply); + Assert.Contains("A2AClient__Enabled", reply); + } +} +``` + +- [ ] **Step 2: Run to see them fail** — build error (`a2aClient:` parameter, `TelegramA2AResultDelivery`). + +- [ ] **Step 3: Extract `ChatFollowUps`** + +`hosts/Volcanic.AgenticCore.Telegram/Chats/ChatFollowUps.cs`: + +```csharp +using Volcanic.AgenticCore.Abstractions.Extensions.Events; +using Volcanic.AgenticCore.Telegram.Inbound; + +namespace Volcanic.AgenticCore.Telegram.Chats; + +/// +/// Runs a prompt in a chat's session as a follow-up with the local-time header, never through the inbox (so it is never +/// debounced into a user's message). Idle: returns when the run settles. Busy: queued, returns when the run that took it +/// settles. +/// +internal static class ChatFollowUps +{ + public static async Task RunAsync(ChatContext chat, string prompt, TimeZoneInfo zone, TimeProvider time, CancellationToken cancellationToken) + { + var text = MessageContext.TimestampHeader(time.GetUtcNow(), zone) + "\n" + prompt; + var queued = chat.Session.IsStreaming; + await chat.Session.PromptAsync(text, null, StreamingBehavior.FollowUp, cancellationToken).ConfigureAwait(false); + if (queued) await chat.Session.WaitForIdleAsync(cancellationToken).ConfigureAwait(false); + } +} +``` + +In `TelegramJobDelivery.RunPromptAsync`, replace the three lines that build `text`, check `queued`, call `PromptAsync` and `WaitForIdleAsync` with `await ChatFollowUps.RunAsync(chat, prompt, TimeZoneFor(target), time, cancellationToken).ConfigureAwait(false);` (keep its try/catch and comment). Run the scheduler chat tests: `dotnet test tests/Volcanic.AgenticCore.Telegram.Tests --filter "FullyQualifiedName~Scheduled"` — still green. + +- [ ] **Step 4: Delivery, hosted service, setup** + +`hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AResultDelivery.cs`: + +```csharp +using System.Globalization; +using Microsoft.Extensions.Logging; +using Volcanic.AgenticCore.Extensions.A2A; +using Volcanic.AgenticCore.Telegram.Chats; + +namespace Volcanic.AgenticCore.Telegram.A2A; + +/// A2A watch results as follow-up prompts in the chat (target = chat id), like scheduled prompts. +public sealed class TelegramA2AResultDelivery(ChatRegistry chats, ChatSettingsStore settings, TelegramOptions options, TimeProvider time, + ILogger logger) : IA2AResultDelivery +{ + public async Task DeliverAsync(string target, string text, CancellationToken cancellationToken) + { + if (!long.TryParse(target, NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture, out var chatId) || !options.IsAllowed(chatId)) + return false; + var chat = await chats.GetAsync(chatId, cancellationToken).ConfigureAwait(false); + try + { + await ChatFollowUps.RunAsync(chat, text, settings.TimeZoneFor(chatId), time, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogError(ex, "A2A result in chat {ChatId} failed", chatId); + chat.Report($"An A2A result arrived but couldn't be processed: {ex.Message}"); + } + return true; + } + + public string? WorkspaceFor(string target) => + long.TryParse(target, NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture, out var chatId) ? ChatRegistry.WorkspaceFor(options, chatId) : null; + + public IEnumerable KnownTargets() => options.AllowedUserIds.Select(id => id.ToString(CultureInfo.InvariantCulture)).ToList(); +} +``` + +`hosts/Volcanic.AgenticCore.Telegram/A2A/A2AWatchHostedService.cs`: + +```csharp +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Volcanic.AgenticCore.Extensions.A2A; + +namespace Volcanic.AgenticCore.Telegram.A2A; + +/// Points the watch registry at the chats and re-arms persisted watches; stops them on shutdown. +public sealed class A2AWatchHostedService(A2AWatchRegistry registry, A2AClientSettings settings, TelegramA2AResultDelivery delivery, + ILogger logger) : IHostedService +{ + public async Task StartAsync(CancellationToken cancellationToken) + { + registry.Logger = logger; + registry.UseDelivery(delivery); + var count = await registry.RestoreAsync(settings, cancellationToken).ConfigureAwait(false); + logger.LogInformation("A2A client: {Count} watch(es) restored", count); + } + + public Task StopAsync(CancellationToken cancellationToken) => registry.StopAsync(); +} +``` + +`hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AClientSetup.cs`: + +```csharp +using Microsoft.Extensions.DependencyInjection; +using Volcanic.AgenticCore.A2A.Server; +using Volcanic.AgenticCore.Extensions.A2A; + +namespace Volcanic.AgenticCore.Telegram.A2A; + +/// +/// The A2A client (off unless A2AClient__Enabled=true). Call before : when the +/// server is on, its push receiver is routed into the watch registry and its public URL becomes the push base URL. +/// +public static class TelegramA2AClientSetup +{ + public static void Register(IServiceCollection services, A2AClientSettings client, A2AHostOptions server, A2AWatchRegistry registry) + { + services.AddSingleton(client); + if (!client.Enabled) return; + client.PushBaseUrl = server.Enabled ? server.PublicUrl : null; + if (server.Enabled) server.OnPush = registry.OnPushAsync; + services.AddSingleton(registry); + services.AddSingleton(); + services.AddHostedService(); + } + + public static string Describe(A2AClientSettings client) => + client.Enabled + ? $"A2A client: on, agents in each chat's .volcanic/a2a/agents.json, {(client.PushBaseUrl is null ? "watches subscribe or poll" : "push at " + client.PushBaseUrl)}" + : "A2A client: off"; +} +``` + +Add `` to the Telegram csproj. + +- [ ] **Step 5: ChatRegistry, ChatCommands, Program, appsettings** + +`ChatRegistry`: add constructor parameters `A2AClientSettings? a2aClient = null, A2AWatchRegistry? a2aWatches = null` (last), and in `CreateAsync` after the triggers extra: + +```csharp + if (a2aClient is { Enabled: true }) + extras.Add(ExtensionRegistration.FromInstance(new A2AClientExtension(a2aClient, Scheduling.TelegramJobDelivery.TargetOf(chatId), a2aWatches))); +``` + +`ChatCommands`: add constructor parameters `A2AClientSettings? a2aClient = null, A2AWatchRegistry? a2aWatches = null` (last). Replace `A2AStatus` with: + +```csharp + private string A2AStatus(ChatContext chat) + { + var serverOn = a2a?.Value is { Enabled: true }; + var clientOn = a2aClient is { Enabled: true }; + if (!serverOn && !clientOn) + return "A2A is off. Set A2A__Enabled=true (and A2A__PublicUrl) to let other agents talk to this assistant, " + + "A2AClient__Enabled=true to let it talk to other agents."; + var workspace = chat.Session.Cwd; + var lines = new List(); + if (serverOn) + { + var callers = CallersFile.Read(workspace, Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance); + lines.Add("**A2A callers** (.volcanic/a2a/callers.json)"); + lines.AddRange(callers.Count == 0 ? ["none"] : callers.Select(c => $"• {c.Name} ({c.Mode}) key {Mask(c.Key)}")); + var tasks = A2ATaskFiles.Recent(workspace, 10); + lines.Add("**Recent A2A tasks**"); + lines.AddRange(tasks.Count == 0 ? ["none"] : tasks.Select(t => $"• {t.Caller}: {t.State}{Age(t.Updated)} ({t.TaskId[..Math.Min(8, t.TaskId.Length)]})")); + } + else lines.Add("A2A server: off (A2A__Enabled)."); + if (clientOn) + { + var agents = AgentsFile.Read(workspace); + lines.Add("**Remote agents** (.volcanic/a2a/agents.json)"); + if (agents.Error is not null) lines.Add(agents.Error); + lines.AddRange(agents.Agents.Count == 0 ? ["none"] : agents.Agents.Select(a => $"• {a.Name}: {a.Url}{(a.Key is null ? "" : " key " + Mask(a.Key))}")); + var watches = a2aWatches?.List(Scheduling.TelegramJobDelivery.TargetOf(chat.ChatId)) ?? []; + lines.Add("**Watches**"); + lines.AddRange(watches.Count == 0 ? ["none"] : watches.Select(w => $"• {w.Agent}: task {w.TaskId[..Math.Min(8, w.TaskId.Length)]} by {w.Source}{Age(w.StartedAt)}")); + } + else lines.Add("A2A client: off (A2AClient__Enabled)."); + return string.Join('\n', lines); + } + + private static string Mask(string key) => key.Length >= 12 ? "…" + key[^4..] : "…"; +``` + +(Add `using Volcanic.AgenticCore.Extensions.A2A;`. Keep the existing `Age` helper.) + +`Program.cs`: after reading `a2a` (server options): + +```csharp +var a2aClient = builder.Configuration.GetSection(A2AClientSettings.SectionName).Get() ?? new A2AClientSettings(); +``` + +and **before** `TelegramA2ASetup.Register(...)`: + +```csharp +TelegramA2AClientSetup.Register(builder.Services, a2aClient, a2a, A2AWatchRegistry.Shared); +``` + +Pass the client into `ChatRegistry` and `ChatCommands`: both are DI-constructed with optional parameters, and `A2AClientSettings` is always registered; `A2AWatchRegistry` only when enabled — so they resolve on their own. Log `TelegramA2AClientSetup.Describe(a2aClient)` next to the A2A server line. + +`appsettings.json`: add `"A2AClient": { "Enabled": false, "WaitSeconds": 60, "WatchTimeout": "1.00:00:00", "CardTimeoutSeconds": 10, "MaxFileBytes": 10485760 }`. + +`ChatHost` (tests): add parameter `A2AClientSettings? a2aClient = null`; when given, create `A2AWatchRegistry = new A2AWatchRegistry(clock, new HttpClientHandler())` (expose as a property) and pass both to `ChatRegistry` and `ChatCommands`; dispose the registry in `DisposeAsync`. + +- [ ] **Step 6: Run tests** + +Run: `dotnet build && dotnet test tests/Volcanic.AgenticCore.Telegram.Tests` +Expected: 0 warnings, all green (6 new). The existing `A2aCommand_DisabledReply` still expects "A2A is off" — still true. + +- [ ] **Step 7: Commit** + +```bash +git add hosts/Volcanic.AgenticCore.Telegram tests/Volcanic.AgenticCore.Telegram.Tests +git commit -m "feat(telegram): A2A client per chat behind A2AClient__Enabled, watch results as follow-ups, /a2a lists agents" +``` + +--- + +### Task 9: Compose, docs, spec + +**Files:** +- Modify: `docker-compose.yml`, `docs/a2a.md`, `docs/extensions.md`, `docs/telegram-bot.md`, `docs/docker.md`, `CLAUDE.md`, `docs/superpowers/specs/2026-10-07-a2a-client-design.md` + +- [ ] **Step 1: Compose** — under `environment:` add + `- A2AClient__Enabled=${A2A_CLIENT_ENABLED:-false} # A2A client: the a2a tool and watches (a2a-client spec)`. + Run the compose tests: `dotnet test tests/Volcanic.AgenticCore.Telegram.Tests --filter "FullyQualifiedName~DockerCompose"`. + +- [ ] **Step 2: Docs** — describe what the code does (read it): + - `docs/a2a.md`: a "Client" part: enabling (`A2AClient__Enabled`, independent of the server; only the owner's chats get the tool, not inbound A2A sessions), `agents.json` (example, rules, read per call, keys never shown), the `a2a` tool actions and results (context/task ids, input-required answering, files in/out under `inbox/a2a//`), watches (push only when the server is on with `PublicUrl`; else subscribe/poll with backoff; `WatchTimeout`; `pending.json`; restart restore; delivery as a follow-up with the time header; undelivered results), hops, config table, security notes (agents.json keys are plaintext in the workspace; the agent can edit agents.json and therefore choose where it sends data — same trust as its bash access), a Volcanic↔Volcanic example (two workspaces, callers.json on one side, agents.json on the other). + - `docs/extensions.md`: list the A2A extension with one paragraph. + - `docs/telegram-bot.md`: `A2AClient__Enabled` in the config table, `/a2a` additions. + - `docs/docker.md`: `A2A_CLIENT_ENABLED`. + - `CLAUDE.md`: add the extension to the Extensions paragraph; mention the client in the Telegram `A2A/` sentence. + - Spec: Status → implemented; record deviations if any arose. + +- [ ] **Step 3: Full verification** + +Run: `dotnet build && dotnet test` +Expected: 0 warnings; every project green. + +- [ ] **Step 4: Commit** + +```bash +git add docker-compose.yml docs CLAUDE.md +git commit -m "docs(a2a): A2A client guide, configuration and Telegram notes" +``` diff --git a/docs/superpowers/specs/2026-10-07-a2a-client-design.md b/docs/superpowers/specs/2026-10-07-a2a-client-design.md index f6b6a92..9733d5c 100644 --- a/docs/superpowers/specs/2026-10-07-a2a-client-design.md +++ b/docs/superpowers/specs/2026-10-07-a2a-client-design.md @@ -28,6 +28,30 @@ possibly after asking the user), and get results of long-running tasks back with - **`input-required`** is returned to the model with the `taskId`; the model answers with `send` and that `taskId`. - **Hops**: each outgoing message carries `metadata["volcanic.hops"]` = inbound hops + 1. +## Revisions (2026-10-08, before implementation) + +Decided with the owner after the server shipped; these override the sections below where they differ. + +- **Off by default, its own switch**: configuration section `A2AClient` (env `A2AClient__…`), `Enabled = false`. + Independent of the server's `A2A__Enabled`. Off: the extension isn't added to any session, no watches are restored, + `/a2a` says the client is off. +- **Only the owner's own chat sessions get the `a2a` tool.** The Telegram host adds the extension per chat (like the + scheduler), never host-wide, so isolated A2A server sessions (driven by untrusted callers) can't call other agents + with the keys in `agents.json`. The hop counter still listens to `a2a.inbound` for hosts that add it elsewhere. +- **Settings**: `Enabled`, `WaitSeconds` (60), `WatchTimeout` (1 day), `CardTimeoutSeconds` (10), `MaxFileBytes` (10 MB). + The push base URL isn't a client setting: the host passes the server's `A2A:PublicUrl` only when the server is on + (the push receiver lives there). Without it, watches subscribe or poll. +- **Send = `returnImmediately` + wait.** `send` always calls `SendMessage` with `returnImmediately: true`, then waits up to + `WaitSeconds` by subscribing (streaming agents) or polling `GetTask`. A dropped streaming request would cancel the + remote run on servers that tie the run to the request (ours does), so the client never streams the send itself. + Agents that ignore `returnImmediately` simply answer late (the call is bounded by `WatchTimeout`). +- **Delivery contract**: `IA2AResultDelivery { Task DeliverAsync(target, text, ct); string WorkspaceFor(target); + IEnumerable KnownTargets(); }` (targets = Telegram chat ids, like the scheduler). The default + `LiveSessionDelivery` delivers through the live session's `IExtensionApi.SendUserMessage(FollowUp)` and returns false + without one; undelivered results stay in `pending.json` and go out on that target's next session start. +- **The registry owns one HTTP handler** (`A2AWatchRegistry`, process-wide `Shared`); tests inject a handler that reaches + an in-process server. + ## `agents.json` ```json @@ -47,15 +71,17 @@ The tool never includes keys in its results or errors. ## Configuration -Read from the same `A2A` section as the server (env `A2A__…`), passed through `A2AClientOptions`: +Section `A2AClient` (env `A2AClient__…`), bound to `A2AClientSettings` (the SDK already has an `A2AClientOptions`): | Key | Default | Meaning | |---|---|---| +| `Enabled` | `false` | add the `a2a` tool to the owner's chats, restore watches | | `WaitSeconds` | `60` | how long `send` blocks before handing over to a watch | | `WatchTimeout` | `1.00:00:00` | when a watch gives up | -| `PublicUrl` | – | our external base URL; required for push | -| `MaxFileBytes` | `10485760` | cap per file part, in and out | | `CardTimeoutSeconds` | `10` | per-agent card fetch timeout | +| `MaxFileBytes` | `10485760` | cap per file part, in and out | + +The push base URL comes from the server's `A2A:PublicUrl`, and only while `A2A:Enabled` is on. ## The `a2a` tool @@ -65,8 +91,8 @@ Arguments: `action` (`agents` \| `send` \| `status` \| `cancel`), `agent`, `mess `supportsExtendedAgentCard`, the extended card. Returns per agent: name, description (entry override or card), skills (id, name, description, examples), streaming/push capabilities, or `unreachable: `. - **`send`** `{agent, message, contextId?, taskId?, files?}` — builds a message (text part, file parts read from workspace - paths, cap `MaxFileBytes`), metadata `volcanic.hops`. Uses `SendStreamingMessage` when the card has `streaming`, - else `SendMessage` with `returnImmediately: true` followed by `GetTask` polling. Then, within `WaitSeconds`: + paths, cap `MaxFileBytes`), metadata `volcanic.hops`. Calls `SendMessage` with `returnImmediately: true`, then waits by + `SubscribeToTask` (card has `streaming`) or `GetTask` polling. Then, within `WaitSeconds`: - terminal (`completed`, `failed`, `canceled`, `rejected`) → returns state, the agent's final text, the artifact list, `contextId`, `taskId`. File artifacts are saved under `/inbox/a2a//` and returned as paths; text artifacts are inlined (long ones saved to a file and returned as a path, by the tool's normal truncation); @@ -89,7 +115,7 @@ tool never throws into the loop. Watch record, persisted in `/.volcanic/a2a/pending.json`: ```json -{ "watchId": "…", "sessionScope": "telegram-111", "agent": "travel-bot", "taskId": "…", "contextId": "…", +{ "watchId": "…", "target": "111", "agent": "travel-bot", "taskId": "…", "contextId": "…", "pushToken": "…", "startedAt": "…", "source": "push" } ``` @@ -138,8 +164,9 @@ message sends `hops + 1`. The server refuses above `MaxHops`. ## Telegram host -- References the extension, registers it host-wide with `TelegramA2AResultDelivery` and `A2AClientOptions` from config. -- Wires `OnPush` to the registry and calls `RestoreAsync` over allowed chats' workspaces at startup. +- With `A2AClient:Enabled`, adds the extension to each chat session (target = chat id), uses + `TelegramA2AResultDelivery`, and restores watches over the allowed chats at startup. +- When the server is on too, wires its `OnPush` to the registry and passes `A2A:PublicUrl` as the push base URL. - `/a2a` additionally lists remote agents from `agents.json` (name, url, key masked) and active watches (agent, task, source, age).