forked from admin/VolcanicAgenticCore
301 lines
13 KiB
C#
301 lines
13 KiB
C#
using System.Globalization;
|
|
using Microsoft.Extensions.Logging;
|
|
using Volcanic.AgenticCore.Telegram.Groups;
|
|
|
|
namespace Volcanic.AgenticCore.Telegram.Triggers;
|
|
|
|
/// <summary>What ending one trigger's snooze did. <see cref="Error"/> is set when Jev failed (the batch is lost).</summary>
|
|
public sealed record SnoozeFlush(Trigger Trigger, int Count, double? Score, bool Matched, string? Error = null);
|
|
|
|
/// <summary>
|
|
/// 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
|
|
/// <see cref="TriggerDispatcher"/>; snoozed triggers queue the message instead (<see cref="SnoozeQueue"/>). When a
|
|
/// snooze ends (a timer on <see cref="TimeProvider"/>, 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 (<see cref="JevStates.BatchQuestion"/>), and a
|
|
/// match becomes one prompt with the whole batch.
|
|
/// </summary>
|
|
public sealed class TriggerEngine(
|
|
TriggerStore store,
|
|
SnoozeQueue queue,
|
|
IJevClient jev,
|
|
TriggerDispatcher dispatcher,
|
|
ITriggerDelivery delivery,
|
|
TimeProvider time,
|
|
ILogger<TriggerEngine> 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<Task> _background = [];
|
|
private ITimer? _timer;
|
|
|
|
public TriggerStore Store => store;
|
|
|
|
public DateTimeOffset Now => time.GetUtcNow();
|
|
|
|
/// <summary>Messages queued for a snoozed trigger.</summary>
|
|
public int Queued(long owner, string triggerId) => queue.Count(store.WorkspaceOf(owner), triggerId);
|
|
|
|
/// <summary>Flushes snoozes that ended while the host was down, then arms the timer.</summary>
|
|
public async Task StartAsync(CancellationToken cancellationToken)
|
|
{
|
|
lock (_lock) _timer ??= time.CreateTimer(_ => OnTimer(), null, Timeout.InfiniteTimeSpan, Timeout.InfiniteTimeSpan);
|
|
await FlushDueAsync(cancellationToken).ConfigureAwait(false);
|
|
}
|
|
|
|
/// <summary>Completes when every background flush and snooze prompt started so far has finished, and the dispatcher's too.</summary>
|
|
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<string>();
|
|
var live = new List<WatchingTrigger>();
|
|
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<string, double> 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));
|
|
}
|
|
|
|
/// <summary>Snoozes one trigger (or, with no id, every trigger of the owner) until <paramref name="until"/>.</summary>
|
|
public async Task<IReadOnlyList<Trigger>> SnoozeAsync(long owner, string? id, DateTimeOffset until, CancellationToken cancellationToken)
|
|
{
|
|
var targets = Targets(owner, id, onlySnoozed: false);
|
|
var snoozed = new List<Trigger>();
|
|
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;
|
|
}
|
|
|
|
/// <summary>Ends the snooze of one trigger (or every snoozed trigger of the owner) now and flushes its queue.</summary>
|
|
public async Task<IReadOnlyList<SnoozeFlush>> UnsnoozeAsync(long owner, string? id, CancellationToken cancellationToken)
|
|
{
|
|
var results = new List<SnoozeFlush>();
|
|
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;
|
|
}
|
|
|
|
/// <summary>Removes the trigger and its snooze queue.</summary>
|
|
public async Task<Trigger?> 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;
|
|
}
|
|
|
|
/// <summary>Re-arms the snooze timer for the soonest snooze end (at most an hour away; it re-arms when it fires).</summary>
|
|
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<Trigger> 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<SnoozeFlush> 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<GroupMessage> 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;
|
|
}
|
|
}
|
|
}
|