From 372c600503dfac58d810db3d7b5a383b711c5137 Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Wed, 7 Oct 2026 14:02:49 +0000 Subject: [PATCH] fix(telegram): clean exit on A2A bind failure, log listen URLs, /a2a polish --- .../A2A/TelegramA2ASetup.cs | 15 +++- .../Chats/ChatCommands.cs | 20 ++++- .../Volcanic.AgenticCore.Telegram/Program.cs | 14 +++- .../Agent/A2AAgentHandler.cs | 45 +++++++---- .../A2AHostTests.cs | 79 +++++++++++++++++++ .../Support/ChatHost.cs | 6 +- 6 files changed, 155 insertions(+), 24 deletions(-) diff --git a/hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2ASetup.cs b/hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2ASetup.cs index 542dc49..5050342 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2ASetup.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2ASetup.cs @@ -1,4 +1,6 @@ using System.Globalization; +using System.Net.Sockets; +using Microsoft.AspNetCore.Connections; using Microsoft.AspNetCore.Hosting.Server; using Microsoft.AspNetCore.Routing; using Microsoft.Extensions.DependencyInjection; @@ -18,11 +20,15 @@ public static class TelegramA2ASetup options.AllowedUserIds.Select(id => new A2AWorkspace( ChatRegistry.WorkspaceFor(options, id), ChatRegistry.ScopeFor(id), id.ToString(CultureInfo.InvariantCulture))); + /// + /// Registers IOptions<A2AHostOptions> from (the one instance the server and + /// /a2a read). Off: replaces Kestrel and nothing else is registered. + /// public static void Register(IServiceCollection services, TelegramOptions options, A2AHostOptions a2a) { - services.AddSingleton(a2a); if (!a2a.Enabled) { + services.AddOptions(); // Enabled = false services.AddSingleton(); return; } @@ -52,4 +58,11 @@ public static class TelegramA2ASetup public static string Describe(A2AHostOptions a2a) => a2a.Enabled ? $"A2A: on at {a2a.PublicUrl ?? "(request host)"}/a2a, callers in each chat's .volcanic/a2a/callers.json" : "A2A: off"; + + /// + /// Kestrel could not open its listener: a port in use surfaces as an IOException wrapping + /// ; other bind errors (no permission, unknown address) as a raw SocketException. + /// + public static bool IsBindFailure(Exception ex) => + ex is IOException { InnerException: AddressInUseException } or SocketException; } diff --git a/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs b/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs index 318f35a..e66f141 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs @@ -1,5 +1,6 @@ using System.Globalization; using System.Text; +using Microsoft.Extensions.Options; using Telegram.Bot.Types; using Volcanic.AgenticCore.A2A.Server; using Volcanic.AgenticCore.A2A.Server.Callers; @@ -15,7 +16,7 @@ namespace Volcanic.AgenticCore.Telegram.Chats; /// The bot's own slash commands. Anything else starting with / goes to the session (extension commands). public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zones = null, JobScheduler? scheduler = null, TriggerServices? triggers = null, - A2AHostOptions? a2a = null) + IOptions? a2a = null, TimeProvider? time = null) { public static readonly IReadOnlyList Menu = [ @@ -227,17 +228,28 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon private string A2AStatus(ChatContext chat) { - if (a2a is not { Enabled: true }) return "A2A is off. Set A2A__Enabled=true (and A2A__PublicUrl) to let other agents talk to this assistant."; + if (a2a?.Value is not { Enabled: true }) return "A2A is off. Set A2A__Enabled=true (and A2A__PublicUrl) to let other agents talk to this assistant."; var workspace = chat.Session.Cwd; var callers = CallersFile.Read(workspace, Microsoft.Extensions.Logging.Abstractions.NullLogger.Instance); var lines = new List { "**A2A callers** (.volcanic/a2a/callers.json)" }; - lines.AddRange(callers.Count == 0 ? ["none"] : callers.Select(c => $"• {c.Name} ({c.Mode}) key …{(c.Key.Length > 4 ? c.Key[^4..] : "")}")); + lines.AddRange(callers.Count == 0 ? ["none"] : callers.Select(c => $"• {c.Name} ({c.Mode}) key …{(c.Key.Length >= 12 ? c.Key[^4..] : "")}")); var tasks = A2ATaskFiles.Recent(workspace, 10); lines.Add("**Recent A2A tasks**"); - lines.AddRange(tasks.Count == 0 ? ["none"] : tasks.Select(t => $"• {t.Caller}: {t.State} ({t.TaskId[..Math.Min(8, t.TaskId.Length)]})")); + lines.AddRange(tasks.Count == 0 ? ["none"] : tasks.Select(t => $"• {t.Caller}: {t.State}{Age(t.Updated)} ({t.TaskId[..Math.Min(8, t.TaskId.Length)]})")); return string.Join('\n', lines); } + /// ", 5m ago" from a task's last status time; empty without one. + private string Age(DateTimeOffset? updated) + { + if (updated is not { } at) return ""; + var age = (time ?? TimeProvider.System).GetUtcNow() - at; + return age.TotalMinutes < 1 ? ", just now" + : age.TotalHours < 1 ? $", {(int)age.TotalMinutes}m ago" + : age.TotalDays < 1 ? $", {(int)age.TotalHours}h ago" + : $", {(int)age.TotalDays}d ago"; + } + private async Task CancelJobAsync(ChatContext chat, string arg, CancellationToken cancellationToken) { if (scheduler is not { IsStarted: true }) return SchedulerSetup.DisabledReply; diff --git a/hosts/Volcanic.AgenticCore.Telegram/Program.cs b/hosts/Volcanic.AgenticCore.Telegram/Program.cs index b1e5c78..1f52628 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Program.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Program.cs @@ -129,6 +129,18 @@ logger.LogInformation("Default time zone: {Zone}", TimeZoneIds.Name(options.Conf logger.LogInformation("Allowed users: {Users}. Workspaces under {Root}", string.Join(", ", options.AllowedUserIds), options.WorkspaceRoot); if (options.AdminUserId is { } admin) logger.LogInformation("Admin (may /impersonate): {Admin}", admin); +// appsettings keeps Kestrel's own "Now listening" quiet (Microsoft.Hosting.Lifetime: Warning). +if (a2a.Enabled) + host.Lifetime.ApplicationStarted.Register(() => logger.LogInformation("A2A: listening on {Urls}", string.Join(", ", host.Urls))); + var polling = host.Services.GetRequiredService(); -await host.RunAsync(); // disposes the host +try +{ + await host.RunAsync(); // disposes the host +} +catch (Exception ex) when (a2a.Enabled && TelegramA2ASetup.IsBindFailure(ex)) +{ + logger.LogCritical("A2A: could not open the HTTP listener ({Error}). Free the port, set ASPNETCORE_URLS, or set A2A__Enabled=false.", ex.Message); + return 3; +} return polling.FatalError is null ? 0 : 1; diff --git a/src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs b/src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs index 8b61e8d..2968f19 100644 --- a/src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs +++ b/src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs @@ -13,8 +13,10 @@ namespace Volcanic.AgenticCore.A2A.Server.Agent; /// One task = one run of the caller context's session. Ends completed (final text as artifact response), /// input-required (the agent called request_input), failed (model error, hop limit, bad file, busy context, any /// unexpected exception) or canceled (the request token or a CancelTask). The task is registered as live in the pool -/// before its first event and stays live until its final status is written, so restart recovery never touches it. -/// A second run of a task whose run is unfinished (a racing continuation) throws before writing anything. +/// before its first event; its run is marked finished just before the final status is written (so a continuation of an +/// input-required task is accepted at once) and it stays live for the pool's grace period after that, so restart +/// recovery never touches it. A second run of a task whose run is unfinished (a racing continuation) throws before +/// writing anything. /// internal sealed class A2AAgentHandler( A2ACaller caller, WorkspaceTaskStore store, A2ASessionPool pool, IOptions options, ISecretRedactor redactor, A2AFileDownloader downloader, @@ -35,25 +37,34 @@ internal sealed class A2AAgentHandler( { var started = Stopwatch.GetTimestamp(); var updater = new TaskUpdater(eventQueue, context.TaskId, context.ContextId); - using var tracking = pool.Track(context.TaskId, out var run); + var tracking = pool.Track(context.TaskId, out var run); Result result; - using (var runCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, run.Cancel.Token)) + try { - var token = runCts.Token; - try + using (var runCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, run.Cancel.Token)) { - result = await RunAsync(context, eventQueue, updater, token).ConfigureAwait(false); + var token = runCts.Token; + try + { + result = await RunAsync(context, eventQueue, updater, token).ConfigureAwait(false); + } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + result = Result.Canceled; + } + catch (Exception ex) + { + logger.LogWarning(ex, "A2A: task {TaskId} of {Caller} failed", context.TaskId, caller.Name); + result = Result.Failed("The run failed: " + redactor.Redact(ex.Message), "failed (error)"); + } + if (token.IsCancellationRequested) result = Result.Canceled; } - catch (OperationCanceledException) when (token.IsCancellationRequested) - { - result = Result.Canceled; - } - catch (Exception ex) - { - logger.LogWarning(ex, "A2A: task {TaskId} of {Caller} failed", context.TaskId, caller.Name); - result = Result.Failed("The run failed: " + redactor.Redact(ex.Message), "failed (error)"); - } - if (token.IsCancellationRequested) result = Result.Canceled; + } + finally + { + // The run is over before its final status is written: a continuation sent the moment InputRequired is + // stored must not be refused as "already running". The grace period still keeps recovery away. + tracking.Dispose(); } var label = await WriteFinalAsync(context, updater, result).ConfigureAwait(false); Log(context, label, started); diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/A2AHostTests.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/A2AHostTests.cs index c666021..2c27ca0 100644 --- a/tests/Volcanic.AgenticCore.Telegram.Tests/A2AHostTests.cs +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/A2AHostTests.cs @@ -1,8 +1,15 @@ using Microsoft.AspNetCore.Hosting.Server; using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using Microsoft.Extensions.Time.Testing; using Volcanic.AgenticCore.A2A.Server; +using Volcanic.AgenticCore.A2A.Server.Agent; using Volcanic.AgenticCore.A2A.Server.Callers; +using Volcanic.AgenticCore.A2A.Server.Hosting; +using Volcanic.AgenticCore.A2A.Server.Tasks; +using Volcanic.AgenticCore.Telegram.Chats; using Volcanic.AgenticCore.Telegram.A2A; using Volcanic.AgenticCore.Telegram.Tests.Support; @@ -72,4 +79,76 @@ public class A2AHostTests await host.SettleAsync(); Assert.Contains("A2A is off", host.LastReply()); } + + [Fact] + public async Task A2aCommand_ShortKeyFullyMasked_TaskAge() + { + var time = new FakeTimeProvider(new DateTimeOffset(2026, 10, 7, 12, 0, 0, TimeSpan.Zero)); + await using var host = new ChatHost(time: time, a2a: new A2AHostOptions { Enabled = true }); + WriteCallers(host.WorkspaceOf(), """[ { "name": "travel-bot", "key": "short-1234" } ]"""); + Directory.CreateDirectory(A2ATaskFiles.Dir(host.WorkspaceOf())); + A2ATaskFiles.Write(A2ATaskFiles.PathFor(host.WorkspaceOf(), "task0001abc"), new A2ATaskFile + { + Caller = "travel-bot", + Task = new global::A2A.AgentTask + { + Id = "task0001abc", ContextId = "c1", + Status = new global::A2A.TaskStatus { State = global::A2A.TaskState.Completed, Timestamp = time.GetUtcNow() - TimeSpan.FromMinutes(5) }, + }, + }); + await host.SendAsync("/a2a"); + time.Advance(TimeSpan.FromSeconds(1)); // past the input window + await host.SettleAsync(); + var reply = host.LastReply(); + Assert.Contains("travel-bot (isolated) key …\n", reply + "\n"); + Assert.DoesNotContain("1234", reply); + Assert.Contains("travel-bot: Completed, 5m ago (task0001)", reply); + } + + /// The bot's A2A registrations on top of the chat side: no construction cycle, and the shared options. + private static ChatHost EnabledHost() => new((h, s) => + { + s.AddLogging(); + s.AddSingleton(_ => h.Chats); + s.AddSingleton(_ => h.Settings); + TelegramA2ASetup.Register(s, h.Options, new A2AHostOptions { Enabled = true, MaxHops = 5 }); + }); + + [Fact] + public async Task A2aEnabled_ServicesResolve() + { + await using var host = EnabledHost(); + Assert.NotNull(host.Services.GetRequiredService()); + Assert.NotNull(host.Services.GetRequiredService()); + Assert.NotNull(host.Services.GetRequiredService()); + Assert.IsType(host.Services.GetRequiredService()); + Assert.IsType(host.Services.GetRequiredService()); + var options = host.Services.GetRequiredService>().Value; + Assert.True(options.Enabled); + Assert.Equal(5, options.MaxHops); + Assert.Equal(host.Options.Model, options.Model); + Assert.NotNull(options.TimeZoneFor); + } + + [Fact] + public async Task A2aSession_GetsBotModelAndOwnersZone() + { + await using var host = EnabledHost(); + var tokyo = TimeZoneInfo.FindSystemTimeZoneById("Asia/Tokyo"); + await host.Zones.SetAsync(Updates.AllowedUser, tokyo, CancellationToken.None); + var pool = host.Services.GetRequiredService(); + var caller = new A2ACaller("travel-bot", "isolated", host.WorkspaceOf(), $"telegram-{Updates.AllowedUser}", Updates.AllowedUser.ToString()); + using var lease = await pool.AcquireAsync(caller, "ctx1", "task1", CancellationToken.None); + Assert.Equal(host.Options.Model, lease.Session.Model?.Key); + Assert.Equal(tokyo.Id, lease.Session.TimeZone.Id); + } + + [Fact] + public void BindFailure_Classified() + { + Assert.True(TelegramA2ASetup.IsBindFailure(new IOException("Failed to bind", new Microsoft.AspNetCore.Connections.AddressInUseException("in use")))); + Assert.True(TelegramA2ASetup.IsBindFailure(new System.Net.Sockets.SocketException((int)System.Net.Sockets.SocketError.AccessDenied))); + Assert.False(TelegramA2ASetup.IsBindFailure(new IOException("disk full"))); + Assert.False(TelegramA2ASetup.IsBindFailure(new InvalidOperationException())); + } } diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs index 762397b..2a3e93c 100644 --- a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs @@ -28,6 +28,9 @@ public sealed class ChatHost : IAsyncDisposable public FakeBotClient Bot { get; } = new(); public FauxProvider Faux { get; } = new(); public TelegramOptions Options { get; } + + /// The services built from configure and the core (sessions, plus whatever the test added). + public IServiceProvider Services => _services; public ChatRegistry Chats { get; } public UpdateRouter Router { get; } private readonly ServiceProvider _services; @@ -116,7 +119,8 @@ public sealed class ChatHost : IAsyncDisposable if (Scheduler is not null) Delivery = new TelegramJobDelivery(Chats, Settings, Options, clock, NullLogger.Instance); var media = new InboundMedia(sender, Options, time ?? TimeProvider.System, NullLogger.Instance, speech); Composer = new InboundComposer(sender, media, NullLogger.Instance); - Router = new UpdateRouter(Options, Chats, new ChatCommands(sender, Zones, Scheduler, Triggers, a2a), sender, Composer, NullLogger.Instance, Impersonation, Feed); + Router = new UpdateRouter(Options, Chats, new ChatCommands(sender, Zones, Scheduler, Triggers, + a2a is null ? null : Microsoft.Extensions.Options.Options.Create(a2a), clock), sender, Composer, NullLogger.Instance, Impersonation, Feed); } private readonly TriggerDispatcher? _dispatcher;