admin/Playa
1
0
Fork 0
forked from admin/VolcanicAgenticCore
Code Pull Requests Activity

feat(telegram): ordered renderer work items and file uploads

Browse Source
This commit is contained in:
EugeneTes
2026-09-26 21:56:05 +02:00
parent b22d0d159e
commit 1eed03e851
6 changed files with 303 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

88
hosts/Volcanic.AgenticCore.Telegram/Rendering/ChatRenderer.cs
View File

@@ -15,6 +15,10 @@ namespace Volcanic.AgenticCore.Telegram.Rendering;
/// 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.
/// <para>
/// Host work (file uploads, notices) runs on the same loop through <see cref="RunInOrderAsync{T}"/>, so it stays in
/// order with the rendered output queued before it.
/// </para>
/// <para>
/// While a run is active one status message ("⚙️ Working...") lists the last <see cref="StatusToolCount"/> tool calls
/// with their state, edited at most every <see cref="TelegramOptions.EditInterval"/>. When something was sent below it,
/// the next tool update moves it back to the bottom; it is deleted when the run settles.
@@ -30,7 +34,10 @@ public sealed class ChatRenderer : IAsyncDisposable
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 abstract record RenderItem;
private sealed record EventItem(AgentSessionEvent Event) : RenderItem;
private sealed record WorkItem(Func<CancellationToken, Task> Run, Action Cancel) : RenderItem;
private readonly Channel<RenderItem> _items = Channel.CreateUnbounded<RenderItem>(new UnboundedChannelOptions { SingleReader = true });
private readonly CancellationTokenSource _stop = new();
private readonly Task _loop;
@@ -63,12 +70,37 @@ public sealed class ChatRenderer : IAsyncDisposable
}
/// <summary>Queue an event; safe to call from any thread (this is the session subscription).</summary>
public void Post(AgentSessionEvent e)
public void Post(AgentSessionEvent e) => Enqueue(new EventItem(e));
private bool Enqueue(RenderItem item)
{
Interlocked.Increment(ref _pending);
if (!_events.Writer.TryWrite(e)) Interlocked.Decrement(ref _pending);
if (_items.Writer.TryWrite(item)) return true;
Interlocked.Decrement(ref _pending);
return false;
}
/// <summary>
/// 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.
/// </summary>
public Task<T> RunInOrderAsync<T>(Func<CancellationToken, Task<T>> work)
{
ArgumentNullException.ThrowIfNull(work);
var done = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);
var item = new WorkItem(async ct =>
{
try { done.TrySetResult(await work(ct).ConfigureAwait(false)); }
catch (OperationCanceledException) { done.TrySetCanceled(); }
catch (Exception ex) { done.TrySetException(ex); }
}, () => done.TrySetCanceled());
if (!Enqueue(item)) done.TrySetCanceled();
return done.Task;
}
/// <summary>Plain text (no parse mode) in order with rendered output; cut to 4096 characters by the sender.</summary>
public Task<int?> SayAsync(string text) => RunInOrderAsync(ct => _sender.SendTextAsync(_chatId, text, cancellationToken: ct));
/// <summary>Completes once every queued event is rendered. For tests.</summary>
internal async Task DrainAsync()
{
@@ -77,7 +109,7 @@ public sealed class ChatRenderer : IAsyncDisposable
private async Task RunAsync()
{
var reader = _events.Reader;
var reader = _items.Reader;
var ct = _stop.Token;
try
{
@@ -107,15 +139,24 @@ public sealed class ChatRenderer : IAsyncDisposable
break;
}
while (reader.TryRead(out var e))
while (reader.TryRead(out var item))
{
try
{
await RenderAsync(e, ct).ConfigureAwait(false);
switch (item)
{
case EventItem { Event: var e }:
await RenderAsync(e, ct).ConfigureAwait(false);
break;
case WorkItem work:
await work.Run(ct).ConfigureAwait(false);
BuryStatus();
break;
}
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
_logger.LogWarning(ex, "Rendering {Event} for chat {ChatId} failed", e.GetType().Name, _chatId);
_logger.LogWarning(ex, "Rendering {Item} for chat {ChatId} failed", item is EventItem ev ? ev.Event.GetType().Name : "work", _chatId);
}
finally
{
@@ -131,6 +172,11 @@ public sealed class ChatRenderer : IAsyncDisposable
finally
{
StopTyping();
while (_items.Reader.TryRead(out var left))
{
if (left is WorkItem work) work.Cancel();
Interlocked.Decrement(ref _pending);
}
}
}
@@ -153,7 +199,7 @@ public sealed class ChatRenderer : IAsyncDisposable
break;
case MessageEnd { Message: AssistantMessage message }:
await SendReplyAsync(message.Text, ct).ConfigureAwait(false);
_ = 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)
@@ -204,10 +250,15 @@ public sealed class ChatRenderer : IAsyncDisposable
if (_statusDirty) await FlushStatusAsync(ct).ConfigureAwait(false);
}
/// <summary>Send a finished assistant message. A chunk that fails to send is retried a couple of times.</summary>
private async Task SendReplyAsync(string text, CancellationToken ct)
/// <summary>
/// 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.
/// </summary>
internal async Task<(int? FirstId, bool Complete)> SendReplyAsync(string text, CancellationToken ct)
{
if (string.IsNullOrWhiteSpace(text)) return;
if (string.IsNullOrWhiteSpace(text)) return (null, true);
int? first = null;
var complete = true;
foreach (var chunk in TelegramFormatter.Split(text))
{
var html = TelegramFormatter.ToHtml(chunk);
@@ -218,9 +269,18 @@ public sealed class ChatRenderer : IAsyncDisposable
await Task.Delay(delay, ct).ConfigureAwait(false);
sent = await _sender.SendHtmlAsync(_chatId, html, chunk, cancellationToken: ct).ConfigureAwait(false);
}
if (sent is null) _logger.LogError("Could not deliver part of a reply to chat {ChatId}", _chatId);
else BuryStatus();
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)
@@ -339,7 +399,7 @@ public sealed class ChatRenderer : IAsyncDisposable
public async ValueTask DisposeAsync()
{
_events.Writer.TryComplete();
_items.Writer.TryComplete();
try
{
await _loop.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);

41
hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramSender.cs
View File

@@ -102,6 +102,47 @@ public sealed class TelegramSender(ITelegramBotClient bot, ILogger<TelegramSende
}
}
/// <summary>Result of <see cref="SendFileAsync"/>: the message id, or the error; <c>PhotoRejected</c> when Telegram refused a photo with a 400.</summary>
public sealed record FileSendResult(int? MessageId, string? Error, bool PhotoRejected);
/// <summary>Delays before retrying an upload after a network failure (1 s, 3 s, like replies). Settable for tests.</summary>
internal IReadOnlyList<TimeSpan> FileRetryDelays { get; set; } = [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(3)];
/// <summary>
/// Upload a file as a photo or a document with a plain-text caption. Never throws except on cancellation: a network
/// failure is retried, an API error is returned (a 400 on a photo sets <c>PhotoRejected</c>).
/// </summary>
public async Task<FileSendResult> SendFileAsync(long chatId, string path, bool asPhoto, string? caption, CancellationToken cancellationToken = default)
{
for (var attempt = 0; ; attempt++)
{
try
{
await using var stream = new FileStream(path, FileMode.Open, FileAccess.Read, FileShare.Read, 81920, useAsync: true);
var file = InputFile.FromStream(stream, Path.GetFileName(path));
var message = asPhoto
? await bot.SendPhoto(chatId, file, caption: caption, cancellationToken: cancellationToken).ConfigureAwait(false)
: await bot.SendDocument(chatId, file, caption: caption, cancellationToken: cancellationToken).ConfigureAwait(false);
return new FileSendResult(message.Id, null, false);
}
catch (ApiRequestException ex)
{
logger.LogWarning(ex, "Uploading {File} to chat {ChatId} failed", Path.GetFileName(path), chatId);
return new FileSendResult(null, ex.Message, asPhoto && ex.ErrorCode == 400);
}
catch (RequestException ex) when (attempt < FileRetryDelays.Count)
{
logger.LogDebug(ex, "Uploading {File} to chat {ChatId} failed; retrying", Path.GetFileName(path), chatId);
await Task.Delay(FileRetryDelays[attempt], cancellationToken).ConfigureAwait(false);
}
catch (Exception ex) when (ex is not OperationCanceledException)
{
logger.LogWarning(ex, "Uploading {File} to chat {ChatId} failed", Path.GetFileName(path), chatId);
return new FileSendResult(null, ex.Message, false);
}
}
}
public async Task SendTypingAsync(long chatId, CancellationToken cancellationToken = default)
{
try