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

feat(telegram): A2A client per chat behind A2AClient__Enabled, watch results as follow-ups, /a2a lists agents

Browse Source
This commit is contained in:
EugeneTes
2026-10-08 14:21:34 +00:00
parent adee225fcb
commit 4a20ff4635
12 changed files with 373 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
hosts/Volcanic.AgenticCore.Telegram/A2A/A2AWatchHostedService.cs Normal file
View File

@@ -0,0 +1,48 @@
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Volcanic.AgenticCore.Extensions.A2A;
namespace Volcanic.AgenticCore.Telegram.A2A;
/// <summary>
/// Points the watch registry at the chats and re-arms persisted watches in the background (restoring first delivers
/// results kept undelivered, which runs whole chat turns, so it must not hold up host startup); stops them on shutdown.
/// </summary>
public sealed class A2AWatchHostedService(A2AWatchRegistry registry, A2AClientSettings settings, TelegramA2AResultDelivery delivery,
ILogger<A2AWatchHostedService> logger) : IHostedService
{
private readonly CancellationTokenSource _stopping = new();
/// <summary>The background restore (tests await it).</summary>
internal Task Restore { get; private set; } = Task.CompletedTask;
public Task StartAsync(CancellationToken cancellationToken)
{
registry.Logger = logger;
registry.UseDelivery(delivery);
var token = _stopping.Token;
Restore = Task.Run(async () =>
{
try
{
var count = await registry.RestoreAsync(settings, token).ConfigureAwait(false);
logger.LogInformation("A2A client: {Count} watch(es) restored", count);
}
catch (OperationCanceledException) when (token.IsCancellationRequested)
{
}
catch (Exception ex)
{
logger.LogError(ex, "A2A client: restoring watches failed");
}
}, CancellationToken.None);
return Task.CompletedTask;
}
/// <summary>Cancels a restore still running (without waiting for a chat turn it started) and stops every watch.</summary>
public async Task StopAsync(CancellationToken cancellationToken)
{
await _stopping.CancelAsync().ConfigureAwait(false);
await registry.StopAsync().ConfigureAwait(false);
}
}

31
hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AClientSetup.cs Normal file
View File

@@ -0,0 +1,31 @@
using Microsoft.Extensions.DependencyInjection;
using Volcanic.AgenticCore.A2A.Server;
using Volcanic.AgenticCore.Extensions.A2A;
namespace Volcanic.AgenticCore.Telegram.A2A;
/// <summary>
/// The A2A client (off unless <c>A2AClient__Enabled=true</c>, independent of the server's <c>A2A__Enabled</c>). Only the
/// per-chat sessions get <see cref="A2AClientExtension"/> (<see cref="Chats.ChatRegistry"/>), never A2A server sessions.
/// Call before <see cref="TelegramA2ASetup.Register"/>, which copies the server options: when the server is on, its push
/// receiver is routed into the watch registry and its public URL becomes the push base URL.
/// </summary>
public static class TelegramA2AClientSetup
{
public static void Register(IServiceCollection services, A2AClientSettings client, A2AHostOptions server, A2AWatchRegistry registry)
{
// Only from the server: without its push receiver nobody would answer a push, so a configured value is ignored.
client.PushBaseUrl = server.Enabled && client.Enabled ? server.PublicUrl : null;
services.AddSingleton(client);
if (!client.Enabled) return;
if (server.Enabled) server.OnPush = registry.OnPushAsync;
services.AddSingleton(registry);
services.AddSingleton<TelegramA2AResultDelivery>();
services.AddHostedService<A2AWatchHostedService>();
}
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";
}

45
hosts/Volcanic.AgenticCore.Telegram/A2A/TelegramA2AResultDelivery.cs Normal file
View File

