using System.Text.Json; using System.Threading.Channels; using Microsoft.Extensions.Logging; using Volcanic.AgenticCore.Abstractions.Events; using Volcanic.AgenticCore.Abstractions.Messages; using Volcanic.AgenticCore.Abstractions.Providers; using Volcanic.AgenticCore.Abstractions.Tools; using Volcanic.AgenticCore.Abstractions.Ui; using static Volcanic.AgenticCore.Abstractions.Events.AgentSessionEvent; namespace Volcanic.AgenticCore.Telegram.Rendering; /// /// Renders one chat's session events. Session handlers run on the agent loop, so events are queued into a channel and /// rendered by a single task. Each assistant message is sent once, complete, when it ends (split into several Telegram /// messages past the size limit); "typing…" repeats while the agent runs. /// /// Host work (file uploads, notices) runs on the same loop through , so it stays in /// order with the rendered output queued before it. /// /// /// While a run is active one status message ("⚙️ Working...") lists the last tool calls /// with their state, edited at most every . When something was sent below it, /// the next tool update moves it back to the bottom; it is deleted when the run settles. /// /// public sealed class ChatRenderer : IAsyncDisposable { public const int StatusToolCount = 5; private const int StatusArgsLength = 80; private readonly long _chatId; private readonly Func? _deliverTo; private readonly TelegramSender _sender; private readonly TelegramOptions _options; private readonly ILogger _logger; private readonly TimeProvider _time; private abstract record RenderItem; private sealed record EventItem(AgentSessionEvent Event) : RenderItem; private sealed record WorkItem(Func Run, Action Cancel) : RenderItem; private readonly Channel _items = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true }); private readonly CancellationTokenSource _stop = new(); private readonly Task _loop; /// Delays before resending a reply chunk that failed (1 s, 3 s). Settable for tests. internal IReadOnlyList ReplyRetryDelays { get; set; } = [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(3)]; private DateTimeOffset _lastFlush = DateTimeOffset.MinValue; private CancellationTokenSource? _typing; private int _pending; // Status message state; touched only by the render loop. private enum ToolState { Running, Done, Failed } private sealed record ToolLine(string Id, string Name, string Args) { public ToolState State { get; set; } } private readonly List _recentTools = []; private bool _running; private int? _statusMessageId; private long _statusChatId; private string? _statusSent; private bool _statusDirty; private bool _statusBuried; private bool Dirty => _statusDirty; /// Where output goes, read at each send; null = (see ). public ChatRenderer(long chatId, TelegramSender sender, TelegramOptions options, ILogger logger, TimeProvider? time = null, Func? deliverTo = null) { _chatId = chatId; _deliverTo = deliverTo; _sender = sender; _options = options; _logger = logger; _time = time ?? TimeProvider.System; _loop = Task.Run(RunAsync); } /// The Telegram chat this chat's output goes to now: its own, or an admin's while that admin impersonates it. public long DeliveryChatId => _deliverTo?.Invoke() ?? _chatId; /// Queue an event; safe to call from any thread (this is the session subscription). public void Post(AgentSessionEvent e) => Enqueue(new EventItem(e)); private bool Enqueue(RenderItem item) { Interlocked.Increment(ref _pending); if (_items.Writer.TryWrite(item)) return true; Interlocked.Decrement(ref _pending); return false; } /// /// Run host work (an upload, a notice) on the render loop, after every event queued before it, so it stays in order /// with rendered output. Afterwards the status message counts as buried. Cancelled when the renderer is disposed first. /// cancels the returned task at once, even while the work is still queued behind /// a slow upload; work that hasn't started by then is skipped, and running work sees it through its token. /// public Task RunInOrderAsync(Func> work, CancellationToken cancellationToken = default) { ArgumentNullException.ThrowIfNull(work); if (cancellationToken.IsCancellationRequested) return Task.FromCanceled(cancellationToken); var done = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); var registration = cancellationToken.Register(() => done.TrySetCanceled(cancellationToken)); var item = new WorkItem(async ct => { try { if (done.Task.IsCompleted) return; // cancelled while queued using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, cancellationToken); done.TrySetResult(await work(linked.Token).ConfigureAwait(false)); } catch (OperationCanceledException) { done.TrySetCanceled(); } catch (Exception ex) { done.TrySetException(ex); } finally { await registration.DisposeAsync().ConfigureAwait(false); } }, () => { registration.Dispose(); done.TrySetCanceled(); }); if (!Enqueue(item)) { registration.Dispose(); done.TrySetCanceled(); } return done.Task; } /// Plain text (no parse mode) in order with rendered output; cut to 4096 characters by the sender. public Task SayAsync(string text, CancellationToken cancellationToken = default) => RunInOrderAsync(ct => _sender.SendTextAsync(DeliveryChatId, text, cancellationToken: ct), cancellationToken); /// /// A Markdown reply rendered like an assistant message (HTML, split past 4096, each chunk retried), in order with /// rendered output. Returns the first message id and whether every chunk was delivered. /// internal Task<(int? FirstId, bool Complete)> ReplyInOrderAsync(string markdown, CancellationToken cancellationToken) => RunInOrderAsync(ct => SendReplyAsync(markdown, ct), cancellationToken); /// Completes once every queued event is rendered. For tests. internal async Task DrainAsync() { while (Volatile.Read(ref _pending) > 0 || Dirty) await Task.Delay(10).ConfigureAwait(false); } private async Task RunAsync() { var reader = _items.Reader; var ct = _stop.Token; try { while (true) { if (Dirty) { var wait = _lastFlush + _options.EditInterval - _time.GetUtcNow(); if (wait <= TimeSpan.Zero) { await FlushAsync(ct).ConfigureAwait(false); continue; } using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct); timeout.CancelAfter(wait); try { if (!await reader.WaitToReadAsync(timeout.Token).ConfigureAwait(false)) break; } catch (OperationCanceledException) when (!ct.IsCancellationRequested) { continue; // edit interval elapsed: flush on the next iteration } } else if (!await reader.WaitToReadAsync(ct).ConfigureAwait(false)) { break; } while (reader.TryRead(out var item)) { try { switch (item) { case EventItem { Event: var e }: await RenderAsync(e, ct).ConfigureAwait(false); break; case WorkItem work when ct.IsCancellationRequested: // disposal gave up waiting work.Cancel(); break; case WorkItem work: await work.Run(ct).ConfigureAwait(false); BuryStatus(); break; } } catch (Exception ex) when (ex is not OperationCanceledException) { _logger.LogWarning(ex, "Rendering {Item} for chat {ChatId} failed", item is EventItem ev ? ev.Event.GetType().Name : "work", _chatId); } finally { Interlocked.Decrement(ref _pending); } } } await DeleteStatusAsync(ct).ConfigureAwait(false); } catch (OperationCanceledException) { } finally { StopTyping(); while (_items.Reader.TryRead(out var left)) { if (left is WorkItem work) work.Cancel(); Interlocked.Decrement(ref _pending); } } } private async Task RenderAsync(AgentSessionEvent e, CancellationToken ct) { switch (e) { case AgentStart: StartTyping(); if (!_running) { _running = true; _recentTools.Clear(); await FlushStatusAsync(ct).ConfigureAwait(false); } break; case AgentSettled: StopTyping(); await DeleteStatusAsync(ct).ConfigureAwait(false); break; case MessageEnd { Message: AssistantMessage message }: _ = await SendReplyAsync(message.Text, ct).ConfigureAwait(false); if (message.StopReason == StopReason.Error) await SendLineAsync($"❌ {message.ErrorMessage ?? "The model request failed."}", ct).ConfigureAwait(false); else if (message.StopReason == StopReason.Aborted) await SendLineAsync("⏹ Stopped.", ct).ConfigureAwait(false); break; case ToolExecutionStart start: _recentTools.Add(new ToolLine(start.ToolCallId, start.ToolName, FormatArgs(start.Args, StatusArgsLength))); if (_recentTools.Count > StatusToolCount) _recentTools.RemoveRange(0, _recentTools.Count - StatusToolCount); _statusDirty = true; break; case ToolExecutionEnd end: if (_recentTools.Find(t => t.Id == end.ToolCallId) is { } tool) { tool.State = end.IsError ? ToolState.Failed : ToolState.Done; _statusDirty = true; } break; case AutoRetryStart retry: await SendLineAsync($"⏳ {retry.ErrorMessage} — retrying in {retry.Delay.TotalSeconds:0} s ({retry.Attempt}/{retry.MaxAttempts})", ct).ConfigureAwait(false); break; case AutoRetryEnd { Success: false } retryEnd: await SendLineAsync($"❌ Giving up after {retryEnd.Attempt} attempts: {retryEnd.FinalError}", ct).ConfigureAwait(false); break; case CompactionStart: await SendLineAsync("🗜 Compacting the conversation…", ct).ConfigureAwait(false); break; case CompactionEnd { ErrorMessage: { } error }: await SendLineAsync($"❌ Compaction failed: {error}", ct).ConfigureAwait(false); break; case CompactionEnd { Result: { } result }: await SendLineAsync($"🗜 Compacted: ~{result.TokensBefore:N0} → ~{result.EstimatedTokensAfter:N0} tokens.", ct).ConfigureAwait(false); break; case Notification notification: await SendLineAsync($"{Icon(notification.Kind)} {notification.Message}", ct).ConfigureAwait(false); break; case ExtensionError error: _logger.LogWarning("Extension {Extension} failed on {Event}: {Error}", error.ExtensionName, error.Event, error.Error); await SendLineAsync($"⚠️ {error.ExtensionName}: {error.Error}", ct).ConfigureAwait(false); break; } } private async Task FlushAsync(CancellationToken ct) { _lastFlush = _time.GetUtcNow(); if (_statusDirty) await FlushStatusAsync(ct).ConfigureAwait(false); } /// /// Send a finished assistant message. A chunk that fails to send is retried a couple of times. Returns the first /// message id and whether every chunk was delivered. /// private async Task<(int? FirstId, bool Complete)> SendReplyAsync(string text, CancellationToken ct) { if (string.IsNullOrWhiteSpace(text)) return (null, true); int? first = null; var complete = true; var target = DeliveryChatId; foreach (var chunk in TelegramFormatter.Split(text)) { var html = TelegramFormatter.ToHtml(chunk); var sent = await _sender.SendHtmlAsync(target, html, chunk, cancellationToken: ct).ConfigureAwait(false); foreach (var delay in ReplyRetryDelays) { if (sent is not null) break; await Task.Delay(delay, ct).ConfigureAwait(false); sent = await _sender.SendHtmlAsync(target, html, chunk, cancellationToken: ct).ConfigureAwait(false); } if (sent is null) { _logger.LogError("Could not deliver part of a reply to chat {ChatId}", _chatId); complete = false; } else { first ??= sent; BuryStatus(); } } return (first, complete); } private async Task SendLineAsync(string text, CancellationToken ct) { if (await _sender.SendHtmlAsync(DeliveryChatId, TelegramFormatter.Escape(text), text, cancellationToken: ct).ConfigureAwait(false) is not null) BuryStatus(); } /// /// Something was sent below the status message: the next status update moves it back to the bottom. Not moved right /// away, since after the final answer the run settles and the status is deleted anyway. /// private void BuryStatus() { if (_statusMessageId is not null) _statusBuried = true; } /// Send, move or edit the status message to show the current tool list. private async Task FlushStatusAsync(CancellationToken ct) { _statusDirty = false; if (!_running) return; var html = StatusHtml(); var target = DeliveryChatId; // Moved to the bottom when buried, and to the other chat when the delivery chat changed (impersonation). if ((_statusBuried || _statusChatId != target) && _statusMessageId is { } buried) { await _sender.DeleteAsync(_statusChatId, buried, ct).ConfigureAwait(false); _statusMessageId = null; } _statusBuried = false; if (_statusMessageId is not { } id) { _statusChatId = target; _statusMessageId = await _sender.SendHtmlAsync(target, html, StripTags(html), cancellationToken: ct).ConfigureAwait(false); _statusSent = html; } else if (html != _statusSent && await _sender.EditHtmlAsync(_statusChatId, id, html, StripTags(html), ct).ConfigureAwait(false)) { _statusSent = html; } } private async Task DeleteStatusAsync(CancellationToken ct) { _running = false; _statusDirty = false; _statusBuried = false; _recentTools.Clear(); if (_statusMessageId is { } id) await _sender.DeleteAsync(_statusChatId, id, ct).ConfigureAwait(false); _statusMessageId = null; _statusSent = null; } private string StatusHtml() { var html = new System.Text.StringBuilder("⚙️ Working..."); foreach (var tool in _recentTools) { var icon = tool.State switch { ToolState.Done => "✓", ToolState.Failed => "✗", _ => "▶" }; html.Append('\n').Append(icon).Append(" ").Append(TelegramFormatter.Escape(tool.Name)).Append("").Append(tool.Args); } return html.ToString(); } private void StartTyping() { if (_typing is not null) return; var cts = CancellationTokenSource.CreateLinkedTokenSource(_stop.Token); _typing = cts; _ = Task.Run(async () => { try { while (!cts.IsCancellationRequested) { await _sender.SendTypingAsync(DeliveryChatId, cts.Token).ConfigureAwait(false); await Task.Delay(_options.TypingInterval, cts.Token).ConfigureAwait(false); } } catch (OperationCanceledException) { } }); } private void StopTyping() { var typing = Interlocked.Exchange(ref _typing, null); typing?.Cancel(); typing?.Dispose(); } /// ": ls -la" for bash, the most telling argument otherwise. internal static string FormatArgs(JsonElement args, int max = 200) { if (args.ValueKind != JsonValueKind.Object) return ""; string? summary = null; foreach (var key in new[] { "command", "path", "pattern", "file_path" }) { if (args.TryGetProperty(key, out var value) && value.ValueKind == JsonValueKind.String) { summary = value.GetString(); break; } } summary ??= args.EnumerateObject().Any() ? args.GetRawText() : null; if (string.IsNullOrWhiteSpace(summary)) return ""; return $": {TelegramFormatter.Escape(Shorten(summary.ReplaceLineEndings(" "), max))}"; } private static string Shorten(string text, int max) => text.Length <= max ? text : text[..max] + "…"; private static string StripTags(string html) => System.Net.WebUtility.HtmlDecode(System.Text.RegularExpressions.Regex.Replace(html, "<[^>]+>", "")); private static string Icon(NotifyKind kind) => kind switch { NotifyKind.Warning => "⚠️", NotifyKind.Error => "❌", _ => "ℹ️" }; public async ValueTask DisposeAsync() { _items.Writer.TryComplete(); try { await _loop.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false); } catch (TimeoutException) { } await _stop.CancelAsync().ConfigureAwait(false); StopTyping(); _stop.Dispose(); } }