admin/VolcanicAgenticCore
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(telegram): clean exit on A2A bind failure, log listen URLs, /a2a polish

Browse Source
This commit is contained in:
EugeneTes
2026-10-07 14:02:49 +00:00
parent a3aee939d4
commit 372c600503
6 changed files with 155 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2ASetup.cs
View File

@@ -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)));
/// <summary>
/// Registers <c>IOptions&lt;A2AHostOptions&gt;</c> from <paramref name="a2a"/> (the one instance the server and
/// <c>/a2a</c> read). Off: <see cref="NoHttpServer"/> replaces Kestrel and nothing else is registered.
/// </summary>
public static void Register(IServiceCollection services, TelegramOptions options, A2AHostOptions a2a)
{
services.AddSingleton(a2a);
if (!a2a.Enabled)
{
services.AddOptions<A2AHostOptions>(); // Enabled = false
services.AddSingleton<IServer, NoHttpServer>();
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";
/// <summary>
/// Kestrel could not open its listener: a port in use surfaces as an IOException wrapping
/// <see cref="AddressInUseException"/>; other bind errors (no permission, unknown address) as a raw SocketException.
/// </summary>
public static bool IsBindFailure(Exception ex) =>
ex is IOException { InnerException: AddressInUseException } or SocketException;
}

20
hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs
View File

@@ -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;
/// <summary>The bot's own slash commands. Anything else starting with <c>/</c> goes to the session (extension commands).</summary>
public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zones = null, JobScheduler? scheduler = null, TriggerServices? triggers = null,
A2AHostOptions? a2a = null)
IOptions<A2AHostOptions>? a2a = null, TimeProvider? time = null)
{
public static readonly IReadOnlyList<BotCommand> 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<string> { "**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);
}
/// <summary>", 5m ago" from a task's last status time; empty without one.</summary>
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<string> CancelJobAsync(ChatContext chat, string arg, CancellationToken cancellationToken)
{
if (scheduler is not { IsStarted: true }) return SchedulerSetup.DisabledReply;

14
hosts/Volcanic.AgenticCore.Telegram/Program.cs
View File

@@ -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<PollingService>();
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;

17
src/Volcanic.AgenticCore.A2A.Server/Agent/A2AAgentHandler.cs
View File

@@ -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 <c>response</c>),
/// input-required (the agent called <c>request_input</c>), 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.
/// </summary>
internal sealed class A2AAgentHandler(
A2ACaller caller, WorkspaceTaskStore store, A2ASessionPool pool, IOptions<A2AHostOptions> options, ISecretRedactor redactor, A2AFileDownloader downloader,
@@ -35,8 +37,10 @@ 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;
try
{
using (var runCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken, run.Cancel.Token))
{
var token = runCts.Token;
@@ -55,6 +59,13 @@ internal sealed class A2AAgentHandler(
}
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);
}

79
tests/Volcanic.AgenticCore.Telegram.Tests/A2AHostTests.cs
View File

@@ -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);
}
/// <summary>The bot's A2A registrations on top of the chat side: no construction cycle, and the shared options.</summary>
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<A2ARequestRouter>());
Assert.NotNull(host.Services.GetRequiredService<A2AServerCache>());
Assert.NotNull(host.Services.GetRequiredService<A2ASessionPool>());
Assert.IsType<TelegramOwnerNotifier>(host.Services.GetRequiredService<IA2AOwnerNotifier>());
Assert.IsType<WorkspaceCallerDirectory>(host.Services.GetRequiredService<IA2ACallerDirectory>());
var options = host.Services.GetRequiredService<IOptions<A2AHostOptions>>().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<A2ASessionPool>();
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()));
}
}

6
tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs
View File

@@ -28,6 +28,9 @@ public sealed class ChatHost : IAsyncDisposable
public FakeBotClient Bot { get; } = new();
public FauxProvider Faux { get; } = new();
public TelegramOptions Options { get; }
/// <summary>The services built from <c>configure</c> and the core (sessions, plus whatever the test added).</summary>
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<TelegramJobDelivery>.Instance);
var media = new InboundMedia(sender, Options, time ?? TimeProvider.System, NullLogger<InboundMedia>.Instance, speech);
Composer = new InboundComposer(sender, media, NullLogger<InboundComposer>.Instance);
Router = new UpdateRouter(Options, Chats, new ChatCommands(sender, Zones, Scheduler, Triggers, a2a), sender, Composer, NullLogger<UpdateRouter>.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<UpdateRouter>.Instance, Impersonation, Feed);
}
private readonly TriggerDispatcher? _dispatcher;