diff --git a/docs/a2a.md b/docs/a2a.md index 5caf6be..267075f 100644 --- a/docs/a2a.md +++ b/docs/a2a.md @@ -260,7 +260,7 @@ One tool, `a2a`, with `action` and, depending on it, `agent`, `message`, `contex |---|---| | `agents` | For every entry, fetches the card (`CardTimeoutSeconds`) and lists name, description, card name, `streaming`/`push` capabilities and skills (id, name, description, up to three examples), or `unreachable ()` | | `send` | Sends `message` and/or `files` to `agent`; `contextId` continues an earlier conversation with it, `taskId` answers its question on that task | -| `status` | `GetTask` for `taskId`: same result shape as `send` | +| `status` | `GetTask` for `taskId`: same result shape as `send`; a settled task's watch ends (its outcome is in the conversation now, so it isn't delivered again) | | `cancel` | Stops this chat's watch of `taskId`, then `CancelTask` | **send** calls `SendMessage` with `returnImmediately: true` and `metadata["volcanic.hops"]` (see [Hops](#hops)), then @@ -314,7 +314,10 @@ because the task already finished), the result says so and carries the task's ou or doesn't answer in time, the watch is re-armed and the error ends with `The task is still watched.` Errors are error tool results with a readable message: unknown agent (with the configured names), `rejected the key -(401)`, `Could not reach agent ...`, `Agent ... refused: ... ()`, `did not answer in time`. Every single remote +(401)`, `Could not reach agent ...`, `Agent ... refused: ... ()`, `did not answer in time`. A `send` with a +`contextId` that the remote refuses as busy (`UnsupportedOperation` mentioning "busy", as ours and VolcanicAgents do) +says that nothing was sent, names the task this chat watches in that context (`its result will arrive here by +itself`) and suggests sending again after it or without `contextId`. A refusal never touches a watch. Every single remote call is bounded by 60 seconds; an agent that ignores `returnImmediately` and needs longer than that to answer `SendMessage` gets `did not answer in time`, and since its task id never arrived, nothing watches that task. @@ -333,14 +336,20 @@ How it learns about updates, chosen when the watch starts: body is never trusted. A safety `GetTask` runs every 10 minutes. If the remote refuses the push config, the watch subscribes or polls instead. 2. **subscribe**: the card advertises streaming. A stream that drops or ends before the task settles is reopened after - 1 s, 5 s, then 15 s; after 3 failures in a row the watch polls instead. A stream is also reopened after a day. + 1 s, 5 s, then 15 s; after 3 failures in a row the watch polls instead. No stream is held for more than 5 minutes: + the watch then reads the task with `GetTask` and subscribes again (a pause, not a failure). Remotes send nothing + while their agent works without writing text, and such a quiet stream can die on the way (a proxy or NAT dropping + it) without either side noticing; before this slice a watch could wait on a dead stream until `WatchTimeout`. 3. **poll**: `GetTask` after 15 s, 30 s, 1 min, 2 min, then every 5 min. Each check reads `agents.json` again: an agent removed from it ends the watch with a `(stopped)` notice; a malformed file is treated as a passing failure (the user may be editing it) and the watch keeps going. A watch keeps its connection (card and client) between checks and reconnects after a failure or when the agent's entry changes. After `WatchTimeout` (1 day by default) from its start the watch gives up with a `(no result)` notice suggesting -`action=status`. On a settled task it saves files like `send` does and, for push watches, deletes the push config +`action=status`. The shared handler turns on TCP keepalive (probes after 60 s of silence, every 15 s, 4 tries) and +HTTP/2 pings, so a dead connection fails instead of hanging. The log has one line when a watch starts or is restored +(task, agent, chat, source, watch id), when a stream slice ends on a running task (`subscribing again`), when a +result is delivered or kept, and for every failed check. On a settled task it saves files like `send` does and, for push watches, deletes the push config (best effort). The delivered text is a follow-up prompt: diff --git a/docs/docker.md b/docs/docker.md index 93bf849..787fcdf 100644 --- a/docs/docker.md +++ b/docs/docker.md @@ -102,6 +102,7 @@ The compose file maps them into the container. `TELEGRAM_MODEL` and `LOG_LEVEL` are read by the compose file only, which passes them to the bot as `Telegram__Model` and `Logging__LogLevel__Default`; `INSTALL_PLAYWRIGHT` and `CLAUDE_CODE_VERSION` are build arguments (see above). +Every log line starts with its UTC time (`2026-10-16 13:48:00 UTC info: ...`, `Logging:Console` in `appsettings.json`). > **Never set `ANTHROPIC_API_KEY`, `ANTHROPIC_AUTH_TOKEN`, `ANTHROPIC_BASE_URL`, `CLAUDE_CODE_OAUTH_TOKEN`, > `CLAUDE_CODE_USE_BEDROCK` or `CLAUDE_CODE_USE_VERTEX`.** The Claude Code provider runs on a Claude subscription only 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 da4d55e..5a6eb07 100644 --- a/docs/superpowers/specs/2026-10-07-a2a-client-design.md +++ b/docs/superpowers/specs/2026-10-07-a2a-client-design.md @@ -101,10 +101,13 @@ Arguments: `action` (`agents` \| `send` \| `status` \| `cancel`), `agent`, `mess - still running → starts a watch, returns `{state: working, taskId, contextId, note: "the result will be delivered to this conversation when it is ready"}`. A message-only reply (the agent answered with a `Message`, no task) is returned as text with its `contextId`. -- **`status`** `{agent, taskId}` — `GetTask`; same result shape as a finished `send`. +- **`status`** `{agent, taskId}` — `GetTask`; same result shape as a finished `send`. A settled task's watch ends, so its + outcome isn't delivered a second time. - **`cancel`** `{agent, taskId}` — `CancelTask`, removes any watch for it. -Errors (network, timeout, 401, JSON-RPC error, unknown agent) become error tool results with a readable message. The +Errors (network, timeout, 401, JSON-RPC error, unknown agent) become error tool results with a readable message. A +"context busy" refusal of a `send` with `contextId` names the task this chat watches there and says its result will +arrive by itself; a refusal never touches a watch. The tool never throws into the loop. ## Watches @@ -125,7 +128,9 @@ Watch record, persisted in `/.volcanic/a2a/pending.json`: 1. **push**: card has `capabilities.pushNotifications` and `PublicUrl` is set → `CreateTaskPushNotificationConfig` with URL `/a2a/push/` and a random 32-byte token. A safety `GetTask` runs every 10 min. 2. **subscribe**: card has `streaming` → `SubscribeToTask`; on disconnect, reconnect with backoff, falling back to polling - after 3 failures. + after 3 failures. A stream is held at most 5 minutes, then `GetTask` and a new subscription (a pause, not a + failure): a quiet stream may have died silently on the way (2026-10-16: hour-long tasks never delivered). The + shared handler uses TCP keepalive and HTTP/2 pings. 3. **poll**: `GetTask` with backoff 15 s → 30 s → 1 min → 2 min → 5 min (cap). - **Push receiving**: the host wires the server's `A2AServerOptions.OnPush` to `A2AWatchRegistry.Shared.OnPushAsync(watchId, token, body)`, which checks the token (constant-time) and returns false for unknown watches or bad tokens (→ 404). diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs index 7e61c35..1de7729 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2ATool.cs @@ -97,7 +97,9 @@ internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatc return await CallAsync(entry, ct, async agent => { SendMessageResponse response; - using (var bounded = Bounded(ct)) + try + { + using var bounded = Bounded(ct); response = await agent.Client.SendMessageAsync(new SendMessageRequest { Message = new A2AMessage @@ -111,6 +113,11 @@ internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatc Configuration = new SendMessageConfiguration { ReturnImmediately = true }, Metadata = new Dictionary { ["volcanic.hops"] = JsonSerializer.SerializeToElement(inboundHops() + 1) }, }, bounded.Token).ConfigureAwait(false); + } + catch (A2AException ex) when (IsContextBusy(ex) && !string.IsNullOrWhiteSpace(args.ContextId)) + { + throw new InvalidOperationException(BusyContext(entry, args.ContextId!), ex); + } if (response.Task is not { } task) { @@ -150,12 +157,29 @@ internal sealed class A2ATool(A2AClientSettings settings, string target, A2AWatc return await CallAsync(entry, ct, async agent => { var task = await GetTaskAsync(agent, taskId, ct).ConfigureAwait(false); - return A2AResults.IsSettled(task.Status.State) - ? await DescribeSettledAsync(workspace, entry.Name, agent, task, ct).ConfigureAwait(false) - : A2AResults.Describe(entry.Name, task, []); + if (!A2AResults.IsSettled(task.Status.State)) return A2AResults.Describe(entry.Name, task, []); + var described = await DescribeSettledAsync(workspace, entry.Name, agent, task, ct).ConfigureAwait(false); + // The outcome is in this conversation now: a watch of the task would only deliver it a second time. + await registry.StopWatchingAsync(target, workspace, taskId).ConfigureAwait(false); + return described; }).ConfigureAwait(false); } + /// The remote refuses a message to a context whose task still runs (our server and VolcanicAgents say "Context busy"). + private static bool IsContextBusy(A2AException ex) => + ex.ErrorCode == A2AErrorCode.UnsupportedOperation && ex.Message.Contains("busy", StringComparison.OrdinalIgnoreCase); + + private string BusyContext(RemoteAgentEntry entry, string contextId) + { + var running = registry.List(target).FirstOrDefault(r => r.Agent == entry.Name && r.ContextId == contextId); + return running is null + ? $"Agent \"{entry.Name}\" is still working on an earlier task in context {contextId} and refused this message. " + + "Nothing was sent. Send it again once that task has ended, or start a new conversation without contextId." + : $"Agent \"{entry.Name}\" is still working on task {running.TaskId} in context {contextId} and refused this message. " + + "Nothing was sent, and that task is still watched: its result will arrive here by itself. " + + "Send this message again after that result, or start a new conversation without contextId."; + } + private async Task CancelAsync(string workspace, A2AArgs args, CancellationToken ct) { var entry = Entry(workspace, args.Agent); diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs index 7a66e72..ad75799 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs @@ -1,4 +1,6 @@ using System.Collections.Concurrent; +using System.Net; +using System.Net.Sockets; using System.Security.Cryptography; using System.Text; using System.Text.Json; @@ -25,15 +27,43 @@ namespace Volcanic.AgenticCore.Extensions.A2A; /// public sealed class A2AWatchRegistry : IAsyncDisposable { - public static A2AWatchRegistry Shared { get; } = new(TimeProvider.System, - new SocketsHttpHandler { PooledConnectionLifetime = TimeSpan.FromMinutes(5) }); + public static A2AWatchRegistry Shared { get; } = new(TimeProvider.System, CreateSharedHandler()); + + /// + /// The shared handler: pooled connections renewed every 5 minutes, and TCP keepalive (plus HTTP/2 pings) so a + /// connection that died silently on the way (a proxy or NAT dropping an idle stream) fails instead of hanging. + /// + internal static SocketsHttpHandler CreateSharedHandler() => new() + { + PooledConnectionLifetime = TimeSpan.FromMinutes(5), + KeepAlivePingDelay = TimeSpan.FromSeconds(60), + KeepAlivePingTimeout = TimeSpan.FromSeconds(20), + ConnectCallback = (context, ct) => ConnectWithKeepAliveAsync(context.DnsEndPoint, ct), + }; + + /// A TCP connection probed after 60 s of silence, every 15 s, given up after 4 unanswered probes. + internal static async ValueTask ConnectWithKeepAliveAsync(DnsEndPoint endPoint, CancellationToken cancellationToken) + { + var socket = new Socket(SocketType.Stream, ProtocolType.Tcp) { NoDelay = true }; + try + { + socket.SetSocketOption(SocketOptionLevel.Socket, SocketOptionName.KeepAlive, true); + socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveTime, 60); + socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveInterval, 15); + socket.SetSocketOption(SocketOptionLevel.Tcp, SocketOptionName.TcpKeepAliveRetryCount, 4); + await socket.ConnectAsync(endPoint, cancellationToken).ConfigureAwait(false); + return new NetworkStream(socket, ownsSocket: true); + } + catch + { + socket.Dispose(); + throw; + } + } public static readonly TimeSpan[] DefaultPollDelays = [TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(30), TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(2), TimeSpan.FromMinutes(5)]; - /// Longest single subscription; the stream is then reopened (a pause, not a failure). - private static readonly TimeSpan MaxSubscription = TimeSpan.FromDays(1); - private static readonly TimeSpan DeleteTimeout = TimeSpan.FromSeconds(10); /// After this many failed checks in a row, a subscribe watch polls instead. @@ -79,6 +109,13 @@ public sealed class A2AWatchRegistry : IAsyncDisposable internal TimeSpan[] ReconnectDelays { get; init; } = [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(15)]; internal TimeSpan SafetyPollInterval { get; init; } = TimeSpan.FromMinutes(10); + /// + /// Longest single subscription: the watch then reads the task with GetTask and subscribes again (a pause, not + /// a failure). Remotes send nothing while their agent works without text, and a stream that went quiet may have died + /// on the way without either side noticing, so no stream is trusted for longer than this. + /// + internal TimeSpan SubscriptionSlice { get; init; } = TimeSpan.FromMinutes(5); + /// Bound of one remote call (get, create/delete push config), never past the watch deadline. internal TimeSpan CallTimeout { get; init; } = TimeSpan.FromSeconds(60); @@ -157,6 +194,8 @@ public sealed class A2AWatchRegistry : IAsyncDisposable watch.DisposeStop(); return watch.Record; } + Logger.LogInformation("A2A: watching task {TaskId} of {Agent} for {Target} by {Source} (watch {WatchId})", + task.Id, agent.Entry.Name, target, watch.Record.Source, watchId); watch.Loop = Task.Run(() => RunAsync(watch, settings)); return watch.Record; } @@ -199,6 +238,8 @@ public sealed class A2AWatchRegistry : IAsyncDisposable if (!ReferenceEquals(record, stored)) PendingFile.Upsert(workspace, record); // Checked after the watch is in the map: a StopAsync that set the flag later sees (and stops) it. if (_stopped) { _watches.TryRemove(new(record.WatchId, watch)); watch.DisposeStop(); return count; } + Logger.LogInformation("A2A: restored the watch of task {TaskId} of {Agent} for {Target} by {Source} (watch {WatchId})", + record.TaskId, record.Agent, target, record.Source, record.WatchId); watch.Loop = Task.Run(() => RunAsync(watch, settings)); count++; } @@ -230,6 +271,7 @@ public sealed class A2AWatchRegistry : IAsyncDisposable { if (_stopped) break; // shutting down: the rest stay in pending.json for the next start if (!await TryDeliverAsync(target, record.Undelivered!, cancellationToken).ConfigureAwait(false)) { failed = true; continue; } + Logger.LogInformation("A2A: delivered the kept result of task {TaskId} of {Agent} to {Target}", record.TaskId, record.Agent, target); using (await LockAsync(workspace).ConfigureAwait(false)) PendingFile.Remove(workspace, record.WatchId); } @@ -343,6 +385,9 @@ public sealed class A2AWatchRegistry : IAsyncDisposable (task, pause) = await SubscribeAsync(agent, record.TaskId, deadline, ct).ConfigureAwait(false); if ((task is null || !A2AResults.IsSettled(task.Status.State)) && !pause) throw new IOException("the task stream ended before the task settled"); + if (pause && task is not null && !A2AResults.IsSettled(task.Status.State)) + Logger.LogInformation("A2A: task {TaskId} of {Agent} is still {State}; subscribing again", record.TaskId, record.Agent, + A2AResults.StateName(task.Status.State)); } else { @@ -415,7 +460,7 @@ public sealed class A2AWatchRegistry : IAsyncDisposable /// /// Follows the task's stream until it settles or ends, then reads the task. Paused: the stream was ended by - /// us (the deadline or ), which is not a failure. + /// us (the deadline or ), which is not a failure. /// private async Task<(AgentTask? Task, bool Paused)> SubscribeAsync(RemoteAgent agent, string taskId, DateTimeOffset deadline, CancellationToken ct) { @@ -424,7 +469,7 @@ public sealed class A2AWatchRegistry : IAsyncDisposable using (var until = CancellationTokenSource.CreateLinkedTokenSource(ct)) { var remaining = deadline - Time.GetUtcNow(); - until.CancelAfter(remaining <= TimeSpan.Zero ? TimeSpan.Zero : Min(remaining, MaxSubscription)); + until.CancelAfter(remaining <= TimeSpan.Zero ? TimeSpan.Zero : Min(remaining, SubscriptionSlice)); try { await foreach (var ev in agent.Client.SubscribeToTaskAsync(new SubscribeToTaskRequest { Id = taskId }, until.Token).ConfigureAwait(false)) @@ -436,7 +481,8 @@ public sealed class A2AWatchRegistry : IAsyncDisposable catch (A2AException) // e.g. the task is already terminal: read it instead { } - catch (OperationCanceledException) when (!ct.IsCancellationRequested && until.IsCancellationRequested) + // Ending a stream mid-read may surface as an I/O error rather than a cancel: ours either way. + catch (Exception ex) when (ex is not OutOfMemoryException && !ct.IsCancellationRequested && until.IsCancellationRequested) { paused = true; } @@ -488,6 +534,10 @@ public sealed class A2AWatchRegistry : IAsyncDisposable { // A failed delivery writes nothing: the text is stored, and a cancel meanwhile has removed it for good. delivered = await TryDeliverAsync(record.Target, text, CancellationToken.None).ConfigureAwait(false); + Logger.LogInformation(delivered + ? "A2A: delivered the result of task {TaskId} of {Agent} to {Target}" + : "A2A: kept the result of task {TaskId} of {Agent} for {Target} until it can be delivered", + record.TaskId, record.Agent, record.Target); if (delivered) using (await LockAsync(watch.Workspace).ConfigureAwait(false)) PendingFile.Remove(watch.Workspace, record.WatchId); diff --git a/hosts/Volcanic.AgenticCore.Telegram/appsettings.json b/hosts/Volcanic.AgenticCore.Telegram/appsettings.json index dd07a43..2fdc347 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/appsettings.json +++ b/hosts/Volcanic.AgenticCore.Telegram/appsettings.json @@ -4,7 +4,8 @@ "Default": "Information", "Microsoft.Hosting.Lifetime": "Warning", "System.Net.Http": "Warning" - } + }, + "Console": { "FormatterName": "simple", "FormatterOptions": { "TimestampFormat": "yyyy-MM-dd HH:mm:ss 'UTC' ", "UseUtcTimestamp": true } } }, "Telegram": { "BotToken": "", diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs index b9029cf..434bfba 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AToolTests.cs @@ -242,6 +242,49 @@ public class A2AToolTests Assert.DoesNotContain(host.UserTexts(), t => t.Contains("[A2A result", StringComparison.Ordinal)); } + [Fact] + public async Task Status_SettledTask_EndsItsWatch_SoTheResultIsNotDeliveredTwice() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Answer by status").WithDelay(TimeSpan.FromMilliseconds(300))); + RemoteRewriter? rewriter = null; + // The watch's stream stays silent (a dead connection): only status can see the task finish. + await using var host = new A2AClientHost(remote, new A2AClientSettings { Enabled = true, WaitSeconds = 0 }, + inner => rewriter = new RemoteRewriter(inner) { SilentSubscribes = true }); + var session = await host.CreateSessionAsync(); + var (started, _) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "go" }); + var taskId = Field(started, "task"); + Assert.Single(host.Registry.List(A2AClientHost.Target)); + string status; + do + { + await Task.Delay(100); + (status, _) = await host.CallToolAsync(session, new { action = "status", agent = "remote", taskId }); + } while (!status.Contains("Answer by status")); + Assert.Empty(host.Registry.List(A2AClientHost.Target)); + Assert.Empty(PendingFile.Read(host.Workspace)); + Assert.DoesNotContain(host.UserTexts(), t => t.Contains("[A2A result", StringComparison.Ordinal)); + } + + [Fact] + public async Task Send_BusyContext_SaysWhichTaskRunsAndThatItIsStillWatched() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("first").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 = "long job" }); + var taskId = Field(started, "task"); + var contextId = Field(started, "context"); + var (refused, isError) = await host.CallToolAsync(session, new { action = "send", agent = "remote", message = "and another", contextId }); + Assert.True(isError, refused); + Assert.Contains($"still working on task {taskId} in context {contextId}", refused); + Assert.Contains("still watched", refused); + Assert.Contains("without contextId", refused); + var watch = Assert.Single(host.Registry.List(A2AClientHost.Target)); // the refusal leaves the running task's watch alone + Assert.Equal(taskId, watch.TaskId); + } + [Fact] public async Task Send_AbortedWhileWaiting_TaskWatched_AndResultArrives() { diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs index 3407f8a..3a525d6 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs @@ -368,6 +368,54 @@ public sealed class A2AWatchTests : IDisposable Assert.Equal(3, http.Subscribes); // three failed subscribes, then polling only } + [Fact] + public async Task Watch_SilentStream_SubscribesAgainEachSlice_AndDelivers() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("Finished in silence").WithDelay(TimeSpan.FromMilliseconds(700))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + var logger = new ListLogger(); + var http = new RemoteRewriter(remote.Handler) { SilentSubscribes = true }; + await using var registry = new A2AWatchRegistry(TimeProvider.System, http, logger) + { + SourceOverride = WatchSources.Subscribe, SubscriptionSlice = TimeSpan.FromMilliseconds(200), + ReconnectDelays = [TimeSpan.FromSeconds(30)], // a failure would stall the watch past the test's wait + }; + 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); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty, timeoutMs: 10_000); + Assert.Contains("Finished in silence", Assert.Single(delivery.Delivered).Text); + Assert.True(http.Subscribes >= 2, $"subscribes: {http.Subscribes}"); // each slice ended, the task read, then a new stream + Assert.Equal(WatchSources.Subscribe, record.Source); // pauses aren't failures: it never fell back to polling + Assert.DoesNotContain(logger.Lines, l => l.Contains("failed", StringComparison.Ordinal)); + Assert.Contains(logger.Lines, l => l.Contains($"watching task {task.Id} of remote for 1 by subscribe", StringComparison.Ordinal)); + Assert.Contains(logger.Lines, l => l.Contains($"task {task.Id} of remote is still", StringComparison.Ordinal) && l.Contains("subscribing again", StringComparison.Ordinal)); + await WaitUntilAsync(() => logger.Lines.Any(l => l.Contains($"delivered the result of task {task.Id} of remote to 1", StringComparison.Ordinal))); + } + + [Fact] + public async Task SharedHandler_ConnectsWithTcpKeepAlive() + { + using var listener = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0); + listener.Start(); + var port = ((System.Net.IPEndPoint)listener.LocalEndpoint).Port; + var accepted = listener.AcceptSocketAsync(); + await using var stream = Assert.IsType( + await A2AWatchRegistry.ConnectWithKeepAliveAsync(new System.Net.DnsEndPoint("localhost", port), CancellationToken.None)); + using var server = await accepted; + var socket = stream.Socket; + Assert.True(socket.Connected); + Assert.NotEqual(0, (int)socket.GetSocketOption(System.Net.Sockets.SocketOptionLevel.Socket, System.Net.Sockets.SocketOptionName.KeepAlive)!); + Assert.Equal(60, (int)socket.GetSocketOption(System.Net.Sockets.SocketOptionLevel.Tcp, System.Net.Sockets.SocketOptionName.TcpKeepAliveTime)!); + Assert.Equal(15, (int)socket.GetSocketOption(System.Net.Sockets.SocketOptionLevel.Tcp, System.Net.Sockets.SocketOptionName.TcpKeepAliveInterval)!); + using var handler = A2AWatchRegistry.CreateSharedHandler(); + Assert.NotNull(handler.ConnectCallback); + } + [Fact] public async Task Watch_ReusesTheConnectionBetweenChecks() { diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/RemoteRewriter.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/RemoteRewriter.cs index 8f37da3..1664ae1 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/RemoteRewriter.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/Support/RemoteRewriter.cs @@ -9,7 +9,8 @@ namespace Volcanic.AgenticCore.Extensions.Tests.A2A.Support; /// /// Between our client and the remote: counts subscribes and card fetches, can hide the card's streaming capability -/// (so the client polls), and can answer SendMessage itself with a plain message instead of a task. +/// (so the client polls), can answer SendMessage itself with a plain message instead of a task, and can answer +/// SubscribeToTask with a stream that never sends anything (a connection that died silently on the way). /// public sealed class RemoteRewriter(HttpMessageHandler inner) : DelegatingHandler(inner) { @@ -21,6 +22,9 @@ public sealed class RemoteRewriter(HttpMessageHandler inner) : DelegatingHandler /// When set, SendMessage is answered with a message carrying this text (the remote never sees it). public string? MessageReply { get; init; } + /// When true, every subscribe gets an open stream that stays silent until the client gives up on it. + public volatile bool SilentSubscribes; + /// When true, every request fails as if the remote were down. public volatile bool Down; @@ -30,7 +34,16 @@ public sealed class RemoteRewriter(HttpMessageHandler inner) : DelegatingHandler { if (Down) throw new HttpRequestException("remote is down"); string? body = request.Content is null ? null : await request.Content.ReadAsStringAsync(cancellationToken); - if (body is not null && body.Contains("\"SubscribeToTask\"", StringComparison.Ordinal)) Interlocked.Increment(ref _subscribes); + if (body is not null && body.Contains("\"SubscribeToTask\"", StringComparison.Ordinal)) + { + Interlocked.Increment(ref _subscribes); + if (SilentSubscribes) + { + var silent = new StreamContent(new SilentStream()); + silent.Headers.ContentType = new System.Net.Http.Headers.MediaTypeHeaderValue("text/event-stream"); + return new HttpResponseMessage(HttpStatusCode.OK) { Content = silent }; + } + } if (MessageReply is not null && body is not null && body.Contains("\"SendMessage\"", StringComparison.Ordinal)) { var id = JsonNode.Parse(body)!["id"]?.DeepClone(); @@ -57,3 +70,25 @@ public sealed class RemoteRewriter(HttpMessageHandler inner) : DelegatingHandler return response; } } + +/// A response body that never yields a byte; a read ends only when the reader cancels it. +internal sealed class SilentStream : Stream +{ + public override bool CanRead => true; + public override bool CanSeek => false; + public override bool CanWrite => false; + public override long Length => throw new NotSupportedException(); + public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + public override void Flush() { } + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException("async reads only"); + public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + await Task.Delay(Timeout.Infinite, cancellationToken); + return 0; + } + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) => + ReadAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); +}