@@ -0,0 +1,45 @@
using System.Globalization;
using Microsoft.Extensions.Logging;
using Volcanic.AgenticCore.Extensions.A2A;
using Volcanic.AgenticCore.Telegram.Chats;
namespace Volcanic.AgenticCore.Telegram.A2A;
/// <summary>
/// A2A watch results as follow-up prompts in the chat (target = chat id, invariant culture), like scheduled prompts:
/// never through the inbox. The chat's session is created on demand; <see cref="ChatRegistry.GetAsync"/> returns it only
/// once its renderer is subscribed, so the answer is always rendered. <see cref="WorkspaceFor"/> never creates a session.
/// </summary>
public sealed class TelegramA2AResultDelivery(ChatRegistry chats, ChatSettingsStore settings, TelegramOptions options, TimeProvider time,
ILogger<TelegramA2AResultDelivery> logger) : IA2AResultDelivery
{
private static bool TryChatIdOf(string target, out long chatId) =>
long.TryParse(target, NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture, out chatId);
/// <summary>False (kept for later) for a target that isn't an allowed chat or whose session can't be opened.</summary>
public async Task<bool> DeliverAsync(string target, string text, CancellationToken cancellationToken)
{
if (!TryChatIdOf(target, 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 (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
throw;
}
catch (Exception ex)
{
// The prompt was handed to the session: delivered. The user sees why nothing came of it.
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) => TryChatIdOf(target, out var chatId) ? ChatRegistry.WorkspaceFor(options, chatId) : null;
/// <summary>Allowed chats only: a chat that left the allow-list keeps its pending.json untouched.</summary>
public IEnumerable<string> KnownTargets() => options.AllowedUserIds.Select(id => id.ToString(CultureInfo.InvariantCulture)).ToList();
}

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

@@ -6,6 +6,7 @@ using Volcanic.AgenticCore.A2A.Server;
using Volcanic.AgenticCore.A2A.Server.Callers;
using Volcanic.AgenticCore.A2A.Server.Tasks;
using Volcanic.AgenticCore.Abstractions.Providers;
using Volcanic.AgenticCore.Extensions.A2A;
using Volcanic.AgenticCore.Extensions.Scheduler;
using Volcanic.AgenticCore.Hosting;
using Volcanic.AgenticCore.Telegram.Rendering;
@@ -16,7 +17,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,
IOptions<A2AHostOptions>? a2a = null, TimeProvider? time = null)
IOptions<A2AHostOptions>? a2a = null, TimeProvider? time = null, A2AClientSettings? a2aClient = null, A2AWatchRegistry? a2aWatches = null)
{
public static readonly IReadOnlyList<BotCommand> Menu =
[
@@ -30,7 +31,7 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon
new() { Command = "timezone", Description = "Show or set this chat's time zone" },
.. SchedulerSetup.Menu,
TriggersSetup.MenuCommand,
new() { Command = "a2a", Description = "A2A callers and tasks of this chat" },
new() { Command = "a2a", Description = "A2A callers, remote agents and tasks of this chat" },
new() { Command = "help", Description = "What this bot does" },
];
@@ -156,7 +157,7 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon
/compact – summarise the conversation to free context
/thinking [level] – show or set the thinking level
/timezone [IANA name|reset] – this chat's time zone
""" + SchedulerSetup.HelpLines + "\n/a2a – A2A callers and recent tasks of this chat" + TriggersSetup.HelpLines + GoogleCalendarSetup.HelpLines(chat.Session.Commands);
""" + SchedulerSetup.HelpLines + "\n/a2a – A2A callers, remote agents and recent tasks of this chat" + TriggersSetup.HelpLines + GoogleCalendarSetup.HelpLines(chat.Session.Commands);
private string Status(ChatContext chat)
{
@@ -228,17 +229,43 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon
private string A2AStatus(ChatContext chat)
{
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 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 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 >= 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}{Age(t.Updated)} ({t.TaskId[..Math.Min(8, t.TaskId.Length)]})"));
var lines = new List<string>();
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)} ({ShortId(t.TaskId)})"));
}
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))}"));
lines.AddRange(agents.Problems.Select(p => $"• skipped: {p}"));
var watches = a2aWatches?.List(TelegramJobDelivery.TargetOf(chat.ChatId)) ?? [];
lines.Add("**Watches** (remote tasks still running)");
lines.AddRange(watches.Count == 0 ? ["none"] : watches.Select(w => $"• {w.Agent}: task {ShortId(w.TaskId)} by {w.Source}{Age(w.StartedAt)}"));
}
else lines.Add("A2A client: off (`A2AClient__Enabled`).");
return string.Join('\n', lines);
}
/// <summary>The last four characters of a key of 12 or more, else nothing: "…abcd" or "…".</summary>
private static string Mask(string key) => key.Length >= 12 ? "…" + key[^4..] : "…";
private static string ShortId(string id) => id[..Math.Min(8, id.Length)];
/// <summary>", 5m ago" from a task's last status time; empty without one.</summary>
private string Age(DateTimeOffset? updated)
{

20
hosts/Volcanic.AgenticCore.Telegram/Chats/ChatFollowUps.cs Normal file
View File

@@ -0,0 +1,20 @@
using Volcanic.AgenticCore.Abstractions.Extensions.Events;
using Volcanic.AgenticCore.Telegram.Inbound;
namespace Volcanic.AgenticCore.Telegram.Chats;
/// <summary>
/// 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 and never answers a pending question). Idle: starts (or joins) a run and returns when
/// it settles. Busy: queued as a follow-up (never steering), returns when the run that took it settles.
/// </summary>
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);
}
}

