From be3c77b4d0691bcecb65df65f8bcbc4c8e6cd741 Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Mon, 5 Oct 2026 16:26:16 +0000 Subject: [PATCH] feat(telegram): watch groups silently and route Jev topic triggers to the owner's chat --- .../BotSetupService.cs | 3 + .../Chats/ChatCommands.cs | 10 +- .../Chats/ChatRegistry.cs | 5 +- .../Chats/UpdateRouter.cs | 15 +- .../Groups/GroupFeed.cs | 108 ++++++++ .../Groups/GroupMembership.cs | 33 +++ .../Polling/PollingService.cs | 2 +- .../Volcanic.AgenticCore.Telegram/Program.cs | 6 +- .../Triggers/TriggerTool.cs | 234 ++++++++++++++++++ .../Triggers/TriggersSetup.cs | 58 +++++ .../GroupChatTests.cs | 173 +++++++++++++ .../Support/ChatHost.cs | 35 ++- .../Support/FakeBotClient.cs | 7 + .../Support/Updates.cs | 37 +++ 14 files changed, 716 insertions(+), 10 deletions(-) create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Groups/GroupFeed.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Groups/GroupMembership.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerTool.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggersSetup.cs create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/GroupChatTests.cs diff --git a/hosts/Volcanic.AgenticCore.Telegram/BotSetupService.cs b/hosts/Volcanic.AgenticCore.Telegram/BotSetupService.cs index 3ee4014..eece049 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/BotSetupService.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/BotSetupService.cs @@ -60,6 +60,9 @@ public sealed class BotSetupService(ITelegramBotClient bot, TelegramOptions opti var menu = GoogleCalendarSetup.MenuFor(ChatCommands.Menu, calendar); await bot.SetMyCommands(menu, cancellationToken: stoppingToken).ConfigureAwait(false); logger.LogInformation("Bot @{Username} ({Id}) is ready", me.Username, me.Id); + if (options.Triggers.IsEnabled && !me.CanReadAllGroupMessages) + logger.LogWarning("Privacy mode is on: in groups the bot only sees commands, mentions and replies, so group triggers miss most messages. " + + "Turn it off with @BotFather (/setprivacy → Disable), then remove and re-add the bot to each group, or make it a group admin."); await SetAdminMenuAsync(menu, stoppingToken).ConfigureAwait(false); return true; } diff --git a/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs b/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs index 3cf2b44..935f20a 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs @@ -6,11 +6,12 @@ using Volcanic.AgenticCore.Extensions.Scheduler; using Volcanic.AgenticCore.Hosting; using Volcanic.AgenticCore.Telegram.Rendering; using Volcanic.AgenticCore.Telegram.Scheduling; +using Volcanic.AgenticCore.Telegram.Triggers; 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) +public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zones = null, JobScheduler? scheduler = null, TriggerServices? triggers = null) { public static readonly IReadOnlyList Menu = [ @@ -23,6 +24,7 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon new() { Command = "thinking", Description = "Show or set the thinking level" }, new() { Command = "timezone", Description = "Show or set this chat's time zone" }, .. SchedulerSetup.Menu, + TriggersSetup.MenuCommand, new() { Command = "help", Description = "What this bot does" }, ]; @@ -109,6 +111,10 @@ public sealed class ChatCommands(TelegramSender sender, ChatTimeZoneService? zon await ReplyAsync(chat, await TimeZoneAsync(chat, command.Args, cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false); return true; + case "triggers": + await ReplyAsync(chat, triggers is null ? TriggersSetup.DisabledReply : "**Group triggers**\n" + triggers.Describe(chat.ChatId, chat.Session.TimeZone), cancellationToken).ConfigureAwait(false); + return true; + case "jobs": await ReplyAsync(chat, Jobs(chat), cancellationToken).ConfigureAwait(false); return true; @@ -140,7 +146,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 + GoogleCalendarSetup.HelpLines(chat.Session.Commands); + """ + SchedulerSetup.HelpLines + TriggersSetup.HelpLines + GoogleCalendarSetup.HelpLines(chat.Session.Commands); private string Status(ChatContext chat) { diff --git a/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs b/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs index 2f00f49..6e161f9 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs @@ -11,6 +11,7 @@ using Volcanic.AgenticCore.Features.Extensions; using Volcanic.AgenticCore.Hosting; using Volcanic.AgenticCore.Telegram.Inbound; using Volcanic.AgenticCore.Telegram.Rendering; +using Volcanic.AgenticCore.Telegram.Triggers; namespace Volcanic.AgenticCore.Telegram.Chats; @@ -78,7 +79,8 @@ public sealed class ChatRegistry( ChatSettingsStore? settings = null, ChatTimeZoneService? zones = null, JobScheduler? scheduler = null, - Impersonation? impersonation = null) : IAsyncDisposable + Impersonation? impersonation = null, + TriggerServices? triggers = null) : IAsyncDisposable { private readonly ConcurrentDictionary>> _chats = new(); private readonly ILogger _logger = loggerFactory.CreateLogger(); @@ -127,6 +129,7 @@ public sealed class ChatRegistry( if (zones is not null) extras.Add(ExtensionRegistration.FromInstance(new ChatSettingsExtension(chatId, zones))); 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)))); AgentSession session; try { diff --git a/hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs b/hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs index 99bc3b4..e2d6878 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs @@ -1,6 +1,7 @@ using Microsoft.Extensions.Logging; using Telegram.Bot.Types; using Telegram.Bot.Types.Enums; +using Volcanic.AgenticCore.Telegram.Groups; using Volcanic.AgenticCore.Telegram.Inbound; using Volcanic.AgenticCore.Telegram.Polling; using Volcanic.AgenticCore.Telegram.Rendering; @@ -12,7 +13,9 @@ namespace Volcanic.AgenticCore.Telegram.Chats; /// the chat's pending dialog, bot commands run directly, everything else goes through the chat's : /// the router takes a lease in update order, composes the message in the background, and the inbox turns the batch into /// a prompt that runs in the background. The admin's /impersonate is handled first; while it is on, the admin's -/// updates are routed as the impersonated user's (see ). +/// updates are routed as the impersonated user's (see ). Group messages and the bot's own +/// membership changes go to the before any allow-list check (every member's messages are +/// classified); without one (no OpenRouter key) they are dropped. Nothing is ever sent to a group. /// public sealed class UpdateRouter( TelegramOptions options, @@ -21,7 +24,8 @@ public sealed class UpdateRouter( TelegramSender sender, InboundComposer composer, ILogger logger, - Impersonation? impersonation = null) : IUpdateRouter + Impersonation? impersonation = null, + GroupFeed? groups = null) : IUpdateRouter { /// The id the bot acts for: the impersonated user's for the admin while impersonating, else . private long Effective(long id) => impersonation?.EffectiveId(id) ?? id; @@ -33,9 +37,16 @@ public sealed class UpdateRouter( case { CallbackQuery: { } query }: await OnCallbackAsync(query, cancellationToken).ConfigureAwait(false); break; + case { Message: { } message } when GroupFeed.IsGroup(message.Chat): + if (groups is not null) await groups.OnMessageAsync(message, cancellationToken).ConfigureAwait(false); + else logger.LogDebug("Ignored a message in group {ChatId}: group triggers are off", message.Chat.Id); + break; case { Message: { } message }: await OnMessageAsync(message, cancellationToken).ConfigureAwait(false); break; + case { MyChatMember: { } member }: + if (groups is not null) await groups.OnMyChatMemberAsync(member, cancellationToken).ConfigureAwait(false); + break; } } diff --git a/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupFeed.cs b/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupFeed.cs new file mode 100644 index 0000000..1b509d6 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupFeed.cs @@ -0,0 +1,108 @@ +using Microsoft.Extensions.Logging; +using Telegram.Bot; +using Telegram.Bot.Types; +using Telegram.Bot.Types.Enums; +using Volcanic.AgenticCore.Telegram.Triggers; + +namespace Volcanic.AgenticCore.Telegram.Groups; + +/// +/// The router's entry for groups. A message from an unknown group is dropped unless an allowed user posted it (which +/// vouches for the group); a known group's messages go to in the background, at most +/// at a time, so polling never waits on Jev. my_chat_member +/// updates add the group (when an allowed user added the bot) or pause it (when the bot left). Nothing here ever +/// sends anything to a group. +/// +public sealed class GroupFeed( + GroupRegistry groups, + TriggerEngine engine, + ITelegramBotClient bot, + TelegramOptions options, + ILogger logger) +{ + private readonly SemaphoreSlim _gate = new(Math.Max(1, options.Triggers.MaxConcurrency)); + private readonly Lock _lock = new(); + private readonly List _inFlight = []; + + public static bool IsGroup(Chat chat) => chat?.Type is ChatType.Group or ChatType.Supergroup; + + public async Task OnMessageAsync(Message message, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(message); + if (message.MigrateToChatId is { } to) + { + await groups.MigrateAsync(message.Chat.Id, to, cancellationToken).ConfigureAwait(false); + await engine.Store.MoveGroupAsync(message.Chat.Id, to, cancellationToken).ConfigureAwait(false); + return; + } + if (message.From?.Id == bot.BotId || message.MigrateFromChatId is not null) return; + + var known = groups.Find(message.Chat.Id); + if ((known is null || !known.Active) && options.IsAllowed(message.From?.Id)) + { + await groups.VouchAsync(message.Chat, message.From!.Id, cancellationToken).ConfigureAwait(false); + known = groups.Find(message.Chat.Id); + } + if (known is null || !known.Active) + { + logger.LogInformation("Group \"{GroupTitle}\" ({GroupId}) msg {MessageId} from {Sender}: skipped, {Reason}", + GroupMessage.TitleOf(message.Chat), message.Chat.Id, message.Id, GroupMessage.SenderOf(message), known is null ? "unknown group" : "inactive group"); + return; + } + await groups.RefreshAsync(message.Chat, cancellationToken).ConfigureAwait(false); + + var item = GroupMessage.From(message); + var task = Task.Run(async () => + { + await _gate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await engine.ProcessAsync(item, cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + } + catch (Exception ex) + { + logger.LogError(ex, "Classifying message {MessageId} of group {GroupId} failed", item.MessageId, item.GroupId); + } + finally + { + _gate.Release(); + } + }, CancellationToken.None); + lock (_lock) + { + _inFlight.RemoveAll(t => t.IsCompleted); + _inFlight.Add(task); + } + } + + public async Task OnMyChatMemberAsync(ChatMemberUpdated update, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(update); + if (!IsGroup(update.Chat)) return; + switch (update.NewChatMember.Status) + { + case ChatMemberStatus.Member or ChatMemberStatus.Administrator or ChatMemberStatus.Restricted: + if (options.IsAllowed(update.From.Id)) + await groups.VouchAsync(update.Chat, update.From.Id, cancellationToken).ConfigureAwait(false); + else if (groups.Find(update.Chat.Id) is null) + logger.LogWarning("The bot was added to group {GroupId} \"{Title}\" by user {UserId}, who is not allowed; its messages are ignored", + update.Chat.Id, GroupMessage.TitleOf(update.Chat), update.From.Id); + break; + case ChatMemberStatus.Left or ChatMemberStatus.Kicked: + await groups.SetInactiveAsync(update.Chat.Id, cancellationToken).ConfigureAwait(false); + break; + } + } + + /// Completes when every message handed to the engine so far has been classified (tests). + public async Task IdleAsync() + { + Task[] pending; + lock (_lock) pending = [.. _inFlight]; + await Task.WhenAll(pending).ConfigureAwait(false); + await engine.IdleAsync().ConfigureAwait(false); + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupMembership.cs b/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupMembership.cs new file mode 100644 index 0000000..66d160f --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Groups/GroupMembership.cs @@ -0,0 +1,33 @@ +using Telegram.Bot; +using Telegram.Bot.Types; +using Telegram.Bot.Types.Enums; + +namespace Volcanic.AgenticCore.Telegram.Groups; + +/// Whether a user is in a group: only members may put a trigger on it. +public interface IGroupMembership +{ + Task IsMemberAsync(long groupId, long userId, CancellationToken cancellationToken); +} + +/// getChatMember; any failure (the bot was removed, the user is unknown) counts as not a member. +public sealed class TelegramGroupMembership(ITelegramBotClient bot) : IGroupMembership +{ + public async Task IsMemberAsync(long groupId, long userId, CancellationToken cancellationToken) + { + try + { + var member = await bot.GetChatMember(groupId, userId, cancellationToken).ConfigureAwait(false); + return member.Status switch + { + ChatMemberStatus.Creator or ChatMemberStatus.Administrator or ChatMemberStatus.Member => true, + ChatMemberStatus.Restricted => member is ChatMemberRestricted { IsMember: true }, + _ => false, + }; + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + return false; + } + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs b/hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs index 3ea1360..7a09921 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs @@ -33,7 +33,7 @@ public sealed class PollingService( IHostApplicationLifetime lifetime, ILogger logger) : BackgroundService, IUpdateHandler { - internal static readonly UpdateType[] AllowedUpdates = [UpdateType.Message, UpdateType.CallbackQuery]; + internal static readonly UpdateType[] AllowedUpdates = [UpdateType.Message, UpdateType.CallbackQuery, UpdateType.MyChatMember]; private int _offset; diff --git a/hosts/Volcanic.AgenticCore.Telegram/Program.cs b/hosts/Volcanic.AgenticCore.Telegram/Program.cs index 586c0ee..d47bd60 100644 --- a/hosts/Volcanic.AgenticCore.Telegram/Program.cs +++ b/hosts/Volcanic.AgenticCore.Telegram/Program.cs @@ -1,4 +1,5 @@ -// Telegram front end for the agent: long polling, one agent session per private chat, Claude Code as the model. +// Telegram front end for the agent: long polling, one agent session per private chat, Claude Code as the model; +// groups are watched silently for the owners' topic triggers (Jev through OpenRouter). // // TELEGRAM_BOT_TOKEN=123:abc TELEGRAM_ALLOWED_USER_IDS=111111 dotnet run --project hosts/Volcanic.AgenticCore.Telegram @@ -21,6 +22,7 @@ using Volcanic.AgenticCore.Telegram.Polling; using Volcanic.AgenticCore.Telegram.Rendering; using Volcanic.AgenticCore.Telegram.Scheduling; using Volcanic.AgenticCore.Telegram.Speech; +using Volcanic.AgenticCore.Telegram.Triggers; var builder = Host.CreateApplicationBuilder(args); @@ -96,6 +98,7 @@ builder.Services.AddSingleton(); builder.Services.AddSingleton(); builder.Services.AddHostedService(sp => sp.GetRequiredService()); if (scheduling.Enabled) SchedulerSetup.Register(builder.Services); // the hosted service starts after PollingService +if (options.Triggers.IsEnabled) TriggersSetup.Register(builder.Services, options, agentDir); // groups are ignored without a key // 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(); builder.Services.AddHostedService(); @@ -108,6 +111,7 @@ HostProcessHardening.Apply(processEnvironment, logger); logger.LogInformation("{ProcessEnvironment}", TelegramProcessEnvironment.Describe(processEnvironment)); logger.LogInformation("{Mcp}", mcp.DescribePlaywright()); logger.LogInformation("{Speech}", options.Speech.Describe()); +logger.LogInformation("{Triggers}", options.Triggers.Describe()); if (ClaudeInstallation.Locate(Environment.GetEnvironmentVariable) is { } claude) logger.LogInformation("Using Claude Code at {Path}, model {Model}", claude, options.Model); else diff --git a/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerTool.cs b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerTool.cs new file mode 100644 index 0000000..abfda57 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerTool.cs @@ -0,0 +1,234 @@ +using System.ComponentModel; +using System.Globalization; +using System.Text; +using Volcanic.AgenticCore.Abstractions.Extensions; +using Volcanic.AgenticCore.Abstractions.Tools; +using Volcanic.AgenticCore.Telegram.Groups; + +namespace Volcanic.AgenticCore.Telegram.Triggers; + +/// What the trigger tool, the extension and /triggers share; registered only when triggers are on. +public sealed class TriggerServices(TriggerEngine engine, GroupRegistry groups, IGroupMembership membership, TriggerOptions options) +{ + public TriggerEngine Engine => engine; + public TriggerStore Store => engine.Store; + public GroupRegistry Groups => groups; + public IGroupMembership Membership => membership; + public TriggerOptions Options => options; + + /// The chat's triggers as text (list and /triggers). + public string Describe(long owner, TimeZoneInfo zone) + { + ArgumentNullException.ThrowIfNull(zone); + var now = engine.Now; + var triggers = Store.List(owner); + var body = triggers.Count == 0 + ? "No triggers." + : string.Join("\n", triggers.Select(t => TriggerPrompts.Line(t, zone, now, t.SnoozedUntil is null ? 0 : engine.Queued(owner, t.Id)))); + return body + $"\nNow: {TriggerPrompts.Stamp(now, zone)} ({zone.Id})"; + } +} + +public sealed record TriggerArgs( + [property: Description("groups, create, list, update, delete, snooze or unsnooze")] string Action, + [property: Description("update, delete, snooze, unsnooze: the trigger id (see list). snooze and unsnooze without id apply to every trigger of this chat")] string? Id = null, + [property: Description("create: the group's title or id (see groups)")] string? Group = null, + [property: Description("create, update: a short label, e.g. \"deploys\"")] string? Name = null, + [property: Description("create, update: the yes/no question asked about ONE group message, e.g. \"Is this message about deployments, releases or production outages?\"")] string? Instructions = null, + [property: Description("create, update: what counts as yes, concretely (empty clears)")] string? CriteriaTrue = null, + [property: Description("create, update: what counts as no, concretely (empty clears)")] string? CriteriaFalse = null, + [property: Description("create, update: the yes-probability (0-1, exclusive) a message needs to match; default 0.7")] double? Threshold = null, + [property: Description("update: false pauses the trigger, true resumes it")] bool? Enabled = null, + [property: Description("snooze: for how long, e.g. 90m, 2h, 1d, 1h30m")] string? Duration = null, + [property: Description("snooze: until this ISO 8601 date-time; without an offset it's the chat's local time")] string? Until = null); + +/// +/// The model's view of this chat's group triggers. Bound to one owner (the private chat); it never sees another +/// chat's triggers. Membership is checked against the owner's Telegram user id, which is the private chat's id. +/// +public sealed class TriggerTool(long owner, TriggerServices services, Func zone) : Tool +{ + public const string ToolName = "trigger"; + + public override string Name => ToolName; + + public override string Description => + "Watch Telegram groups the bot is in for topics the user cares about. A trigger is a yes/no question a fast classifier (Jev) asks about every message of one group; " + + "when a message scores at or above the threshold you get it as a prompt (with its score) and decide how to tell the user. " + + "Actions: groups (watched groups), create, list, update, delete, snooze (collect quietly, then check everything as one batch when the snooze ends), unsnooze."; + + public override string? PromptSnippet => "trigger: watch Telegram groups for topics and get relevant messages as prompts"; + + public override IReadOnlyList PromptGuidelines => + [ + "trigger: write instructions as a yes/no question about ONE message, with concrete criteriaTrue/criteriaFalse (what counts, what doesn't). Restate the trigger in plain words and get the user's confirmation before create.", + "trigger: a match arrives as a prompt starting with [Trigger …] and listing each message with its score. Tell the user what matters, briefly, with the link; never post in the group.", + "trigger: when the user says a match wasn't relevant: if it scored well above the threshold, tighten the criteria (the question is ambiguous); if it was close to the threshold, raise the threshold. When the user reports a missed message, lower the threshold or broaden the criteria. Always say what you changed.", + "trigger: when the user wants quiet (\"not until tonight\", \"only once an hour\"), snooze: messages are collected and checked as one batch when it ends, and you get one prompt if anything is related. For \"once an hour\" snooze again after each batch.", + "trigger: the bot only sees group messages if privacy mode is off (@BotFather /setprivacy → Disable) or it is a group admin; an allowed user must add it to the group (or post there once).", + ]; + + public override ToolExecutionMode ExecutionMode => ToolExecutionMode.Sequential; + + protected override async Task ExecuteAsync(TriggerArgs args, ToolInvocation invocation, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(args); + try + { + return (args.Action ?? "").Trim().ToLowerInvariant() switch + { + "groups" => ToolResult.Text(Groups()), + "create" => await CreateAsync(args, cancellationToken).ConfigureAwait(false), + "list" => ToolResult.Text(services.Describe(owner, zone())), + "update" => await UpdateAsync(args, cancellationToken).ConfigureAwait(false), + "delete" => await DeleteAsync(args, cancellationToken).ConfigureAwait(false), + "snooze" => await SnoozeAsync(args, cancellationToken).ConfigureAwait(false), + "unsnooze" => await UnsnoozeAsync(args, cancellationToken).ConfigureAwait(false), + _ => throw Error("action must be groups, create, list, update, delete, snooze or unsnooze."), + }; + } + catch (TriggerLimitException ex) + { + throw Error(ex.Message); + } + } + + private string Groups() + { + var groups = services.Groups.All().Where(g => g.Active).ToList(); + if (groups.Count == 0) + return "The bot isn't watching any group yet. An allowed user has to add it to a group (or post there once while it's a member), " + + "and privacy mode must be off (@BotFather /setprivacy → Disable) or the bot must be an admin, so it sees every message."; + var mine = services.Store.List(owner); + var sb = new StringBuilder("Watched groups:"); + foreach (var g in groups) + { + sb.Append(CultureInfo.InvariantCulture, $"\n- {g.Title} (id {g.Id.ToString(CultureInfo.InvariantCulture)})"); + var ids = mine.Where(t => t.GroupId == g.Id).Select(t => t.Id).ToList(); + if (ids.Count > 0) sb.Append(" · your triggers: ").Append(string.Join(", ", ids)); + } + return sb.ToString(); + } + + private async Task CreateAsync(TriggerArgs args, CancellationToken cancellationToken) + { + var group = ResolveGroup(args.Group); + if (string.IsNullOrWhiteSpace(args.Name)) throw Error("create needs name."); + if (string.IsNullOrWhiteSpace(args.Instructions)) throw Error("create needs instructions (a yes/no question about one message)."); + var threshold = Threshold(args.Threshold) ?? services.Options.DefaultThreshold; + if (!await services.Membership.IsMemberAsync(group.Id, owner, cancellationToken).ConfigureAwait(false)) + throw Error($"The user isn't a member of \"{group.Title}\" (or the bot can't check), so no trigger can watch it for them."); + var created = await services.Store.AddAsync(owner, new NewTrigger(args.Name.Trim(), group.Id, group.Title, args.Instructions.Trim(), + Clean(args.CriteriaTrue), Clean(args.CriteriaFalse), threshold), cancellationToken).ConfigureAwait(false); + return ToolResult.Text("Created: " + TriggerPrompts.Line(created, zone(), services.Engine.Now, 0)); + } + + private async Task UpdateAsync(TriggerArgs args, CancellationToken cancellationToken) + { + var id = RequireId(args.Id, "update"); + var threshold = Threshold(args.Threshold); + if (args.Name is not null && args.Name.Trim().Length == 0) throw Error("name can't be empty."); + if (args.Instructions is not null && args.Instructions.Trim().Length == 0) throw Error("instructions can't be empty."); + var updated = await services.Store.UpdateAsync(owner, id, t => t with + { + Name = args.Name?.Trim() ?? t.Name, + Instructions = args.Instructions?.Trim() ?? t.Instructions, + CriteriaTrue = args.CriteriaTrue is null ? t.CriteriaTrue : Clean(args.CriteriaTrue), + CriteriaFalse = args.CriteriaFalse is null ? t.CriteriaFalse : Clean(args.CriteriaFalse), + Threshold = threshold ?? t.Threshold, + Enabled = args.Enabled ?? t.Enabled, + }, cancellationToken).ConfigureAwait(false) ?? throw Error($"No trigger {id} in this chat."); + return ToolResult.Text("Updated: " + TriggerPrompts.Line(updated, zone(), services.Engine.Now, services.Engine.Queued(owner, updated.Id))); + } + + private async Task DeleteAsync(TriggerArgs args, CancellationToken cancellationToken) + { + var id = RequireId(args.Id, "delete"); + var removed = await services.Engine.DeleteAsync(owner, id, cancellationToken).ConfigureAwait(false) ?? throw Error($"No trigger {id} in this chat."); + return ToolResult.Text($"Deleted {removed.Id} (\"{removed.Name}\")."); + } + + private async Task SnoozeAsync(TriggerArgs args, CancellationToken cancellationToken) + { + var tz = zone(); + var now = services.Engine.Now; + DateTimeOffset until; + if (!string.IsNullOrWhiteSpace(args.Duration) && !string.IsNullOrWhiteSpace(args.Until)) throw Error("snooze takes duration or until, not both."); + if (!string.IsNullOrWhiteSpace(args.Duration)) + until = TriggerPrompts.TryParseDuration(args.Duration, out var duration) ? now + duration : throw Error("duration looks like 90m, 2h, 1d or 1h30m."); + else if (!string.IsNullOrWhiteSpace(args.Until)) + until = TriggerPrompts.TryParseUntil(args.Until, tz, out var at) ? at : throw Error("until must be an ISO 8601 date-time, e.g. 2026-10-05T18:00."); + else + throw Error("snooze needs duration or until."); + if (until <= now) throw Error("That time has already passed."); + if (until - now > services.Options.MaxSnooze) + throw Error($"A snooze can last at most {services.Options.MaxSnooze.TotalDays.ToString("0.#", CultureInfo.InvariantCulture)} days."); + + var snoozed = await services.Engine.SnoozeAsync(owner, Id(args.Id), until, cancellationToken).ConfigureAwait(false); + if (snoozed.Count == 0) throw Error(args.Id is null ? "This chat has no triggers." : $"No trigger {args.Id.Trim()} in this chat."); + return ToolResult.Text($"Snoozed {string.Join(", ", snoozed.Select(t => t.Id))} until {TriggerPrompts.Stamp(until, tz)} ({tz.Id}). " + + "Messages are collected meanwhile and checked as one batch when the snooze ends; you get one prompt if anything is related."); + } + + private async Task UnsnoozeAsync(TriggerArgs args, CancellationToken cancellationToken) + { + var id = Id(args.Id); + if (id is not null && services.Store.Find(owner, id) is null) throw Error($"No trigger {id} in this chat."); + var results = await services.Engine.UnsnoozeAsync(owner, id, cancellationToken).ConfigureAwait(false); + if (results.Count == 0) return ToolResult.Text("Nothing is snoozed."); + var lines = results.Select(r => r switch + { + { Error: { } error } => $"{r.Trigger.Id}: Jev failed ({error}); the {Count(r.Count)} were lost.", + { Count: 0 } => $"{r.Trigger.Id}: nothing was queued.", + { Matched: true } => $"{r.Trigger.Id}: {Count(r.Count)}, related (score {r.Score!.Value.ToString("0.00", CultureInfo.InvariantCulture)}); they follow as a separate prompt.", + { Score: { } score } => $"{r.Trigger.Id}: {Count(r.Count)}, nothing related (best score {score.ToString("0.00", CultureInfo.InvariantCulture)}).", + _ => $"{r.Trigger.Id}: {Count(r.Count)} dropped (the trigger is off).", + }); + return ToolResult.Text("Unsnoozed.\n" + string.Join("\n", lines)); + } + + private KnownGroup ResolveGroup(string? text) + { + var key = text?.Trim() ?? ""; + if (key.Length == 0) throw Error("create needs group (see action groups)."); + var groups = services.Groups.All().Where(g => g.Active).ToList(); + if (long.TryParse(key, NumberStyles.AllowLeadingSign, CultureInfo.InvariantCulture, out var id)) + return groups.Find(g => g.Id == id) ?? throw Error($"The bot isn't watching a group with id {key} (see action groups)."); + var exact = groups.Where(g => string.Equals(g.Title, key, StringComparison.OrdinalIgnoreCase)).ToList(); + var matches = exact.Count > 0 ? exact : groups.Where(g => g.Title.Contains(key, StringComparison.OrdinalIgnoreCase)).ToList(); + return matches.Count switch + { + 1 => matches[0], + 0 => throw Error($"No watched group matches \"{key}\" (see action groups)."), + _ => throw Error($"\"{key}\" matches several groups: {string.Join(", ", matches.Select(g => $"{g.Title} ({g.Id.ToString(CultureInfo.InvariantCulture)})"))}. Use the id."), + }; + } + + private static double? Threshold(double? value) => + value is { } v && (v <= 0 || v >= 1 || double.IsNaN(v)) ? throw Error("threshold must be between 0 and 1 (exclusive).") : value; + + private static string RequireId(string? id, string action) => Id(id) ?? throw Error($"{action} needs id (see action list)."); + + private static string? Id(string? id) => string.IsNullOrWhiteSpace(id) ? null : id.Trim().ToLowerInvariant(); + + private static string? Clean(string? text) => string.IsNullOrWhiteSpace(text) ? null : text.Trim(); + + private static string Count(int n) => n.ToString(CultureInfo.InvariantCulture) + (n == 1 ? " message" : " messages"); + + private static ToolExecutionException Error(string message) => new(ToolResult.Text(message)); +} + +/// Host extension per private chat: the trigger tool. +public sealed class TriggerExtension(long chatId, TriggerServices services, Func zone) : IExtension +{ + public const string ExtensionName = "telegram-triggers"; + + public string Name => ExtensionName; + + public ValueTask ActivateAsync(IExtensionApi api, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(api); + api.RegisterTool(new TriggerTool(chatId, services, zone)); + return ValueTask.CompletedTask; + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggersSetup.cs b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggersSetup.cs new file mode 100644 index 0000000..1bc63fe --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggersSetup.cs @@ -0,0 +1,58 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.DependencyInjection.Extensions; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Telegram.Bot.Types; +using Volcanic.AgenticCore.Telegram.Chats; +using Volcanic.AgenticCore.Telegram.Groups; + +namespace Volcanic.AgenticCore.Telegram.Triggers; + +/// Everything the Telegram host does for group triggers, kept out of the shared host files. +public static class TriggersSetup +{ + public const string DisabledReply = "Group triggers are off on this bot (no OPENROUTER_API_KEY)."; + + public const string HelpLines = "\n/triggers – this chat's group triggers (ask me to create, change or snooze them)"; + + public static readonly BotCommand MenuCommand = new() { Command = "triggers", Description = "Group triggers watching for topics" }; + + /// The Jev client, groups, store, engine, delivery, feed and the hosted service. Only when triggers are on. + public static void Register(IServiceCollection services, TelegramOptions options, string agentDir) + { + ArgumentNullException.ThrowIfNull(services); + ArgumentNullException.ThrowIfNull(options); + var triggers = options.Triggers; + services.TryAddSingleton(TimeProvider.System); + services.AddSingleton(triggers); + // The engine (a singleton) keeps this typed client for the process lifetime: let the handler recycle its pooled + // connections instead, so DNS changes are still picked up. + services.AddHttpClient(client => client.Timeout = TimeSpan.FromSeconds(30)) + .ConfigurePrimaryHttpMessageHandler(() => new SocketsHttpHandler { PooledConnectionLifetime = TimeSpan.FromMinutes(5) }) + .SetHandlerLifetime(Timeout.InfiniteTimeSpan); + services.AddSingleton(sp => new GroupRegistry(GroupRegistry.DefaultPath(agentDir), sp.GetRequiredService>(), sp.GetRequiredService())); + services.AddSingleton(sp => new TriggerStore(options.WorkspaceRoot, triggers.MaxTriggersPerChat, sp.GetRequiredService>(), sp.GetRequiredService())); + services.AddSingleton(new SnoozeQueue(triggers.MaxQueuedPerTrigger)); + // Resolves the registry at call time: ChatRegistry → trigger extension → engine → delivery would be a cycle. + services.AddSingleton(sp => new TelegramTriggerDelivery( + () => sp.GetRequiredService(), sp.GetRequiredService(), sp.GetRequiredService>())); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + services.AddSingleton(); + services.AddHostedService(); + } +} + +/// Flushes snoozes that ended while the host was down and arms the snooze timer. +public sealed class TriggersHostedService(TriggerEngine engine) : IHostedService +{ + public Task StartAsync(CancellationToken cancellationToken) => engine.StartAsync(cancellationToken); + + public Task StopAsync(CancellationToken cancellationToken) + { + engine.Dispose(); + return Task.CompletedTask; + } +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/GroupChatTests.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/GroupChatTests.cs new file mode 100644 index 0000000..cc21843 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/GroupChatTests.cs @@ -0,0 +1,173 @@ +using Microsoft.Extensions.Time.Testing; +using Telegram.Bot.Requests; +using Telegram.Bot.Types.Enums; +using Volcanic.AgenticCore.Abstractions.Messages; +using Volcanic.AgenticCore.Telegram.Tests.Support; +using Volcanic.AgenticCore.Telegram.Triggers; +using Volcanic.AgenticCore.Testing; + +namespace Volcanic.AgenticCore.Telegram.Tests; + +/// Groups end to end: the bot watches silently, triggers are set up in the private chat, matches reach it. +public sealed class GroupChatTests +{ + private const long Owner = Updates.AllowedUser; + private const long Group = Updates.GroupId; + private static readonly DateTimeOffset Now = new(2026, 10, 5, 12, 0, 0, TimeSpan.Zero); + + private static ChatHost Host(FakeTimeProvider? time = null) => + new(time: time ?? new FakeTimeProvider(Now), triggers: new TriggerOptions { ApiKey = "k", Debounce = TimeSpan.FromSeconds(10) }); + + private static Task AddTriggerAsync(ChatHost host, long owner = Owner) => + host.Triggers!.Store.AddAsync(owner, new NewTrigger("deploys", Group, "Dev Team", "Is this about deploys?", null, null, 0.7), CancellationToken.None); + + private static bool SentToGroup(ChatHost host) => + host.Bot.Requests.Any(r => r is SendMessageRequest { ChatId.Identifier: Group } or SetMessageReactionRequest { ChatId.Identifier: Group } + or SendChatActionRequest { ChatId.Identifier: Group }); + + [Fact] + public async Task A_strangers_matching_message_reaches_the_owners_session_and_nothing_goes_to_the_group() + { + var time = new FakeTimeProvider(Now); + await using var host = Host(time); + await host.RouteAsync(Updates.BotMembership(ChatMemberStatus.Member)); // an allowed user adds the bot + await AddTriggerAsync(host); + host.Faux.Enqueue(FauxTurn.Say("Heads up: Alice says prod deploy is failing.")); + + await host.RouteAsync(Updates.GroupText("prod deploy is failing again", Updates.Stranger)); + await host.RouteAsync(Updates.GroupText("anyone for lunch?", Updates.Stranger)); + await host.Feed!.IdleAsync(); + time.Advance(TimeSpan.FromSeconds(10)); + await host.Engine!.IdleAsync(); + await host.SettleAsync(); + + var prompt = Assert.Single(host.Faux.Requests).Messages.Last().Text; + Assert.Contains("[Trigger \"deploys\" (t1) matched 1 message in \"Dev Team\"]", prompt, StringComparison.Ordinal); + Assert.Contains("Alice, score 0.95: prod deploy is failing again", prompt, StringComparison.Ordinal); + Assert.DoesNotContain("lunch", prompt, StringComparison.Ordinal); + Assert.Contains(host.Texts, t => t.Contains("Heads up", StringComparison.Ordinal)); + Assert.False(SentToGroup(host)); + Assert.Null(host.Chats.Find(Group)); // a group never gets a session + Assert.Equal(2, host.TriggerLog.Entries.Count(e => e.Text.StartsWith("Group \"Dev Team\"", StringComparison.Ordinal))); + } + + [Fact] + public async Task A_group_nobody_allowed_vouched_for_is_ignored_until_an_allowed_user_posts() + { + await using var host = Host(); + await AddTriggerAsync(host); + await host.RouteAsync(Updates.BotMembership(ChatMemberStatus.Member, byUser: Updates.Stranger)); + + await host.RouteAsync(Updates.GroupText("deploy now", Updates.Stranger)); + await host.Feed!.IdleAsync(); + Assert.Null(host.Groups!.Find(Group)); + Assert.Empty(((FakeJev)host.Jev!).Calls); + + await host.RouteAsync(Updates.GroupText("deploy please", Owner)); + await host.Feed.IdleAsync(); + Assert.True(host.Groups.Find(Group)!.Active); + Assert.Single(((FakeJev)host.Jev).Calls); + Assert.False(SentToGroup(host)); + } + + [Fact] + public async Task Removing_the_bot_pauses_the_group_and_migration_moves_its_triggers() + { + await using var host = Host(); + await host.RouteAsync(Updates.BotMembership(ChatMemberStatus.Administrator)); + await AddTriggerAsync(host); + + await host.RouteAsync(Updates.BotMembership(ChatMemberStatus.Kicked)); + await host.RouteAsync(Updates.GroupText("deploy", Updates.Stranger)); + await host.Feed!.IdleAsync(); + Assert.False(host.Groups!.Find(Group)!.Active); + Assert.Empty(((FakeJev)host.Jev!).Calls); + + await host.RouteAsync(Updates.BotMembership(ChatMemberStatus.Member, groupId: -5, title: "Old")); + await host.Triggers!.Store.AddAsync(Owner, new NewTrigger("old", -5, "Old", "q?", null, null, 0.7), CancellationToken.None); + var migrate = Updates.GroupText("", Updates.Stranger, groupId: -5, title: "Old", type: ChatType.Group); + migrate.Message!.Text = null; + migrate.Message.MigrateToChatId = -1009; + await host.RouteAsync(migrate); + + Assert.Equal(-1009, host.Triggers.Store.Find(Owner, "t2")!.GroupId); + Assert.NotNull(host.Groups.Find(-1009)); + } + + [Fact] + public async Task Without_triggers_group_messages_are_dropped() + { + await using var host = new ChatHost(); + await host.RouteAsync(Updates.GroupText("deploy", Owner)); + Assert.Null(host.Chats.Find(Group)); + Assert.Null(host.Chats.Find(Owner)); + Assert.Empty(host.Bot.Sent); + } + + [Fact] + public async Task The_agent_creates_lists_and_snoozes_a_trigger_through_the_tool() + { + await using var host = Host(); + await host.RouteAsync(Updates.BotMembership(ChatMemberStatus.Member)); + host.Faux.Enqueue( + FauxTurn.Tool("trigger", new { action = "groups" }), + FauxTurn.Tool("trigger", new { action = "create", group = "dev", name = "deploys", instructions = "Is this message about deploys?", criteriaTrue = "Deploys, releases, rollbacks", criteriaFalse = "Anything else", threshold = 0.8 }), + FauxTurn.Tool("trigger", new { action = "snooze", duration = "2h" }), + FauxTurn.Say("Done.")); + var chat = await host.ChatAsync(); + + await chat.Session.PromptAsync("watch Dev Team for deploys, but keep quiet for two hours"); + + var results = chat.Session.Messages.OfType().ToList(); + Assert.All(results, r => Assert.False(r.IsError, r.Text)); + Assert.Contains("- Dev Team (id -1001234567890)", results[0].Text, StringComparison.Ordinal); + Assert.StartsWith("Created: t1 \"deploys\" · Dev Team · threshold 0.80", results[1].Text, StringComparison.Ordinal); + Assert.StartsWith("Snoozed t1 until", results[2].Text, StringComparison.Ordinal); + var trigger = host.Triggers!.Store.Find(Owner, "t1")!; + Assert.Equal(("Deploys, releases, rollbacks", Now.AddHours(2)), (trigger.CriteriaTrue, trigger.SnoozedUntil)); + + await host.SendAsync("/triggers"); + await Wait.UntilAsync(() => host.Bot.Sent.Any(s => s.Text.Contains("snoozed until", StringComparison.Ordinal)), "the /triggers reply"); + } + + [Fact] + public async Task Only_members_of_a_group_can_put_a_trigger_on_it() + { + await using var host = Host(); + await host.RouteAsync(Updates.BotMembership(ChatMemberStatus.Member)); + host.Bot.NonMembers[Owner] = true; + host.Faux.Enqueue(FauxTurn.Tool("trigger", new { action = "create", group = "Dev Team", name = "x", instructions = "q?" }), FauxTurn.Say("Sorry.")); + var chat = await host.ChatAsync(); + + await chat.Session.PromptAsync("watch Dev Team"); + + var result = chat.Session.Messages.OfType().Single(); + Assert.True(result.IsError); + Assert.Contains("isn't a member", result.Text, StringComparison.Ordinal); + Assert.Empty(host.Triggers!.Store.List(Owner)); + } + + [Fact] + public async Task Tool_errors_explain_what_is_wrong() + { + await using var host = Host(); + await host.RouteAsync(Updates.BotMembership(ChatMemberStatus.Member)); + await host.RouteAsync(Updates.BotMembership(ChatMemberStatus.Member, groupId: -1009, title: "Dev Ops")); + host.Faux.Enqueue( + FauxTurn.Tool("trigger", new { action = "create", group = "dev", name = "x", instructions = "q?" }), + FauxTurn.Tool("trigger", new { action = "create", group = "Dev Team", name = "x", instructions = "q?", threshold = 1.5 }), + FauxTurn.Tool("trigger", new { action = "snooze", duration = "soon" }), + FauxTurn.Tool("trigger", new { action = "update", id = "t7", threshold = 0.5 }), + FauxTurn.Say("Sorry.")); + var chat = await host.ChatAsync(); + + await chat.Session.PromptAsync("try things"); + + var results = chat.Session.Messages.OfType().ToList(); + Assert.All(results, r => Assert.True(r.IsError)); + Assert.Contains("matches several groups", results[0].Text, StringComparison.Ordinal); + Assert.Contains("threshold must be between 0 and 1", results[1].Text, StringComparison.Ordinal); + Assert.Contains("duration looks like", results[2].Text, StringComparison.Ordinal); + Assert.Contains("No trigger t7", results[3].Text, StringComparison.Ordinal); + } +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs index 0a997ed..1a52f80 100644 --- a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs @@ -12,6 +12,8 @@ using Volcanic.AgenticCore.Telegram.Chats; using Volcanic.AgenticCore.Telegram.Inbound; using Volcanic.AgenticCore.Telegram.Rendering; using Volcanic.AgenticCore.Telegram.Scheduling; +using Volcanic.AgenticCore.Telegram.Groups; +using Volcanic.AgenticCore.Telegram.Triggers; using Volcanic.AgenticCore.Testing; namespace Volcanic.AgenticCore.Telegram.Tests.Support; @@ -53,7 +55,7 @@ public sealed class ChatHost : IAsyncDisposable public ChatHost(Action? configure = null, Action? core = null, TimeProvider? time = null, Volcanic.AgenticCore.Extensions.GoogleCalendar.GoogleCalendarOptions? calendar = null, Volcanic.AgenticCore.Telegram.Speech.ISpeechToText? speech = null, string? root = null, SchedulerOptions? scheduler = null, - IScriptRunner? runner = null) + IScriptRunner? runner = null, TriggerOptions? triggers = null, IJevClient? jev = null) { _ownsRoot = root is null; _root = root ?? Path.Combine(Path.GetTempPath(), "volcanic-telegram", Guid.NewGuid().ToString("N")); @@ -68,6 +70,7 @@ public sealed class ChatHost : IAsyncDisposable TypingInterval = TimeSpan.FromMilliseconds(50), InputDebounce = TimeSpan.FromMilliseconds(20), }; + if (triggers is not null) Options.Triggers = triggers; var store = new InMemorySessionStore(); var services = new ServiceCollection(); if (time is not null) services.AddSingleton(time); // sessions render the prompt date from it @@ -97,13 +100,36 @@ public sealed class ChatHost : IAsyncDisposable } Zones = new ChatTimeZoneService(Settings, id => Chats!.Find(id)?.Session, clock, Scheduler); Impersonation = new Impersonation(Options); - Chats = new ChatRegistry(_services.GetRequiredService(), sender, Options, loggers, time, calendar, Settings, Zones, Scheduler, Impersonation); + if (triggers is not null) + { + Jev = jev ?? new FakeJev(); + Groups = new GroupRegistry(GroupRegistry.DefaultPath(agentDir), NullLogger.Instance, clock); + var triggerStore = new TriggerStore(Options.WorkspaceRoot, triggers.MaxTriggersPerChat, NullLogger.Instance, clock); + var delivery = new TelegramTriggerDelivery(() => Chats!, clock, NullLogger.Instance); + _dispatcher = new TriggerDispatcher(delivery, triggers, clock, NullLogger.Instance); + Engine = new TriggerEngine(triggerStore, new SnoozeQueue(triggers.MaxQueuedPerTrigger), Jev, _dispatcher, delivery, clock, TriggerLog); + Triggers = new TriggerServices(Engine, Groups, new TelegramGroupMembership(Bot), triggers); + Feed = new GroupFeed(Groups, Engine, Bot, Options, NullLogger.Instance); + } + Chats = new ChatRegistry(_services.GetRequiredService(), sender, Options, loggers, time, calendar, Settings, Zones, Scheduler, Impersonation, Triggers); 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), sender, Composer, NullLogger.Instance, Impersonation); + Router = new UpdateRouter(Options, Chats, new ChatCommands(sender, Zones, Scheduler, Triggers), sender, Composer, NullLogger.Instance, Impersonation, Feed); } + private readonly TriggerDispatcher? _dispatcher; + + /// Set when the host was built with triggers: options. + public TriggerServices? Triggers { get; } + public TriggerEngine? Engine { get; } + public GroupRegistry? Groups { get; } + public GroupFeed? Feed { get; } + public IJevClient? Jev { get; } + + /// The engine's log lines (one per group message). + public ListLogger TriggerLog { get; } = new(); + /// Off until is set on . public Impersonation Impersonation { get; } @@ -140,6 +166,9 @@ public sealed class ChatHost : IAsyncDisposable public async ValueTask DisposeAsync() { if (Scheduler is not null) await Scheduler.StopAsync(CancellationToken.None); + if (Feed is not null) await Feed.IdleAsync(); + Engine?.Dispose(); + _dispatcher?.Dispose(); await Chats.DisposeAsync(); await _services.DisposeAsync(); if (_ownsRoot) diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/FakeBotClient.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/FakeBotClient.cs index 076d191..9fd814c 100644 --- a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/FakeBotClient.cs +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/FakeBotClient.cs @@ -34,6 +34,9 @@ public sealed class FakeBotClient : ITelegramBotClient /// When set, a download of a scripted file writes one byte and then fails with a network error. public bool FailDownloads { get; set; } + /// Users getChatMember reports as having left (everyone else is a member). + public ConcurrentDictionary NonMembers { get; } = new(); + /// Called for every non-poll request before it is answered; throw to simulate a failure. public Action? OnSend { get; set; } @@ -127,6 +130,10 @@ public sealed class FakeBotClient : ITelegramBotClient case GetFileRequest getFile: result = new TGFile { FileId = getFile.FileId, FileUniqueId = getFile.FileId + "-u", FilePath = $"photos/{getFile.FileId}.jpg", FileSize = 3 }; break; + case GetChatMemberRequest member: + var user = new User { Id = member.UserId, FirstName = "Member" }; + result = NonMembers.ContainsKey(member.UserId) ? new ChatMemberLeft { User = user } : new ChatMemberMember { User = user }; + break; case SendPhotoRequest photo: result = RecordUpload("sendPhoto", photo.ChatId, photo.Photo, photo.Caption, photo.ParseMode); break; diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/Updates.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/Updates.cs index aa9547d..2ae5334 100644 --- a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/Updates.cs +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/Updates.cs @@ -39,6 +39,43 @@ public static class Updates }, }; + public const long GroupId = -1001234567890; + + /// A text message in a group (a supergroup unless says otherwise). + public static Update GroupText(string text, long userId = Stranger, long groupId = GroupId, string title = "Dev Team", ChatType type = ChatType.Supergroup, + string firstName = "Alice") => new() + { + Id = Interlocked.Increment(ref _nextUpdateId), + Message = new Message + { + Id = Interlocked.Increment(ref _nextMessageId), + From = new User { Id = userId, FirstName = firstName }, + Chat = new Chat { Id = groupId, Type = type, Title = title }, + Text = text, + Date = DateTime.UtcNow, + }, + }; + + /// The bot's own membership in a group changed (added by , or removed). + public static Update BotMembership(ChatMemberStatus status, long byUser = AllowedUser, long groupId = GroupId, string title = "Dev Team") => new() + { + Id = Interlocked.Increment(ref _nextUpdateId), + MyChatMember = new ChatMemberUpdated + { + Chat = new Chat { Id = groupId, Type = ChatType.Supergroup, Title = title }, + From = new User { Id = byUser, FirstName = "Adder" }, + Date = DateTime.UtcNow, + OldChatMember = new ChatMemberLeft { User = new User { Id = BotUserId, IsBot = true, FirstName = "Fake" } }, + NewChatMember = status switch + { + ChatMemberStatus.Administrator => new ChatMemberAdministrator { User = new User { Id = BotUserId, IsBot = true, FirstName = "Fake" } }, + ChatMemberStatus.Kicked => new ChatMemberBanned { User = new User { Id = BotUserId, IsBot = true, FirstName = "Fake" } }, + ChatMemberStatus.Left => new ChatMemberLeft { User = new User { Id = BotUserId, IsBot = true, FirstName = "Fake" } }, + _ => new ChatMemberMember { User = new User { Id = BotUserId, IsBot = true, FirstName = "Fake" } }, + }, + }, + }; + public static Update WithId(int id) => new() { Id = id, Message = Text("#" + id).Message }; /// The fake bot's user id ().