From 4ce71201b07d9b13441ff5402466af46fb285fb9 Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Mon, 5 Oct 2026 16:21:41 +0000 Subject: [PATCH] feat(telegram): trigger engine with Jev classification, debounce and snooze batches --- .../Triggers/JevStates.cs | 77 +++++ .../Triggers/TelegramTriggerDelivery.cs | 35 ++ .../Triggers/TriggerDispatcher.cs | 105 ++++++ .../Triggers/TriggerEngine.cs | 300 ++++++++++++++++++ .../Triggers/TriggerPrompts.cs | 141 ++++++++ .../Support/TriggerFakes.cs | 63 ++++ .../TriggerEngineTests.cs | 252 +++++++++++++++ 7 files changed, 973 insertions(+) create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Triggers/JevStates.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Triggers/TelegramTriggerDelivery.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerDispatcher.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerEngine.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerPrompts.cs create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/Support/TriggerFakes.cs create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/TriggerEngineTests.cs diff --git a/hosts/Volcanic.AgenticCore.Telegram/Triggers/JevStates.cs b/hosts/Volcanic.AgenticCore.Telegram/Triggers/JevStates.cs new file mode 100644 index 0000000..c1fa443 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Triggers/JevStates.cs @@ -0,0 +1,77 @@ +using System.Globalization; +using System.Text.Json.Nodes; +using Volcanic.AgenticCore.Telegram.Groups; + +namespace Volcanic.AgenticCore.Telegram.Triggers; + +/// What Jev is asked: the state for one message or a snooze batch, and each trigger's question. +public static class JevStates +{ + /// { group, sender, text, reply_to? }. + public static JsonObject Single(GroupMessage message) + { + ArgumentNullException.ThrowIfNull(message); + var state = new JsonObject { ["group"] = message.GroupTitle, ["sender"] = message.Sender, ["text"] = message.Text ?? "" }; + AddReply(state, message); + return state; + } + + /// + /// { group, messages: [ { time, sender, text, reply_to? } ] }, split so each state stays under + /// characters of JSON (one oversized message still gets a state of its own). + /// + public static IReadOnlyList Batches(string groupTitle, IReadOnlyList messages, TimeZoneInfo zone, int maxChars) + { + ArgumentNullException.ThrowIfNull(messages); + ArgumentNullException.ThrowIfNull(zone); + var states = new List(); + JsonArray? current = null; + var size = 0; + foreach (var message in messages) + { + var item = new JsonObject + { + ["time"] = TimeZoneInfo.ConvertTime(message.Time, zone).ToString("yyyy-MM-dd HH:mm", CultureInfo.InvariantCulture), + ["sender"] = message.Sender, + ["text"] = message.Text ?? "", + }; + AddReply(item, message); + var itemSize = item.ToJsonString().Length + 1; + if (current is null || (size + itemSize > maxChars && current.Count > 0)) + { + current = []; + states.Add(new JsonObject { ["group"] = groupTitle, ["messages"] = current }); + size = groupTitle.Length + 30; + } + current.Add(item); + size += itemSize; + } + return states; + } + + public static JevQuestion Question(string id, Trigger trigger) + { + ArgumentNullException.ThrowIfNull(trigger); + return new JevQuestion(id, trigger.Instructions, trigger.CriteriaTrue, trigger.CriteriaFalse); + } + + /// + /// The trigger's one-message question wrapped for a batch: asked plainly, a batch scores high on loosely related + /// chatter and lower on the real thing (measured: 0.92 vs 0.99 plain, 0.12 vs 0.99 wrapped). + /// + public static JevQuestion BatchQuestion(string id, Trigger trigger) + { + ArgumentNullException.ThrowIfNull(trigger); + var both = !string.IsNullOrWhiteSpace(trigger.CriteriaTrue) && !string.IsNullOrWhiteSpace(trigger.CriteriaFalse); + return new JevQuestion(id, + "Answer yes if at least one message in the batch passes this check: " + trigger.Instructions, + both ? "At least one message: " + trigger.CriteriaTrue : null, + both ? "No message does; all are like: " + trigger.CriteriaFalse : null); + } + + private static void AddReply(JsonObject target, GroupMessage message) + { + if (message.ReplyTo is { } reply) + target["reply_to"] = new JsonObject { ["sender"] = reply.Sender, ["text"] = reply.Text ?? "" }; + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Triggers/TelegramTriggerDelivery.cs b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TelegramTriggerDelivery.cs new file mode 100644 index 0000000..8b66a36 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TelegramTriggerDelivery.cs @@ -0,0 +1,35 @@ +using Microsoft.Extensions.Logging; +using Volcanic.AgenticCore.Abstractions.Extensions.Events; +using Volcanic.AgenticCore.Telegram.Chats; +using Volcanic.AgenticCore.Telegram.Inbound; + +namespace Volcanic.AgenticCore.Telegram.Triggers; + +/// +/// Trigger prompts go to the owner's session like scheduled prompts (): +/// with the timestamp header, as a follow-up (never steering), bypassing the inbox. The registry is resolved lazily: +/// ChatRegistry → trigger extension → engine → this would otherwise be a construction cycle. +/// +public sealed class TelegramTriggerDelivery(Func chats, TimeProvider time, ILogger logger) : ITriggerDelivery +{ + public TimeZoneInfo TimeZoneFor(long owner) => chats().TimeZoneFor(owner); + + public async Task DeliverAsync(long owner, string prompt, CancellationToken cancellationToken) + { + var chat = await chats().GetAsync(owner, cancellationToken).ConfigureAwait(false); + var text = MessageContext.TimestampHeader(time.GetUtcNow(), TimeZoneFor(owner)) + "\n" + prompt; + try + { + await chat.Session.PromptAsync(text, null, StreamingBehavior.FollowUp, cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + throw; + } + catch (Exception ex) + { + logger.LogError(ex, "Trigger prompt in chat {ChatId} failed", owner); + chat.Report($"A group trigger could not be handled: {ex.Message}"); + } + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerDispatcher.cs b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerDispatcher.cs new file mode 100644 index 0000000..7642186 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerDispatcher.cs @@ -0,0 +1,105 @@ +using Microsoft.Extensions.Logging; +using Volcanic.AgenticCore.Telegram.Groups; + +namespace Volcanic.AgenticCore.Telegram.Triggers; + +/// Where trigger prompts go: the owner's private session. +public interface ITriggerDelivery +{ + /// Runs in the owner's session (as a follow-up while it is busy). + Task DeliverAsync(long owner, string prompt, CancellationToken cancellationToken); + + TimeZoneInfo TimeZoneFor(long owner); +} + +/// +/// Collects live matches per (owner, trigger): the first match opens a window, +/// later ones join it without extending it (a busy group can't hold a match back), and the window becomes one prompt. +/// +public sealed class TriggerDispatcher(ITriggerDelivery delivery, TriggerOptions options, TimeProvider time, ILogger logger) : IDisposable +{ + private sealed class Window(Trigger trigger) + { + public Trigger Trigger { get; set; } = trigger; + public List<(GroupMessage Message, double Score)> Hits { get; } = []; + public ITimer? Timer { get; set; } + } + + private readonly Lock _lock = new(); + private readonly Dictionary<(long Owner, string Trigger), Window> _windows = []; + private readonly List _deliveries = []; + + public void Add(long owner, Trigger trigger, GroupMessage message, double score) + { + ArgumentNullException.ThrowIfNull(trigger); + ArgumentNullException.ThrowIfNull(message); + var key = (owner, trigger.Id); + Window window; + lock (_lock) + { + if (_windows.TryGetValue(key, out var open)) + { + open.Trigger = trigger; + open.Hits.Add((message, score)); + return; + } + window = new Window(trigger); + window.Hits.Add((message, score)); + _windows[key] = window; + } + // Outside the lock: a zero window may fire inside CreateTimer. + var timer = time.CreateTimer(_ => Fire(key, window), null, options.Debounce, Timeout.InfiniteTimeSpan); + lock (_lock) + { + if (_windows.TryGetValue(key, out var current) && ReferenceEquals(current, window)) window.Timer = timer; + else timer.Dispose(); // already fired + } + } + + /// Completes when every prompt handed to the delivery so far has been delivered (for tests and shutdown). + public Task IdleAsync() + { + lock (_lock) return Task.WhenAll([.. _deliveries]); + } + + private void Fire((long Owner, string Trigger) key, Window window) + { + List<(GroupMessage, double)> hits; + lock (_lock) + { + if (!_windows.Remove(key, out var current) || !ReferenceEquals(current, window)) return; + window.Timer?.Dispose(); + hits = [.. window.Hits]; + } + var prompt = TriggerPrompts.Live(window.Trigger, hits, delivery.TimeZoneFor(key.Owner)); + var task = DeliverAsync(key.Owner, window.Trigger, prompt, hits.Count); + lock (_lock) + { + _deliveries.RemoveAll(t => t.IsCompleted); + _deliveries.Add(task); + } + } + + private async Task DeliverAsync(long owner, Trigger trigger, string prompt, int count) + { + try + { + await Task.Yield(); // never deliver on the timer thread + logger.LogInformation("Trigger {TriggerId} \"{Name}\": {Count} match(es) delivered to chat {Owner}", trigger.Id, trigger.Name, count, owner); + await delivery.DeliverAsync(owner, prompt, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) + { + logger.LogError(ex, "Delivering trigger {TriggerId} to chat {Owner} failed", trigger.Id, owner); + } + } + + public void Dispose() + { + lock (_lock) + { + foreach (var window in _windows.Values) window.Timer?.Dispose(); + _windows.Clear(); + } + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerEngine.cs b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerEngine.cs new file mode 100644 index 0000000..1031d30 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerEngine.cs @@ -0,0 +1,300 @@ +using System.Globalization; +using Microsoft.Extensions.Logging; +using Volcanic.AgenticCore.Telegram.Groups; + +namespace Volcanic.AgenticCore.Telegram.Triggers; + +/// What ending one trigger's snooze did. is set when Jev failed (the batch is lost). +public sealed record SnoozeFlush(Trigger Trigger, int Count, double? Score, bool Matched, string? Error = null); + +/// +/// Classifies group messages against the triggers watching their group and logs one line per message with the +/// results. Live triggers share one Jev call per message (one question each) and matches go to the +/// ; snoozed triggers queue the message instead (). When a +/// snooze ends (a timer on , re-armed on every change and capped at an hour), the queue goes +/// to Jev as one batch with the trigger's question wrapped for a batch (), and a +/// match becomes one prompt with the whole batch. +/// +public sealed class TriggerEngine( + TriggerStore store, + SnoozeQueue queue, + IJevClient jev, + TriggerDispatcher dispatcher, + ITriggerDelivery delivery, + TimeProvider time, + ILogger logger) : IDisposable +{ + private const int MaxExcerpt = 200; + private static readonly TimeSpan MaxTimerDelay = TimeSpan.FromHours(1); + private readonly SemaphoreSlim _flushGate = new(1, 1); + private readonly Lock _lock = new(); + private readonly List _background = []; + private ITimer? _timer; + + public TriggerStore Store => store; + + public DateTimeOffset Now => time.GetUtcNow(); + + /// Messages queued for a snoozed trigger. + public int Queued(long owner, string triggerId) => queue.Count(store.WorkspaceOf(owner), triggerId); + + /// Flushes snoozes that ended while the host was down, then arms the timer. + public async Task StartAsync(CancellationToken cancellationToken) + { + lock (_lock) _timer ??= time.CreateTimer(_ => OnTimer(), null, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); + await FlushDueAsync(cancellationToken).ConfigureAwait(false); + } + + /// Completes when every background flush and snooze prompt started so far has finished, and the dispatcher's too. + public async Task IdleAsync() + { + while (true) // a flush can start a delivery: wait until nothing new appeared + { + Task[] pending; + lock (_lock) pending = [.. _background.Where(t => !t.IsCompleted)]; + if (pending.Length == 0) break; + await Task.WhenAll(pending).ConfigureAwait(false); + } + await dispatcher.IdleAsync().ConfigureAwait(false); + } + + public async Task ProcessAsync(GroupMessage message, CancellationToken cancellationToken) + { + ArgumentNullException.ThrowIfNull(message); + if (string.IsNullOrWhiteSpace(message.Text)) + { + Log(LogLevel.Information, message, "(no text)", "skipped"); + return; + } + var excerpt = Excerpt(message.Text); + var watching = store.Watching(message.GroupId).Where(w => w.Trigger.Enabled).ToList(); + if (watching.Count == 0) + { + Log(LogLevel.Information, message, excerpt, "no triggers"); + return; + } + + var now = time.GetUtcNow(); + var results = new List(); + var live = new List(); + foreach (var w in watching) + { + if (!w.Trigger.IsSnoozed(now)) + { + live.Add(w); + continue; + } + var (count, dropped) = await queue.AppendAsync(store.WorkspaceOf(w.Owner), w.Trigger.Id, message, cancellationToken).ConfigureAwait(false); + results.Add($"{Label(w)} snoozed (queued {count.ToString(CultureInfo.InvariantCulture)}{(dropped ? ", oldest dropped" : "")})"); + } + + if (live.Count > 0) + { + var questions = live.Select((w, i) => JevStates.Question(QuestionId(i), w.Trigger)).ToList(); + IReadOnlyDictionary answers; + try + { + answers = await jev.AskAsync(JevStates.Single(message), questions, cancellationToken).ConfigureAwait(false); + } + catch (JevException ex) + { + results.Add($"Jev failed after {ex.Attempts.ToString(CultureInfo.InvariantCulture)} attempts: {ex.Message}"); + Log(LogLevel.Warning, message, excerpt, string.Join(", ", results)); + return; + } + for (var i = 0; i < live.Count; i++) + { + var w = live[i]; + var score = answers[QuestionId(i)]; + var match = score >= w.Trigger.Threshold; + results.Add($"{Label(w)}={score.ToString("0.00", CultureInfo.InvariantCulture)}{(match ? " MATCH" : "")}"); + if (match) dispatcher.Add(w.Owner, w.Trigger, message, score); + } + } + Log(LogLevel.Information, message, excerpt, string.Join(", ", results)); + } + + /// Snoozes one trigger (or, with no id, every trigger of the owner) until . + public async Task> SnoozeAsync(long owner, string? id, DateTimeOffset until, CancellationToken cancellationToken) + { + var targets = Targets(owner, id, onlySnoozed: false); + var snoozed = new List(); + foreach (var t in targets) + if (await store.UpdateAsync(owner, t.Id, x => x with { SnoozedUntil = until }, cancellationToken).ConfigureAwait(false) is { } updated) + snoozed.Add(updated); + Arm(); + return snoozed; + } + + /// Ends the snooze of one trigger (or every snoozed trigger of the owner) now and flushes its queue. + public async Task> UnsnoozeAsync(long owner, string? id, CancellationToken cancellationToken) + { + var results = new List(); + await _flushGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + foreach (var t in Targets(owner, id, onlySnoozed: id is null)) + if (await store.UpdateAsync(owner, t.Id, x => x with { SnoozedUntil = null }, cancellationToken).ConfigureAwait(false) is { } cleared) + results.Add(await FlushAsync(owner, cleared, cancellationToken).ConfigureAwait(false)); + } + finally + { + _flushGate.Release(); + Arm(); + } + return results; + } + + /// Removes the trigger and its snooze queue. + public async Task DeleteAsync(long owner, string id, CancellationToken cancellationToken) + { + var removed = await store.RemoveAsync(owner, id, cancellationToken).ConfigureAwait(false); + if (removed is not null) await queue.DeleteAsync(store.WorkspaceOf(owner), removed.Id, cancellationToken).ConfigureAwait(false); + Arm(); + return removed; + } + + /// Re-arms the snooze timer for the soonest snooze end (at most an hour away; it re-arms when it fires). + public void Arm() + { + ITimer? timer; + lock (_lock) timer = _timer; + if (timer is null) return; + if (store.NextSnoozeEnd() is not { } next) + { + timer.Change(Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan); + return; + } + var delay = next - time.GetUtcNow(); + timer.Change(delay < TimeSpan.Zero ? TimeSpan.Zero : delay > MaxTimerDelay ? MaxTimerDelay : delay, Timeout.InfiniteTimeSpan); + } + + private IReadOnlyList Targets(long owner, string? id, bool onlySnoozed) + { + if (!string.IsNullOrWhiteSpace(id)) return store.Find(owner, id.Trim()) is { } one ? [one] : []; + return [.. store.List(owner).Where(t => !onlySnoozed || t.SnoozedUntil is not null)]; + } + + private void OnTimer() => Track(FlushDueAsync(CancellationToken.None)); + + private async Task FlushDueAsync(CancellationToken cancellationToken) + { + await _flushGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + foreach (var due in store.SnoozeDue(time.GetUtcNow())) + if (await store.UpdateAsync(due.Owner, due.Trigger.Id, x => x with { SnoozedUntil = null }, cancellationToken).ConfigureAwait(false) is { } cleared) + await FlushAsync(due.Owner, cleared, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogError(ex, "Ending trigger snoozes failed"); + } + finally + { + _flushGate.Release(); + Arm(); + } + } + + private async Task FlushAsync(long owner, Trigger trigger, CancellationToken cancellationToken) + { + var workspace = store.WorkspaceOf(owner); + var messages = await queue.TakeAllAsync(workspace, trigger.Id, cancellationToken).ConfigureAwait(false); + var label = $"{owner.ToString(CultureInfo.InvariantCulture)}/{trigger.Id} {trigger.Name}"; + if (messages.Count == 0) + { + logger.LogInformation("Trigger {Trigger}: snooze ended, nothing queued", label); + return new SnoozeFlush(trigger, 0, null, false); + } + if (!trigger.Enabled) + { + logger.LogInformation("Trigger {Trigger}: snooze ended while off; {Count} queued messages dropped", label, messages.Count); + return new SnoozeFlush(trigger, messages.Count, null, false); + } + + var zone = delivery.TimeZoneFor(owner); + var states = JevStates.Batches(messages[^1].GroupTitle, messages, zone, TriggerOptions.ChunkChars); + var question = JevStates.BatchQuestion(QuestionId(0), trigger); + double best = 0; + for (var i = 0; i < states.Count; i++) + { + double score; + try + { + score = (await jev.AskAsync(states[i], [question], cancellationToken).ConfigureAwait(false))[question.Id]; + } + catch (JevException ex) + { + logger.LogWarning("Trigger {Trigger}: snooze batch failed, Jev failed after {Attempts} attempts: {Error}; {Count} queued messages lost", + label, ex.Attempts, ex.Message, messages.Count); + return new SnoozeFlush(trigger, messages.Count, null, false, ex.Message); + } + var size = states[i]["messages"]!.AsArray().Count; + logger.LogInformation("Trigger {Trigger}: snooze batch chunk {Chunk}/{Chunks} ({Size} messages) = {Score:0.00}", + label, i + 1, states.Count, size, score); + best = Math.Max(best, score); + } + + var matched = best >= trigger.Threshold; + logger.LogInformation("Trigger {Trigger}: snooze ended, {Count} messages, best {Score:0.00} → {Result}", + label, messages.Count, best, matched ? "MATCH, delivered" : "no match"); + if (matched) Track(DeliverSnoozedAsync(owner, trigger, messages, best, zone, workspace)); + return new SnoozeFlush(trigger, messages.Count, best, matched); + } + + private async Task DeliverSnoozedAsync(long owner, Trigger trigger, IReadOnlyList messages, double score, TimeZoneInfo zone, string workspace) + { + try + { + await Task.Yield(); // the session run can take minutes: never on the flush path + var prompt = TriggerPrompts.Snoozed(trigger, messages, score, zone); + if (prompt.Length > TriggerOptions.DigestChars) + { + var name = $"{trigger.Id}-{time.GetUtcNow().ToString("yyyyMMddHHmmss", CultureInfo.InvariantCulture)}.md"; + var relative = Path.Combine(".volcanic", "triggers", "digests", name); + var path = Path.Combine(workspace, relative); + Directory.CreateDirectory(Path.GetDirectoryName(path)!); + await File.WriteAllTextAsync(path, TriggerPrompts.Digest(trigger, messages, zone)).ConfigureAwait(false); + prompt = TriggerPrompts.Snoozed(trigger, messages, score, zone, relative); + } + await delivery.DeliverAsync(owner, prompt, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) + { + logger.LogError(ex, "Delivering the snooze batch of trigger {TriggerId} to chat {Owner} failed", trigger.Id, owner); + } + } + + private void Track(Task task) + { + lock (_lock) + { + _background.RemoveAll(t => t.IsCompleted); + _background.Add(task); + } + } + + private void Log(LogLevel level, GroupMessage message, string excerpt, string results) => + logger.Log(level, "Group \"{GroupTitle}\" ({GroupId}) msg {MessageId} from {Sender}: {Excerpt} → {Results}", + message.GroupTitle, message.GroupId, message.MessageId, message.Sender, excerpt, results); + + private static string Label(WatchingTrigger w) => $"{w.Owner.ToString(CultureInfo.InvariantCulture)}/{w.Trigger.Id} {w.Trigger.Name}"; + + private static string QuestionId(int index) => "q" + index.ToString(CultureInfo.InvariantCulture); + + private static string Excerpt(string text) + { + var flat = text.ReplaceLineEndings(" "); + return "\"" + (flat.Length > MaxExcerpt ? flat[..MaxExcerpt] + "…" : flat) + "\""; + } + + public void Dispose() + { + lock (_lock) + { + _timer?.Dispose(); + _timer = null; + } + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerPrompts.cs b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerPrompts.cs new file mode 100644 index 0000000..88e3393 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Triggers/TriggerPrompts.cs @@ -0,0 +1,141 @@ +using System.Globalization; +using System.Text; +using System.Text.RegularExpressions; +using Volcanic.AgenticCore.Telegram.Groups; + +namespace Volcanic.AgenticCore.Telegram.Triggers; + +/// The texts triggers produce: the owner's prompts, list lines, and the snooze duration syntax. +public static partial class TriggerPrompts +{ + /// The prompt for matches collected in one debounce window. + public static string Live(Trigger trigger, IReadOnlyList<(GroupMessage Message, double Score)> hits, TimeZoneInfo zone) + { + ArgumentNullException.ThrowIfNull(trigger); + ArgumentNullException.ThrowIfNull(hits); + var group = hits.Count > 0 ? hits[^1].Message.GroupTitle : trigger.GroupTitle; + var sb = new StringBuilder().Append(CultureInfo.InvariantCulture, + $"[Trigger \"{trigger.Name}\" ({trigger.Id}) matched {Count(hits.Count, "message")} in \"{group}\"]"); + var format = TimeFormat(hits.Select(h => h.Message), zone); + foreach (var (message, score) in hits) AppendMessage(sb, message, zone, format, score); + return sb.ToString(); + } + + /// + /// The prompt for a snooze batch Jev found related: the whole batch (the agent picks out what matters), or, when + /// is set, where the batch was written instead. + /// + public static string Snoozed(Trigger trigger, IReadOnlyList messages, double score, TimeZoneInfo zone, string? digestPath = null) + { + ArgumentNullException.ThrowIfNull(trigger); + ArgumentNullException.ThrowIfNull(messages); + var group = messages.Count > 0 ? messages[^1].GroupTitle : trigger.GroupTitle; + var since = messages.Count > 0 ? Stamp(messages[0].Time, zone) : ""; + var sb = new StringBuilder().Append(CultureInfo.InvariantCulture, + $"[Trigger \"{trigger.Name}\" ({trigger.Id}): snooze ended; {Count(messages.Count, "message")} from \"{group}\" since {since} contain related ones (score {score.ToString("0.00", CultureInfo.InvariantCulture)})]"); + if (digestPath is not null) + { + sb.Append(CultureInfo.InvariantCulture, $"\nThe messages are in {digestPath}; read it and pick out what matters."); + return sb.ToString(); + } + AppendMessages(sb, messages, zone); + return sb.ToString(); + } + + /// The batch as plain text, for a digest file. + public static string Digest(Trigger trigger, IReadOnlyList messages, TimeZoneInfo zone) + { + ArgumentNullException.ThrowIfNull(trigger); + ArgumentNullException.ThrowIfNull(messages); + var sb = new StringBuilder().Append(CultureInfo.InvariantCulture, $"# Trigger \"{trigger.Name}\" ({trigger.Id}): messages collected while snoozed\n"); + AppendMessages(sb, messages, zone); + return sb.Append('\n').ToString(); + } + + /// One trigger for list and /triggers. + public static string Line(Trigger trigger, TimeZoneInfo zone, DateTimeOffset now, int queued) + { + ArgumentNullException.ThrowIfNull(trigger); + var sb = new StringBuilder().Append(CultureInfo.InvariantCulture, + $"{trigger.Id} \"{trigger.Name}\" · {trigger.GroupTitle} · threshold {trigger.Threshold.ToString("0.00", CultureInfo.InvariantCulture)}"); + if (!trigger.Enabled) sb.Append(" · off"); + if (trigger.IsSnoozed(now) && trigger.SnoozedUntil is { } until) + sb.Append(CultureInfo.InvariantCulture, $" · snoozed until {Stamp(until, zone)}, {queued.ToString(CultureInfo.InvariantCulture)} queued"); + sb.Append("\n asks: ").Append(trigger.Instructions); + if (!string.IsNullOrWhiteSpace(trigger.CriteriaTrue)) sb.Append("\n yes: ").Append(trigger.CriteriaTrue); + if (!string.IsNullOrWhiteSpace(trigger.CriteriaFalse)) sb.Append("\n no: ").Append(trigger.CriteriaFalse); + return sb.ToString(); + } + + /// "Mon 05 Oct 18:00" in . + public static string Stamp(DateTimeOffset at, TimeZoneInfo zone) => + TimeZoneInfo.ConvertTime(at, zone).ToString("ddd dd MMM HH:mm", CultureInfo.InvariantCulture); + + /// 90m, 2h, 1d, 1h30m, 1d 2h; positive only. + public static bool TryParseDuration(string? text, out TimeSpan duration) + { + duration = TimeSpan.Zero; + var match = DurationPattern().Match(text ?? ""); + if (!match.Success || (!match.Groups[1].Success && !match.Groups[2].Success && !match.Groups[3].Success)) return false; + try + { + duration = TimeSpan.FromDays(Number(match.Groups[1])) + TimeSpan.FromHours(Number(match.Groups[2])) + TimeSpan.FromMinutes(Number(match.Groups[3])); + } + catch (OverflowException) + { + return false; + } + return duration > TimeSpan.Zero; + + static double Number(Group g) => g.Success ? double.Parse(g.Value, CultureInfo.InvariantCulture) : 0; + } + + /// An ISO 8601 date-time; without an offset it is local time in . + public static bool TryParseUntil(string? text, TimeZoneInfo zone, out DateTimeOffset until) + { + ArgumentNullException.ThrowIfNull(zone); + until = default; + var value = (text ?? "").Trim(); + if (value.Length == 0) return false; + if (OffsetSuffix().IsMatch(value)) + return DateTimeOffset.TryParse(value, CultureInfo.InvariantCulture, DateTimeStyles.None, out until); + if (!DateTime.TryParse(value, CultureInfo.InvariantCulture, DateTimeStyles.None, out var local)) return false; + try + { + until = TimeZoneInfo.ConvertTimeToUtc(DateTime.SpecifyKind(local, DateTimeKind.Unspecified), zone); + return true; + } + catch (ArgumentException) // a local time skipped by a DST change + { + return false; + } + } + + private static void AppendMessages(StringBuilder sb, IReadOnlyList messages, TimeZoneInfo zone) + { + var format = TimeFormat(messages, zone); + foreach (var message in messages) AppendMessage(sb, message, zone, format, null); + } + + private static void AppendMessage(StringBuilder sb, GroupMessage message, TimeZoneInfo zone, string format, double? score) + { + var time = TimeZoneInfo.ConvertTime(message.Time, zone).ToString(format, CultureInfo.InvariantCulture); + sb.Append('\n').Append(time).Append(' ').Append(message.Sender); + if (score is { } s) sb.Append(", score ").Append(s.ToString("0.00", CultureInfo.InvariantCulture)); + sb.Append(": ").Append(message.Text); + if (message.ReplyTo is { } reply) sb.Append("\n in reply to ").Append(reply.Sender).Append(": ").Append(reply.Text ?? "(no text)"); + if (message.Link is { } link) sb.Append("\n ").Append(link); + } + + /// "HH:mm" when every message is from one local day, else with the date. + private static string TimeFormat(IEnumerable messages, TimeZoneInfo zone) => + messages.Select(m => TimeZoneInfo.ConvertTime(m.Time, zone).Date).Distinct().Skip(1).Any() ? "dd MMM HH:mm" : "HH:mm"; + + private static string Count(int n, string noun) => n.ToString(CultureInfo.InvariantCulture) + " " + noun + (n == 1 ? "" : "s"); + + [GeneratedRegex(@"^\s*(?:(\d{1,4})\s*d)?\s*(?:(\d{1,5})\s*h)?\s*(?:(\d{1,6})\s*m(?:in)?)?\s*$", RegexOptions.IgnoreCase | RegexOptions.CultureInvariant)] + private static partial Regex DurationPattern(); + + [GeneratedRegex(@"(?:Z|[+-]\d{2}:?\d{2})$", RegexOptions.IgnoreCase | RegexOptions.CultureInvariant)] + private static partial Regex OffsetSuffix(); +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/TriggerFakes.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/TriggerFakes.cs new file mode 100644 index 0000000..bc25a6e --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/TriggerFakes.cs @@ -0,0 +1,63 @@ +using System.Collections.Concurrent; +using System.Text.Json.Nodes; +using Microsoft.Extensions.Logging; +using Volcanic.AgenticCore.Telegram.Triggers; + +namespace Volcanic.AgenticCore.Telegram.Tests.Support; + +/// Jev without the network: decides each answer; every call is recorded. +public sealed class FakeJev : IJevClient +{ + public sealed record Call(JsonNode State, IReadOnlyList Questions); + + private readonly ConcurrentQueue _calls = new(); + + public IReadOnlyList Calls => [.. _calls]; + + /// The yes-probability for (state JSON, question). Default: 0.95 when the state mentions "deploy", else 0.05. + public Func Score { get; set; } = + (state, _) => state.Contains("deploy", StringComparison.OrdinalIgnoreCase) ? 0.95 : 0.05; + + /// When set, every call fails with it. + public JevException? Failure { get; set; } + + public Task> AskAsync(JsonNode state, IReadOnlyList questions, CancellationToken cancellationToken) + { + _calls.Enqueue(new Call(state.DeepClone(), questions)); + if (Failure is { } failure) throw failure; + var json = state.ToJsonString(); + IReadOnlyDictionary answers = questions.ToDictionary(q => q.Id, q => Score(json, q)); + return Task.FromResult(answers); + } +} + +/// Records trigger prompts per owner instead of running a session. +public sealed class RecordingDelivery(TimeZoneInfo? zone = null) : ITriggerDelivery +{ + private readonly ConcurrentQueue<(long Owner, string Prompt)> _prompts = new(); + + public IReadOnlyList<(long Owner, string Prompt)> Prompts => [.. _prompts]; + + public Task DeliverAsync(long owner, string prompt, CancellationToken cancellationToken) + { + _prompts.Enqueue((owner, prompt)); + return Task.CompletedTask; + } + + public TimeZoneInfo TimeZoneFor(long owner) => zone ?? TimeZoneInfo.Utc; +} + +/// Keeps every formatted log line. +public sealed class ListLogger : ILogger +{ + private readonly ConcurrentQueue<(LogLevel Level, string Text)> _entries = new(); + + public IReadOnlyList<(LogLevel Level, string Text)> Entries => [.. _entries]; + + public IDisposable? BeginScope(TState state) where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) => + _entries.Enqueue((logLevel, formatter(state, exception))); +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/TriggerEngineTests.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/TriggerEngineTests.cs new file mode 100644 index 0000000..85f77aa --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/TriggerEngineTests.cs @@ -0,0 +1,252 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Time.Testing; +using Volcanic.AgenticCore.Telegram.Groups; +using Volcanic.AgenticCore.Telegram.Tests.Support; +using Volcanic.AgenticCore.Telegram.Triggers; + +namespace Volcanic.AgenticCore.Telegram.Tests; + +/// Classification, logging, the debounce window and snoozes, with a fake Jev and a recording delivery. +public sealed class TriggerEngineTests : IAsyncDisposable +{ + private const long Owner = 1001; + private const long OtherOwner = 2002; + private const long Group = -1001234567890; + private static readonly DateTimeOffset Start = new(2026, 10, 5, 12, 0, 0, TimeSpan.Zero); + + private readonly string _root = Path.Combine(Path.GetTempPath(), "volcanic-engine", Guid.NewGuid().ToString("N")); + private readonly FakeTimeProvider _time = new(Start); + private readonly FakeJev _jev = new(); + private readonly RecordingDelivery _delivery = new(); + private readonly ListLogger _log = new(); + private readonly TriggerOptions _options = new() { ApiKey = "k", Debounce = TimeSpan.FromSeconds(10) }; + private readonly TriggerStore _store; + private readonly TriggerDispatcher _dispatcher; + private readonly TriggerEngine _engine; + + public TriggerEngineTests() + { + _store = new TriggerStore(Path.Combine(_root, "workspaces"), 20, NullLogger.Instance, _time); + _dispatcher = new TriggerDispatcher(_delivery, _options, _time, NullLogger.Instance); + _engine = new TriggerEngine(_store, new SnoozeQueue(100), _jev, _dispatcher, _delivery, _time, _log); + } + + public async ValueTask DisposeAsync() + { + await _engine.IdleAsync(); + _engine.Dispose(); + _dispatcher.Dispose(); + try { Directory.Delete(_root, true); } catch (IOException) { } + } + + private Task AddAsync(long owner = Owner, string name = "deploys", double threshold = 0.7) => + _store.AddAsync(owner, new NewTrigger(name, Group, "Dev Team", "Is this about deploys?", "Mentions deploys", "Small talk", threshold), CancellationToken.None); + + private static GroupMessage Message(string? text, int id = 1, int minute = 2) => + new(Group, "Dev Team", null, true, id, Start.AddMinutes(minute), "Alice (@alice)", text); + + private async Task AdvanceAsync(TimeSpan by) + { + _time.Advance(by); + await _engine.IdleAsync(); + } + + [Fact] + public async Task A_matching_message_reaches_the_owner_once_the_window_closes() + { + var trigger = await AddAsync(); + + await _engine.ProcessAsync(Message("prod deploy is failing again"), CancellationToken.None); + await AdvanceAsync(TimeSpan.FromSeconds(9)); + Assert.Empty(_delivery.Prompts); + await AdvanceAsync(TimeSpan.FromSeconds(1)); + + var (owner, prompt) = Assert.Single(_delivery.Prompts); + Assert.Equal(Owner, owner); + Assert.Equal($"[Trigger \"deploys\" ({trigger.Id}) matched 1 message in \"Dev Team\"]\n12:02 Alice (@alice), score 0.95: prod deploy is failing again\n https://t.me/c/1234567890/1", prompt); + var question = Assert.Single(Assert.Single(_jev.Calls).Questions); + Assert.Equal(("Is this about deploys?", "Mentions deploys", "Small talk"), (question.Instructions, question.CriteriaTrue, question.CriteriaFalse)); + Assert.Contains(_log.Entries, e => e.Text == "Group \"Dev Team\" (-1001234567890) msg 1 from Alice (@alice): \"prod deploy is failing again\" → 1001/t1 deploys=0.95 MATCH"); + } + + [Fact] + public async Task Matches_in_one_window_become_one_prompt_and_later_ones_do_not_extend_it() + { + await AddAsync(); + + await _engine.ProcessAsync(Message("deploy started", 1), CancellationToken.None); + await AdvanceAsync(TimeSpan.FromSeconds(6)); + await _engine.ProcessAsync(Message("lunch?", 2), CancellationToken.None); + await _engine.ProcessAsync(Message("deploy done", 3), CancellationToken.None); + await AdvanceAsync(TimeSpan.FromSeconds(4)); // 10 s after the first match + + var prompt = Assert.Single(_delivery.Prompts).Prompt; + Assert.StartsWith("[Trigger \"deploys\" (t1) matched 2 messages", prompt, StringComparison.Ordinal); + Assert.Contains("deploy started", prompt, StringComparison.Ordinal); + Assert.Contains("deploy done", prompt, StringComparison.Ordinal); + Assert.DoesNotContain("lunch", prompt, StringComparison.Ordinal); + Assert.Contains(_log.Entries, e => e.Text.EndsWith("\"lunch?\" → 1001/t1 deploys=0.05", StringComparison.Ordinal)); + } + + [Fact] + public async Task Every_owners_trigger_on_the_group_is_asked_in_one_call() + { + await AddAsync(); + await AddAsync(OtherOwner, "outages", threshold: 0.99); + + await _engine.ProcessAsync(Message("deploy broke prod"), CancellationToken.None); + await AdvanceAsync(TimeSpan.FromSeconds(10)); + + Assert.Equal(2, Assert.Single(_jev.Calls).Questions.Count); + Assert.Equal(Owner, Assert.Single(_delivery.Prompts).Owner); // 0.95 is under the other trigger's 0.99 + } + + [Fact] + public async Task Messages_without_text_or_triggers_are_logged_but_not_classified() + { + await _engine.ProcessAsync(Message("deploy"), CancellationToken.None); + await AddAsync(); + await _engine.ProcessAsync(Message(null, 2), CancellationToken.None); + + Assert.Empty(_jev.Calls); + Assert.Contains(_log.Entries, e => e.Text.EndsWith("\"deploy\" → no triggers", StringComparison.Ordinal)); + Assert.Contains(_log.Entries, e => e.Text.EndsWith("msg 2 from Alice (@alice): (no text) → skipped", StringComparison.Ordinal)); + } + + [Fact] + public async Task A_jev_failure_is_logged_as_a_warning_and_drops_the_message() + { + await AddAsync(); + _jev.Failure = new JevException("Jev: HTTP 503 busy") { Attempts = 4 }; + + await _engine.ProcessAsync(Message("deploy"), CancellationToken.None); + await AdvanceAsync(TimeSpan.FromSeconds(10)); + + Assert.Empty(_delivery.Prompts); + Assert.Contains(_log.Entries, e => e.Level == LogLevel.Warning && e.Text.EndsWith("→ Jev failed after 4 attempts: Jev: HTTP 503 busy", StringComparison.Ordinal)); + } + + [Fact] + public async Task A_snoozed_trigger_queues_and_its_batch_goes_to_jev_once_when_the_snooze_ends() + { + await AddAsync(); + await _engine.StartAsync(CancellationToken.None); + await _engine.SnoozeAsync(Owner, null, Start.AddHours(1), CancellationToken.None); + + await _engine.ProcessAsync(Message("anyone for pizza?", 1, 5), CancellationToken.None); + await _engine.ProcessAsync(Message("prod is down after the deploy", 2, 20), CancellationToken.None); + Assert.Empty(_jev.Calls); + Assert.Equal(2, _engine.Queued(Owner, "t1")); + Assert.Contains(_log.Entries, e => e.Text.EndsWith("→ 1001/t1 deploys snoozed (queued 2)", StringComparison.Ordinal)); + + await AdvanceAsync(TimeSpan.FromHours(1)); + + var call = Assert.Single(_jev.Calls); + Assert.StartsWith("Answer yes if at least one message in the batch passes this check: ", Assert.Single(call.Questions).Instructions, StringComparison.Ordinal); + Assert.Equal(2, call.State["messages"]!.AsArray().Count); + var prompt = Assert.Single(_delivery.Prompts).Prompt; + Assert.StartsWith("[Trigger \"deploys\" (t1): snooze ended; 2 messages from \"Dev Team\" since Mon 05 Oct 12:05 contain related ones (score 0.95)]", prompt, StringComparison.Ordinal); + Assert.Contains("12:05 Alice (@alice): anyone for pizza?", prompt, StringComparison.Ordinal); + Assert.Null(_store.Find(Owner, "t1")!.SnoozedUntil); + Assert.Equal(0, _engine.Queued(Owner, "t1")); + + await _engine.ProcessAsync(Message("deploy again", 3, 70), CancellationToken.None); // live again + Assert.Equal(2, _jev.Calls.Count); + } + + [Fact] + public async Task An_unrelated_snooze_batch_stays_silent() + { + await AddAsync(); + await _engine.StartAsync(CancellationToken.None); + await _engine.SnoozeAsync(Owner, "t1", Start.AddMinutes(30), CancellationToken.None); + await _engine.ProcessAsync(Message("pizza", 1), CancellationToken.None); + + await AdvanceAsync(TimeSpan.FromMinutes(30)); + + Assert.Single(_jev.Calls); + Assert.Empty(_delivery.Prompts); + Assert.Contains(_log.Entries, e => e.Text.EndsWith("snooze ended, 1 messages, best 0.05 → no match", StringComparison.Ordinal)); + } + + [Fact] + public async Task Unsnooze_flushes_now_and_reports_the_result() + { + await AddAsync(); + await _engine.SnoozeAsync(Owner, null, Start.AddDays(1), CancellationToken.None); + await _engine.ProcessAsync(Message("deploy", 1), CancellationToken.None); + + var flush = Assert.Single(await _engine.UnsnoozeAsync(Owner, null, CancellationToken.None)); + await _engine.IdleAsync(); + + Assert.Equal((1, true), (flush.Count, flush.Matched)); + Assert.Single(_delivery.Prompts); + } + + [Fact] + public async Task A_snooze_that_ended_while_the_host_was_down_is_flushed_at_start() + { + await AddAsync(); + await _engine.SnoozeAsync(Owner, null, Start.AddMinutes(-1), CancellationToken.None); // already over + await new SnoozeQueue(10).AppendAsync(_store.WorkspaceOf(Owner), "t1", Message("deploy"), CancellationToken.None); + + await _engine.StartAsync(CancellationToken.None); + await _engine.IdleAsync(); + + Assert.Single(_delivery.Prompts); + } + + [Fact] + public async Task A_large_batch_is_chunked_and_delivered_as_a_digest_file() + { + await AddAsync(); + await _engine.SnoozeAsync(Owner, null, Start.AddDays(1), CancellationToken.None); + var text = new string('x', 1000); + for (var i = 1; i <= 60; i++) await _engine.ProcessAsync(Message(i == 60 ? "deploy " + text : text, i), CancellationToken.None); + + await _engine.UnsnoozeAsync(Owner, "t1", CancellationToken.None); + await _engine.IdleAsync(); + + Assert.Equal(2, _jev.Calls.Count); // 60 KB of state in 40 KB chunks + var prompt = Assert.Single(_delivery.Prompts).Prompt; + var relative = prompt[(prompt.IndexOf("The messages are in ", StringComparison.Ordinal) + 20)..prompt.IndexOf("; read it", StringComparison.Ordinal)]; + var digest = await File.ReadAllTextAsync(Path.Combine(_store.WorkspaceOf(Owner), relative)); + Assert.Contains("deploy xxx", digest, StringComparison.Ordinal); + } + + [Fact] + public async Task Deleting_a_trigger_deletes_its_queue_and_a_disabled_trigger_ignores_messages() + { + await AddAsync(); + await AddAsync(name: "other"); + await _store.UpdateAsync(Owner, "t2", t => t with { Enabled = false }, CancellationToken.None); + await _engine.SnoozeAsync(Owner, "t1", Start.AddDays(1), CancellationToken.None); + await _engine.ProcessAsync(Message("deploy"), CancellationToken.None); + + await _engine.DeleteAsync(Owner, "t1", CancellationToken.None); + await _engine.ProcessAsync(Message("deploy", 2), CancellationToken.None); + + Assert.False(File.Exists(SnoozeQueue.PathFor(_store.WorkspaceOf(Owner), "t1"))); + Assert.Empty(_jev.Calls); + Assert.Contains(_log.Entries, e => e.Text.EndsWith("\"deploy\" → no triggers", StringComparison.Ordinal)); + } + + [Fact] + public void Durations_and_until_times_parse() + { + Assert.True(TriggerPrompts.TryParseDuration("1h30m", out var d) && d == TimeSpan.FromMinutes(90)); + Assert.True(TriggerPrompts.TryParseDuration("2d", out d) && d == TimeSpan.FromDays(2)); + Assert.True(TriggerPrompts.TryParseDuration("45 min", out d) && d == TimeSpan.FromMinutes(45)); + Assert.False(TriggerPrompts.TryParseDuration("soon", out _)); + Assert.False(TriggerPrompts.TryParseDuration("0m", out _)); + Assert.False(TriggerPrompts.TryParseDuration("", out _)); + + var berlin = TimeZoneInfo.FindSystemTimeZoneById("Europe/Berlin"); + Assert.True(TriggerPrompts.TryParseUntil("2026-10-05T18:00", berlin, out var until)); + Assert.Equal(new DateTimeOffset(2026, 10, 5, 16, 0, 0, TimeSpan.Zero), until); + Assert.True(TriggerPrompts.TryParseUntil("2026-10-05T18:00Z", berlin, out until)); + Assert.Equal(new DateTimeOffset(2026, 10, 5, 18, 0, 0, TimeSpan.Zero), until); + Assert.False(TriggerPrompts.TryParseUntil("tonight", berlin, out _)); + } +}