10
hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs
View File

@@ -4,6 +4,7 @@ using Volcanic.AgenticCore.Abstractions.Events;
using Volcanic.AgenticCore.Abstractions.Extensions.Events;
using Volcanic.AgenticCore.Abstractions.Messages;
using Volcanic.AgenticCore.Abstractions.Ui;
using Volcanic.AgenticCore.Extensions.A2A;
using Volcanic.AgenticCore.Extensions.GoogleCalendar;
using Volcanic.AgenticCore.Extensions.Scheduler;
using Volcanic.AgenticCore.Features.Agent;
@@ -80,7 +81,9 @@ public sealed class ChatRegistry(
ChatTimeZoneService? zones = null,
JobScheduler? scheduler = null,
Impersonation? impersonation = null,
TriggerServices? triggers = null) : IAsyncDisposable
TriggerServices? triggers = null,
A2AClientSettings? a2aClient = null,
A2AWatchRegistry? a2aWatches = null) : IAsyncDisposable
{
private readonly ConcurrentDictionary<long, Lazy<Task<ChatContext>>> _chats = new();
private readonly ILogger _logger = loggerFactory.CreateLogger<ChatRegistry>();
@@ -130,6 +133,9 @@ public sealed class ChatRegistry(
if (scheduler is not null)
extras.Add(ExtensionRegistration.FromInstance(new SchedulerExtension(scheduler, Scheduling.TelegramJobDelivery.TargetOf(chatId))));
if (triggers is not null) extras.Add(ExtensionRegistration.FromInstance(new TriggerExtension(chatId, triggers, () => TimeZoneFor(chatId))));
// Chat sessions only: A2A server sessions are built by the server's own factory call and never get the client.
if (a2aClient is { Enabled: true })
extras.Add(ExtensionRegistration.FromInstance(new A2AClientExtension(a2aClient, Scheduling.TelegramJobDelivery.TargetOf(chatId), a2aWatches)));
AgentSession session;
try
{
@@ -151,6 +157,8 @@ public sealed class ChatRegistry(
throw;
}
// Before the context is returned (GetAsync awaits this task), so a follow-up delivered by the scheduler or an A2A
// watch through GetAsync is always rendered.
var subscription = session.Subscribe(renderer.Post);
_logger.LogInformation("Chat {ChatId}: {Action} session {SessionId} in {Cwd}", chatId, last is null ? "new" : "resumed", session.SessionId, cwd);
foreach (var error in session.ExtensionErrors) _logger.LogWarning("Chat {ChatId}: extension error {Source}: {Error}", chatId, error.Source, error.Error);

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

@@ -12,6 +12,7 @@ using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Telegram.Bot;
using Volcanic.AgenticCore.A2A.Server;
using Volcanic.AgenticCore.Extensions.A2A;
using Volcanic.AgenticCore.Extensions.ClaudeCode;
using Volcanic.AgenticCore.Extensions.Mcp;
using Volcanic.AgenticCore.Extensions.Scheduler;
@@ -42,6 +43,8 @@ mcp.ApplyEnvironment(Environment.GetEnvironmentVariable);
// Off: no HTTP listener at all (NoHttpServer replaces Kestrel). On: the A2A endpoints on Kestrel (ASPNETCORE_URLS / Urls).
var a2a = builder.Configuration.GetSection(A2AHostOptions.SectionName).Get<A2AHostOptions>() ?? new A2AHostOptions();
if (string.IsNullOrWhiteSpace(a2a.PublicUrl)) a2a.PublicUrl = null; // compose passes an empty A2A__PublicUrl
// The A2A client (the a2a tool in each chat, watches for long remote tasks): off unless A2AClient__Enabled=true.
var a2aClient = builder.Configuration.GetSection(A2AClientSettings.SectionName).Get<A2AClientSettings>() ?? new A2AClientSettings();
// One agent directory for sessions, chat-settings.json and scheduler/jobs.json.
var agentDir = VolcanicPaths.DefaultAgentDir;
var scheduling = SchedulerSetup.Options(builder.Configuration.GetSection(SchedulerOptions.SectionName), agentDir);
@@ -108,6 +111,8 @@ if (options.Triggers.IsEnabled) TriggersSetup.Register(builder.Services, options
// Watchers get a short-lived GOOGLE_ACCESS_TOKEN for their chat's calendar, never the client secret or refresh token.
if (scheduling.Enabled && calendar is not null) builder.Services.AddSingleton<IScriptEnvironment, CalendarScriptEnvironment>();
builder.Services.AddHostedService<BotSetupService>();
// Before the server setup copies a2a: sets a2a.OnPush (push receiver -> watches) and the client's PushBaseUrl.
TelegramA2AClientSetup.Register(builder.Services, a2aClient, a2a, A2AWatchRegistry.Shared);
TelegramA2ASetup.Register(builder.Services, options, a2a);
await using var host = builder.Build();
@@ -121,6 +126,7 @@ logger.LogInformation("{Mcp}", mcp.DescribePlaywright());
logger.LogInformation("{Speech}", options.Speech.Describe());
logger.LogInformation("{Triggers}", options.Triggers.Describe());
logger.LogInformation("{A2A}", TelegramA2ASetup.Describe(a2a));
logger.LogInformation("{A2AClient}", TelegramA2AClientSetup.Describe(a2aClient));
if (ClaudeInstallation.Locate(Environment.GetEnvironmentVariable) is { } claude)
logger.LogInformation("Using Claude Code at {Path}, model {Model}", claude, options.Model);
else

9
hosts/Volcanic.AgenticCore.Telegram/Scheduling/TelegramJobDelivery.cs
View File

@@ -1,9 +1,7 @@
using System.Globalization;
using Microsoft.Extensions.Logging;
using Volcanic.AgenticCore.Abstractions.Extensions.Events;
using Volcanic.AgenticCore.Extensions.Scheduler;
using Volcanic.AgenticCore.Telegram.Chats;
using Volcanic.AgenticCore.Telegram.Inbound;
namespace Volcanic.AgenticCore.Telegram.Scheduling;
@@ -33,14 +31,11 @@ public sealed class TelegramJobDelivery(ChatRegistry chats, ChatSettingsStore se
{
var chatId = ChatIdOf(target);
var chat = await chats.GetAsync(chatId, cancellationToken).ConfigureAwait(false);
var text = MessageContext.TimestampHeader(time.GetUtcNow(), TimeZoneFor(target)) + "\n" + prompt;
try
{
// Idle: starts (or joins, as a follow-up) a run and returns when it settles. Busy: queued as a follow-up (never
// steering) and returns at once; the job then stays in flight until the run that consumed it settles.
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);
// steering); the job then stays in flight until the run that consumed it settles.
await ChatFollowUps.RunAsync(chat, prompt, TimeZoneFor(target), time, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{

1
hosts/Volcanic.AgenticCore.Telegram/Volcanic.AgenticCore.Telegram.csproj
View File

@@ -7,6 +7,7 @@
<FrameworkReference Include="Microsoft.AspNetCore.App" />
<ProjectReference Include="../../src/Volcanic.AgenticCore/Volcanic.AgenticCore.csproj" />
<ProjectReference Include="../../src/Volcanic.AgenticCore.A2A.Server/Volcanic.AgenticCore.A2A.Server.csproj" />
<ProjectReference Include="../../extensions/Volcanic.AgenticCore.Extensions.A2A/Volcanic.AgenticCore.Extensions.A2A.csproj" />
<ProjectReference Include="../../extensions/Volcanic.AgenticCore.Extensions.ClaudeCode/Volcanic.AgenticCore.Extensions.ClaudeCode.csproj" />
<ProjectReference Include="../../extensions/Volcanic.AgenticCore.Extensions.GoogleCalendar/Volcanic.AgenticCore.Extensions.GoogleCalendar.csproj" />
<ProjectReference Include="../../extensions/Volcanic.AgenticCore.Extensions.Mcp/Volcanic.AgenticCore.Extensions.Mcp.csproj" />

3
hosts/Volcanic.AgenticCore.Telegram/appsettings.json
View File

@@ -11,5 +11,6 @@
"AllowedUserIds": [],
"Model": "claude-code/claude-opus-5-5"
},
"A2A": { "Enabled": false, "PublicUrl": null, "MaxHops": 3, "SessionIdleTimeout": "00:30:00", "MaxFileBytes": 10485760, "AllowPrivatePushUrls": false }
"A2A": { "Enabled": false, "PublicUrl": null, "MaxHops": 3, "SessionIdleTimeout": "00:30:00", "MaxFileBytes": 10485760, "AllowPrivatePushUrls": false },
"A2AClient": { "Enabled": false, "WaitSeconds": 60, "WatchTimeout": "1.00:00:00", "CardTimeoutSeconds": 10, "MaxFileBytes": 10485760 }
}