From cb877f697a28dd8f51ec4a0037c1d072051c36b5 Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Fri, 16 Oct 2026 14:46:59 +0000 Subject: [PATCH] fix(a2a-client): poll watches crashed on null delays; lost tasks and crashed watches end with a notice - A2AWatchRegistry.Shared was initialised before DefaultPollDelays (static fields run in source order), so the bot's registry had null poll delays and every poll watch crashed on its first wait, then sat in the registry doing nothing until a restart. The delays now come first; a test reads them through Shared. - TaskNotFound from the remote ends the watch at once with a "(lost)" notice instead of retrying for a day. - A crashed watch loop restarts by polling after a pause; the third crash in a row ends it with a "(stopped)" notice. --- docs/a2a.md | 5 +- .../specs/2026-10-07-a2a-client-design.md | 2 + .../A2AWatchRegistry.cs | 75 +++++++++++++++++-- .../A2A/A2AWatchTests.cs | 63 ++++++++++++++++ 4 files changed, 136 insertions(+), 9 deletions(-) diff --git a/docs/a2a.md b/docs/a2a.md index 267075f..2d9f5c4 100644 --- a/docs/a2a.md +++ b/docs/a2a.md @@ -346,7 +346,10 @@ Each check reads `agents.json` again: an agent removed from it ends the watch wi 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`. The shared handler turns on TCP keepalive (probes after 60 s of silence, every 15 s, 4 tries) and +`action=status`. A remote that answers `TaskNotFound` has forgotten the task (typically it restarted): the watch ends +at once with a `(lost)` notice suggesting to send the request again. A crash of a watch's loop (a bug, not a remote +failure) restarts it by polling after 1 s, 5 s, 15 s; the third crash in a row ends it with a `(stopped)` notice +suggesting `action=status`, so no watch stays listed while doing nothing. 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 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 5a6eb07..5464f62 100644 --- a/docs/superpowers/specs/2026-10-07-a2a-client-design.md +++ b/docs/superpowers/specs/2026-10-07-a2a-client-design.md @@ -136,6 +136,8 @@ Watch record, persisted in `/.volcanic/a2a/pending.json`: token, body)`, which checks the token (constant-time) and returns false for unknown watches or bad tokens (→ 404). - **End**: on a terminal state or `input-required`/`auth-required`, the registry builds a result text and hands it to `IA2AResultDelivery`, then removes the watch (and deletes the push config remotely, best effort). +- **Lost / crashed**: `TaskNotFound` from the remote ends the watch at once with a `(lost)` notice. A crashed watch loop + restarts by polling; after 3 crashes in a row it ends with a `(stopped)` notice. - **Timeout**: after `WatchTimeout`, deliver "travel-bot gave no result for task within 24h; you can check it with action=status" and remove the watch. - **Restore**: at startup the host calls `A2AWatchRegistry.Shared.RestoreAsync(workspaces)`, which reads every diff --git a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs index ad75799..b0cc4f5 100644 --- a/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs +++ b/extensions/Volcanic.AgenticCore.Extensions.A2A/A2AWatchRegistry.cs @@ -27,6 +27,11 @@ namespace Volcanic.AgenticCore.Extensions.A2A; /// public sealed class A2AWatchRegistry : IAsyncDisposable { + // Static fields initialise in source order: everything an instance reads (DefaultPollDelays) must come before Shared, + // or Shared gets nulls (it did: every poll watch of the bot crashed on its first wait). + public static readonly TimeSpan[] DefaultPollDelays = + [TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(30), TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(2), TimeSpan.FromMinutes(5)]; + public static A2AWatchRegistry Shared { get; } = new(TimeProvider.System, CreateSharedHandler()); /// @@ -61,14 +66,14 @@ public sealed class A2AWatchRegistry : IAsyncDisposable } } - public static readonly TimeSpan[] DefaultPollDelays = - [TimeSpan.FromSeconds(15), TimeSpan.FromSeconds(30), TimeSpan.FromMinutes(1), TimeSpan.FromMinutes(2), TimeSpan.FromMinutes(5)]; - private static readonly TimeSpan DeleteTimeout = TimeSpan.FromSeconds(10); /// After this many failed checks in a row, a subscribe watch polls instead. private const int SubscribeFailuresBeforePoll = 3; + /// A watch whose loop crashed this many times in a row ends with a (stopped) notice. + private const int CrashesBeforeGivingUp = 3; + private readonly HttpMessageHandler _handler; /// behind , for every A2A call (not for downloads). @@ -337,7 +342,57 @@ public sealed class A2AWatchRegistry : IAsyncDisposable } } + /// + /// Runs a watch until it finishes or is stopped. A crash of its loop (a bug, never a remote failure: those are retried + /// inside) restarts the loop after a pause, by polling; after crashes in a row the + /// watch ends with a (stopped) notice, so it never sits in the registry doing nothing. + /// private async Task RunAsync(Watch watch, A2AClientSettings settings) + { + var ct = watch.Stop.Token; + try + { + for (var crashes = 1; ; crashes++) + { + try + { + await WatchAsync(watch, settings).ConfigureAwait(false); + return; + } + catch (Exception ex) when (ex is not OutOfMemoryException && !ct.IsCancellationRequested) + { + var record = watch.Record; + Logger.LogWarning(ex, "A2A: watch {WatchId} of task {TaskId} of {Agent} crashed ({Count} of {Max})", + record.WatchId, record.TaskId, record.Agent, crashes, CrashesBeforeGivingUp); + if (crashes >= CrashesBeforeGivingUp) + { + await FinishAsync(watch, $"[A2A result from \"{record.Agent}\", task {record.TaskId} (stopped)]\n" + + $"Watching this task failed ({ex.GetType().Name}), so it is no longer watched. " + + $"Check it with action=status, agent={record.Agent}, taskId={record.TaskId}.", null).ConfigureAwait(false); + return; + } + if (record.Source != WatchSources.Poll) await SetSourceAsync(watch, WatchSources.Poll).ConfigureAwait(false); + await WaitAsync(watch, DelayAt(ReconnectDelays, crashes - 1), ct).ConfigureAwait(false); + } + } + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { + } + catch (Exception ex) when (ex is not OutOfMemoryException) + { + // Even the (stopped) notice failed: out of the live watches; its record stays for the next restore. + Logger.LogWarning(ex, "A2A: watch {WatchId} crashed", watch.Record.WatchId); + _watches.TryRemove(new(watch.Record.WatchId, watch)); + } + finally + { + watch.DisposeStop(); + } + } + + /// The watch loop: returns when the watch finished or was stopped, throws only on a bug. + private async Task WatchAsync(Watch watch, A2AClientSettings settings) { var ct = watch.Stop.Token; RemoteAgent? agent = null; @@ -411,6 +466,15 @@ public sealed class A2AWatchRegistry : IAsyncDisposable { return; } + catch (A2AException ex) when (ex.ErrorCode == A2AErrorCode.TaskNotFound) + { + // Definitive: the remote forgot the task (typically it restarted), so no check will ever find it. + Logger.LogInformation("A2A: task {TaskId} of {Agent} is unknown to the agent; the watch ends", record.TaskId, record.Agent); + await FinishAsync(watch, $"[A2A result from \"{record.Agent}\", task {record.TaskId} (lost)]\n" + + $"\"{record.Agent}\" no longer knows this task (it was probably restarted), so its result can't be fetched " + + "and the task is no longer watched. Send the request again with action=send if it is still needed.", agent).ConfigureAwait(false); + return; + } catch (Exception ex) when (ex is not OutOfMemoryException) { failures++; @@ -431,14 +495,9 @@ public sealed class A2AWatchRegistry : IAsyncDisposable catch (OperationCanceledException) when (ct.IsCancellationRequested) { } - catch (Exception ex) when (ex is not OutOfMemoryException) - { - Logger.LogWarning(ex, "A2A: watch {WatchId} crashed", watch.Record.WatchId); - } finally { agent?.Dispose(); - watch.DisposeStop(); } } diff --git a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs index 3a525d6..b8d3740 100644 --- a/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs +++ b/tests/Volcanic.AgenticCore.Extensions.Tests/A2A/A2AWatchTests.cs @@ -649,6 +649,69 @@ public sealed class A2AWatchTests : IDisposable Assert.Contains(PendingFile.Read(_ws), r => r.WatchId == watched.WatchId); } + [Fact] + public void Shared_HasItsDelaysAndSlice() + { + // Static fields initialise in source order: Shared once read DefaultPollDelays before it was set (null). + var shared = A2AWatchRegistry.Shared; + Assert.Same(A2AWatchRegistry.DefaultPollDelays, shared.PollDelays); + Assert.NotEmpty(shared.ReconnectDelays); + Assert.Equal(TimeSpan.FromSeconds(15), A2AWatchRegistry.DelayAt(shared.PollDelays, 0)); + Assert.Equal(TimeSpan.FromMinutes(5), shared.SubscriptionSlice); + } + + [Theory] + [InlineData(WatchSources.Subscribe)] + [InlineData(WatchSources.Poll)] + public async Task Watch_TaskUnknownToTheRemote_EndsAtOnceAsLost(string source) + { + await using var remote = await RemoteAgentServer.StartAsync(); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler) + { + SourceOverride = source, PollDelays = [TimeSpan.FromSeconds(30)], ReconnectDelays = [TimeSpan.FromSeconds(30)], // no retry fits the wait + }; + registry.UseDelivery(delivery); + var settings = new A2AClientSettings(); + using var agent = await registry.Connector(settings).ConnectAsync(AgentsFile.Find(_ws, "remote")!, CancellationToken.None); + var forgotten = new AgentTask { Id = "forgotten-task", ContextId = "ctx-1", Status = new global::A2A.TaskStatus { State = TaskState.Working } }; + await registry.StartAsync("1", _ws, agent, forgotten, settings, CancellationToken.None); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty, timeoutMs: 10_000); + var text = Assert.Single(delivery.Delivered).Text; + Assert.StartsWith("[A2A result from \"remote\", task forgotten-task (lost)]", text); + Assert.Contains("no longer knows this task", text); + await WaitUntilAsync(() => PendingFile.Read(_ws).Count == 0); + Assert.Empty(registry.List("1")); + } + + [Fact] + public async Task Watch_CrashingLoop_RestartsThenEndsAsStopped_NeverLeftHanging() + { + await using var remote = await RemoteAgentServer.StartAsync(); + remote.Faux.Enqueue(FauxTurn.Say("never seen").WithDelay(TimeSpan.FromSeconds(10))); + WriteAgents(); + var delivery = new RecordingDelivery(_ws); + var logger = new ListLogger(); + // No poll delays at all: the first wait of every loop throws, like the null delays of the shared registry did. + await using var registry = new A2AWatchRegistry(TimeProvider.System, remote.Handler, logger) + { + SourceOverride = WatchSources.Poll, PollDelays = [], ReconnectDelays = [TimeSpan.FromMilliseconds(20)], + }; + 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); + await registry.StartAsync("1", _ws, agent, task, settings, CancellationToken.None); + await WaitUntilAsync(() => !delivery.Delivered.IsEmpty); + var text = Assert.Single(delivery.Delivered).Text; + Assert.StartsWith($"[A2A result from \"remote\", task {task.Id} (stopped)]", text); + Assert.Contains("action=status", text); + Assert.Equal(3, logger.Lines.Count(l => l.Contains("crashed (", StringComparison.Ordinal))); + await WaitUntilAsync(() => PendingFile.Read(_ws).Count == 0); + Assert.Empty(registry.List("1")); + } + [Fact] public void PollDelays_BackOffToFiveMinutes() {