forked from admin/VolcanicAgenticCore
310 lines
12 KiB
C#
310 lines
12 KiB
C#
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;
|
||
|
||
/// <summary>
|
||
/// 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: streamed text edits one message (split into several past the size limit) at most every
|
||
/// <see cref="TelegramOptions.EditInterval"/>, tool calls become status lines, and "typing…" repeats while the agent runs.
|
||
/// </summary>
|
||
public sealed class ChatRenderer : IAsyncDisposable
|
||
{
|
||
private const int ToolExcerptLength = 300;
|
||
|
||
private readonly long _chatId;
|
||
private readonly TelegramSender _sender;
|
||
private readonly TelegramOptions _options;
|
||
private readonly ILogger _logger;
|
||
private readonly TimeProvider _time;
|
||
private readonly Channel<AgentSessionEvent> _events = Channel.CreateUnbounded<AgentSessionEvent>(new UnboundedChannelOptions { SingleReader = true });
|
||
private readonly CancellationTokenSource _stop = new();
|
||
private readonly Task _loop;
|
||
|
||
// Streaming reply state; touched only by the render loop.
|
||
private readonly List<(int MessageId, string Sent)> _replyMessages = [];
|
||
private readonly Dictionary<string, (int MessageId, string Args)> _toolMessages = [];
|
||
private string _replyText = "";
|
||
private bool _replyDirty;
|
||
private DateTimeOffset _lastFlush = DateTimeOffset.MinValue;
|
||
private CancellationTokenSource? _typing;
|
||
private int _pending;
|
||
|
||
public ChatRenderer(long chatId, TelegramSender sender, TelegramOptions options, ILogger logger, TimeProvider? time = null)
|
||
{
|
||
_chatId = chatId;
|
||
_sender = sender;
|
||
_options = options;
|
||
_logger = logger;
|
||
_time = time ?? TimeProvider.System;
|
||
_loop = Task.Run(RunAsync);
|
||
}
|
||
|
||
/// <summary>Queue an event; safe to call from any thread (this is the session subscription).</summary>
|
||
public void Post(AgentSessionEvent e)
|
||
{
|
||
Interlocked.Increment(ref _pending);
|
||
if (!_events.Writer.TryWrite(e)) Interlocked.Decrement(ref _pending);
|
||
}
|
||
|
||
/// <summary>Completes once every queued event is rendered. For tests.</summary>
|
||
internal async Task DrainAsync()
|
||
{
|
||
while (Volatile.Read(ref _pending) > 0 || _replyDirty) await Task.Delay(10).ConfigureAwait(false);
|
||
}
|
||
|
||
private async Task RunAsync()
|
||
{
|
||
var reader = _events.Reader;
|
||
var ct = _stop.Token;
|
||
try
|
||
{
|
||
while (true)
|
||
{
|
||
if (_replyDirty)
|
||
{
|
||
var wait = _lastFlush + _options.EditInterval - _time.GetUtcNow();
|
||
if (wait <= TimeSpan.Zero)
|
||
{
|
||
await FlushReplyAsync(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 e))
|
||
{
|
||
try
|
||
{
|
||
await RenderAsync(e, ct).ConfigureAwait(false);
|
||
}
|
||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||
{
|
||
_logger.LogWarning(ex, "Rendering {Event} for chat {ChatId} failed", e.GetType().Name, _chatId);
|
||
}
|
||
finally
|
||
{
|
||
Interlocked.Decrement(ref _pending);
|
||
}
|
||
}
|
||
}
|
||
if (_replyDirty) await FlushReplyAsync(ct).ConfigureAwait(false);
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
}
|
||
finally
|
||
{
|
||
StopTyping();
|
||
}
|
||
}
|
||
|
||
private async Task RenderAsync(AgentSessionEvent e, CancellationToken ct)
|
||
{
|
||
switch (e)
|
||
{
|
||
case AgentStart:
|
||
StartTyping();
|
||
break;
|
||
case AgentSettled:
|
||
StopTyping();
|
||
break;
|
||
|
||
case MessageStart { Message: AssistantMessage }:
|
||
ResetReply();
|
||
break;
|
||
case MessageUpdate { AssistantMessageEvent: AssistantMessageEvent.TextDelta } update:
|
||
_replyText = update.Message.Text;
|
||
_replyDirty = true;
|
||
break;
|
||
case MessageEnd { Message: AssistantMessage message }:
|
||
_replyText = message.Text;
|
||
_replyDirty = true;
|
||
await FlushReplyAsync(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);
|
||
ResetReply();
|
||
break;
|
||
|
||
case ToolExecutionStart start:
|
||
// Keep the reply above the tool line in the chat.
|
||
if (_replyDirty) await FlushReplyAsync(ct).ConfigureAwait(false);
|
||
var args = FormatArgs(start.Args);
|
||
if (await SendHtmlLineAsync($"▶ <b>{TelegramFormatter.Escape(start.ToolName)}</b>{args}", ct).ConfigureAwait(false) is { } id)
|
||
_toolMessages[start.ToolCallId] = (id, args);
|
||
break;
|
||
case ToolExecutionEnd end:
|
||
var started = _toolMessages.Remove(end.ToolCallId, out var tool);
|
||
var line = end.IsError
|
||
? $"✗ <b>{TelegramFormatter.Escape(end.ToolName)}</b>{tool.Args}{FormatExcerpt(end.Result)}"
|
||
: $"✓ <b>{TelegramFormatter.Escape(end.ToolName)}</b>{tool.Args}";
|
||
if (started)
|
||
await _sender.EditHtmlAsync(_chatId, tool.MessageId, line, StripTags(line), ct).ConfigureAwait(false);
|
||
else
|
||
await SendHtmlLineAsync(line, ct).ConfigureAwait(false);
|
||
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 FlushReplyAsync(CancellationToken ct)
|
||
{
|
||
_replyDirty = false;
|
||
_lastFlush = _time.GetUtcNow();
|
||
if (string.IsNullOrWhiteSpace(_replyText)) return;
|
||
|
||
var chunks = TelegramFormatter.Split(_replyText);
|
||
for (var i = 0; i < chunks.Count; i++)
|
||
{
|
||
var chunk = chunks[i];
|
||
var html = TelegramFormatter.ToHtml(chunk);
|
||
if (i < _replyMessages.Count)
|
||
{
|
||
if (_replyMessages[i].Sent == chunk) continue;
|
||
if (await _sender.EditHtmlAsync(_chatId, _replyMessages[i].MessageId, html, chunk, ct).ConfigureAwait(false))
|
||
_replyMessages[i] = (_replyMessages[i].MessageId, chunk);
|
||
}
|
||
else if (await _sender.SendHtmlAsync(_chatId, html, chunk, cancellationToken: ct).ConfigureAwait(false) is { } id)
|
||
{
|
||
_replyMessages.Add((id, chunk));
|
||
}
|
||
else
|
||
{
|
||
break; // try the rest on the next flush
|
||
}
|
||
}
|
||
}
|
||
|
||
private void ResetReply()
|
||
{
|
||
_replyMessages.Clear();
|
||
_replyText = "";
|
||
_replyDirty = false;
|
||
_lastFlush = DateTimeOffset.MinValue;
|
||
}
|
||
|
||
private Task<int?> SendLineAsync(string text, CancellationToken ct) =>
|
||
_sender.SendHtmlAsync(_chatId, TelegramFormatter.Escape(text), text, cancellationToken: ct);
|
||
|
||
private Task<int?> SendHtmlLineAsync(string html, CancellationToken ct) =>
|
||
_sender.SendHtmlAsync(_chatId, html, StripTags(html), cancellationToken: ct);
|
||
|
||
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(_chatId, 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();
|
||
}
|
||
|
||
/// <summary>": <code>ls -la</code>" for bash, the most telling argument otherwise.</summary>
|
||
internal static string FormatArgs(JsonElement args)
|
||
{
|
||
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 $": <code>{TelegramFormatter.Escape(Shorten(summary, 200))}</code>";
|
||
}
|
||
|
||
private static string FormatExcerpt(ToolResult result)
|
||
{
|
||
var text = string.Concat(result.Content.OfType<TextContent>().Select(c => c.Text)).Trim();
|
||
return text.Length == 0 ? "" : $"\n<pre>{TelegramFormatter.Escape(Shorten(text, ToolExcerptLength))}</pre>";
|
||
}
|
||
|
||
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()
|
||
{
|
||
_events.Writer.TryComplete();
|
||
try
|
||
{
|
||
await _loop.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
||
}
|
||
catch (TimeoutException)
|
||
{
|
||
}
|
||
await _stop.CancelAsync().ConfigureAwait(false);
|
||
StopTyping();
|
||
_stop.Dispose();
|
||
}
|
||
}
|