fix(telegram): send answers complete instead of streaming; working message no longer lingers
This commit is contained in:
@@ -68,10 +68,11 @@ Design (`PollingService : BackgroundService`, `BotUpdateHandler : IUpdateHandler
|
||||
bot command → `ChatCommands`; text/photo → prompt in the background. Errors are reported to the chat.
|
||||
- `ChatSessions`: one `AgentSession` per chat, lazy; resumes the most recent session in scope `telegram-<chatId>`.
|
||||
- `ChatRenderer`: subscribes once per session, events into a channel, one render task per chat.
|
||||
Placeholder message on assistant `MessageStart`; text deltas edit it at most every 1.5 s; final edit on `MessageEnd`;
|
||||
text over 4096 chars continues in new messages; typing action every 4 s while running; one status message
|
||||
Each assistant message is sent once, complete, on `MessageEnd` (no streaming); text over 4096 chars continues in
|
||||
new messages; typing action every 4 s while running; one status message
|
||||
per run, "⚙️ Working..." with the last 5 tool calls (`▶` running, `✓` done, `✗` failed), kept as the last message
|
||||
in the chat (deleted and resent when something is sent below it) and deleted when the run settles; retry, compaction, errors and notifications as short lines.
|
||||
in the chat (moved below anything sent under it on its next tool update, edited at most every 1.5 s) and deleted
|
||||
when the run settles; retry, compaction, errors and notifications as short lines.
|
||||
Telegram send/edit failures are logged and never stop the render task.
|
||||
- `TelegramFormatter`: Markdown → Telegram HTML (code blocks, inline code, bold, italics, links; everything else
|
||||
escaped); a send that Telegram rejects as bad markup is retried as plain text. Splitting at 4096.
|
||||
@@ -87,6 +88,7 @@ xunit, a fake `ITelegramBotClient` that records requests and scripts `getUpdates
|
||||
- A 401 stops the application.
|
||||
- A throwing handler does not stop polling.
|
||||
- Unknown users are ignored.
|
||||
- A prompt streams into an edited message; long text is split; `/stop` aborts a running prompt.
|
||||
- A reply is sent once, unedited; long text is split; `/stop` aborts a running prompt.
|
||||
- A recorded Claude Code event stream replayed with its timing leaves only the answers in the chat.
|
||||
- `TelegramUi` select round-trip through a callback query.
|
||||
- Formatter escaping and splitting.
|
||||
|
||||
@@ -30,7 +30,7 @@ Environment variables, or the `Telegram` section of `appsettings.json` (`Telegra
|
||||
| `TELEGRAM_ALLOWED_USER_IDS` / `Telegram:AllowedUserIds` | — | Required, comma separated. The bot refuses to start without it |
|
||||
| `Telegram:WorkspaceRoot` | `~/.volcanic/telegram-workspaces` | Chat `<id>` works in `<root>/<id>` |
|
||||
| `Telegram:Model` | `claude-code/claude-opus-5-5` | Any `provider/model` the Claude Code provider serves |
|
||||
| `Telegram:EditInterval` | `00:00:01.5` | Minimum time between edits of a streaming reply |
|
||||
| `Telegram:EditInterval` | `00:00:01.5` | Minimum time between edits of the "Working..." status message |
|
||||
| `Telegram:ReconnectInitialDelay` / `ReconnectMaxDelay` | `1 s` / `60 s` | Backoff after lost connection |
|
||||
|
||||
Sessions are stored as usual under `~/.volcanic/agent/sessions/telegram-<chatId>/`; after a restart each chat resumes
|
||||
@@ -48,7 +48,8 @@ Write normally. While the agent works, a status message stays at the bottom of t
|
||||
```
|
||||
|
||||
It lists the last five tool calls (`▶` running, `✓` done, `✗` failed) and is deleted when the run finishes.
|
||||
The reply streams into its own message above it (split into several past Telegram's 4096 character limit).
|
||||
Answers are not streamed: each one is sent complete when the model finishes it (split into several messages past
|
||||
Telegram's 4096 character limit).
|
||||
A message sent while the agent works steers the current run. Photos (with or without a caption) are passed to the model.
|
||||
|
||||
| Command | |
|
||||
@@ -80,5 +81,5 @@ Polling uses Telegram.Bot's built-in loop with a hardened error handler:
|
||||
- if the loop itself ever ends while the host runs, it is restarted from the last offset;
|
||||
- a `401` (revoked or wrong token) is the one error it does not retry: the host logs it and exits with code 1.
|
||||
|
||||
Replies that fail to send (Telegram unreachable mid-run) are logged; the render loop keeps going and later edits
|
||||
catch the message up.
|
||||
A reply that fails to send (Telegram unreachable mid-run) is retried after 1 s and 3 s, then logged as lost; the
|
||||
render loop keeps going either way.
|
||||
|
||||
@@ -12,12 +12,12 @@ 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"/>, and "typing…" repeats while the agent runs.
|
||||
/// 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>
|
||||
/// While a run is active one status message ("⚙️ Working...") lists the last <see cref="StatusToolCount"/> tool calls
|
||||
/// with their state. It is kept as the chat's last message (moved down when something is sent below it) and deleted
|
||||
/// when the run settles.
|
||||
/// 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.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class ChatRenderer : IAsyncDisposable
|
||||
@@ -34,10 +34,8 @@ public sealed class ChatRenderer : IAsyncDisposable
|
||||
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 string _replyText = "";
|
||||
private bool _replyDirty;
|
||||
private static readonly TimeSpan[] ReplyRetryDelays = [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(3)];
|
||||
|
||||
private DateTimeOffset _lastFlush = DateTimeOffset.MinValue;
|
||||
private CancellationTokenSource? _typing;
|
||||
private int _pending;
|
||||
@@ -52,7 +50,7 @@ public sealed class ChatRenderer : IAsyncDisposable
|
||||
private bool _statusDirty;
|
||||
private bool _statusBuried;
|
||||
|
||||
private bool Dirty => _replyDirty || _statusDirty;
|
||||
private bool Dirty => _statusDirty;
|
||||
|
||||
public ChatRenderer(long chatId, TelegramSender sender, TelegramOptions options, ILogger logger, TimeProvider? time = null)
|
||||
{
|
||||
@@ -125,7 +123,6 @@ public sealed class ChatRenderer : IAsyncDisposable
|
||||
}
|
||||
}
|
||||
}
|
||||
if (_replyDirty) await FlushReplyAsync(ct).ConfigureAwait(false);
|
||||
await DeleteStatusAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
@@ -152,31 +149,18 @@ public sealed class ChatRenderer : IAsyncDisposable
|
||||
break;
|
||||
case AgentSettled:
|
||||
StopTyping();
|
||||
if (_replyDirty) await FlushReplyAsync(ct).ConfigureAwait(false);
|
||||
await DeleteStatusAsync(ct).ConfigureAwait(false);
|
||||
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);
|
||||
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);
|
||||
ResetReply();
|
||||
break;
|
||||
|
||||
case ToolExecutionStart start:
|
||||
// Text the model wrote before the call goes out first, so it stays above the status message.
|
||||
if (_replyDirty) await FlushReplyAsync(ct).ConfigureAwait(false);
|
||||
_recentTools.Add(new ToolLine(start.ToolCallId, start.ToolName, FormatArgs(start.Args, StatusArgsLength)));
|
||||
if (_recentTools.Count > StatusToolCount) _recentTools.RemoveRange(0, _recentTools.Count - StatusToolCount);
|
||||
_statusDirty = true;
|
||||
@@ -217,58 +201,41 @@ public sealed class ChatRenderer : IAsyncDisposable
|
||||
private async Task FlushAsync(CancellationToken ct)
|
||||
{
|
||||
_lastFlush = _time.GetUtcNow();
|
||||
if (_replyDirty) await FlushReplyAsync(ct).ConfigureAwait(false);
|
||||
if (_statusDirty) await FlushStatusAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task FlushReplyAsync(CancellationToken ct)
|
||||
/// <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)
|
||||
{
|
||||
_replyDirty = false;
|
||||
if (string.IsNullOrWhiteSpace(_replyText)) return;
|
||||
|
||||
var chunks = TelegramFormatter.Split(_replyText);
|
||||
for (var i = 0; i < chunks.Count; i++)
|
||||
if (string.IsNullOrWhiteSpace(text)) return;
|
||||
foreach (var chunk in TelegramFormatter.Split(text))
|
||||
{
|
||||
var chunk = chunks[i];
|
||||
var html = TelegramFormatter.ToHtml(chunk);
|
||||
if (i < _replyMessages.Count)
|
||||
var sent = await _sender.SendHtmlAsync(_chatId, html, chunk, cancellationToken: ct).ConfigureAwait(false);
|
||||
foreach (var delay in ReplyRetryDelays)
|
||||
{
|
||||
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));
|
||||
BuryStatus();
|
||||
}
|
||||
else
|
||||
{
|
||||
break; // try the rest on the next flush
|
||||
if (sent is not null) break;
|
||||
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();
|
||||
}
|
||||
}
|
||||
|
||||
private void ResetReply()
|
||||
{
|
||||
_replyMessages.Clear();
|
||||
_replyText = "";
|
||||
_replyDirty = false;
|
||||
_lastFlush = DateTimeOffset.MinValue;
|
||||
}
|
||||
|
||||
private async Task SendLineAsync(string text, CancellationToken ct)
|
||||
{
|
||||
if (await _sender.SendHtmlAsync(_chatId, TelegramFormatter.Escape(text), text, cancellationToken: ct).ConfigureAwait(false) is not null)
|
||||
BuryStatus();
|
||||
}
|
||||
|
||||
/// <summary>Something was sent below the status message: move it back to the bottom on the next flush.</summary>
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
private void BuryStatus()
|
||||
{
|
||||
if (_statusMessageId is null) return;
|
||||
_statusBuried = true;
|
||||
_statusDirty = true;
|
||||
if (_statusMessageId is not null) _statusBuried = true;
|
||||
}
|
||||
|
||||
/// <summary>Send, move or edit the status message to show the current tool list.</summary>
|
||||
|
||||
@@ -98,7 +98,7 @@ public sealed class TelegramSender(ITelegramBotClient bot, ILogger<TelegramSende
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
logger.LogDebug(ex, "Deleting message {MessageId} in chat {ChatId} failed", messageId, chatId);
|
||||
logger.LogWarning(ex, "Deleting message {MessageId} in chat {ChatId} failed", messageId, chatId);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -10,7 +10,7 @@ namespace Volcanic.AgenticCore.Telegram.Tests;
|
||||
public sealed class ChatTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task A_prompt_streams_into_one_reply_message()
|
||||
public async Task A_reply_is_sent_once_when_complete()
|
||||
{
|
||||
await using var host = new ChatHost();
|
||||
host.Faux.Enqueue(FauxTurn.Say("Hello **there**, how can I help?"));
|
||||
@@ -21,7 +21,7 @@ public sealed class ChatTests
|
||||
var reply = Assert.Single(host.Bot.Sent, s => !s.Text.StartsWith(Working, StringComparison.Ordinal));
|
||||
Assert.Equal(ParseMode.Html, reply.ParseMode);
|
||||
Assert.Equal("Hello <b>there</b>, how can I help?", host.Bot.CurrentTexts.Values.Single());
|
||||
Assert.Contains(host.Bot.Requests, r => r is SendChatActionRequest);
|
||||
Assert.Empty(host.Bot.Edits); // not streamed
|
||||
Assert.Equal("hi", host.Faux.Requests.Single().Messages.Last().Text);
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
using System.Text.Json;
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Telegram.Bot.Requests;
|
||||
using Volcanic.AgenticCore.Abstractions.Events;
|
||||
using Volcanic.AgenticCore.Abstractions.Json;
|
||||
using Volcanic.AgenticCore.Abstractions.Messages;
|
||||
using Volcanic.AgenticCore.Telegram.Rendering;
|
||||
using Volcanic.AgenticCore.Telegram.Tests.Support;
|
||||
|
||||
namespace Volcanic.AgenticCore.Telegram.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Replays a session event stream recorded from a real Claude Code run (text, two bash calls, a final answer) through
|
||||
/// the renderer with its original timing, sped up four times.
|
||||
/// </summary>
|
||||
public sealed class ClaudeCodeReplayTests
|
||||
{
|
||||
private const string Fixture = "Fixtures/claude-code-two-tools.jsonl";
|
||||
private const double Speedup = 4;
|
||||
|
||||
[Fact]
|
||||
public async Task Answers_are_sent_once_and_the_working_message_is_gone_at_the_end()
|
||||
{
|
||||
var bot = new FakeBotClient();
|
||||
var options = new TelegramOptions { EditInterval = TimeSpan.FromSeconds(1.5 / Speedup), TypingInterval = TimeSpan.FromSeconds(1) };
|
||||
var renderer = new ChatRenderer(1, new TelegramSender(bot, NullLogger<TelegramSender>.Instance), options, NullLogger.Instance);
|
||||
|
||||
var events = Load();
|
||||
var start = DateTime.UtcNow;
|
||||
foreach (var (at, e) in events)
|
||||
{
|
||||
var due = start.AddSeconds(at / Speedup) - DateTime.UtcNow;
|
||||
if (due > TimeSpan.Zero) await Task.Delay(due);
|
||||
renderer.Post(e);
|
||||
}
|
||||
await renderer.DrainAsync().WaitAsync(TimeSpan.FromSeconds(10));
|
||||
await renderer.DisposeAsync();
|
||||
|
||||
var answers = events.Select(x => x.Event).OfType<AgentSessionEvent.MessageEnd>()
|
||||
.Select(m => m.Message).OfType<AssistantMessage>().Where(m => m.Text.Length > 0)
|
||||
.Select(m => TelegramFormatter.ToHtml(m.Text)).ToList();
|
||||
Assert.Equal(2, answers.Count);
|
||||
|
||||
// The chat ends with just the answers, each sent once and never edited (no streaming).
|
||||
Assert.Equal(answers, bot.CurrentTexts.OrderBy(kv => kv.Key).Select(kv => kv.Value));
|
||||
foreach (var answer in answers) Assert.Single(bot.Sent, s => s.Text == answer);
|
||||
Assert.All(bot.Edits, e => Assert.StartsWith("⚙️", e.Text));
|
||||
|
||||
// The status was not re-sent below the final answer.
|
||||
var sends = bot.Sent.Select(s => s.Text).ToList();
|
||||
Assert.Equal(answers[^1], sends[^1]);
|
||||
Assert.Contains(bot.Edits, e => e.Text!.Contains("✓ <b>bash</b>: <code>pwd</code>", StringComparison.Ordinal));
|
||||
Assert.Contains(bot.Requests, r => r is SendChatActionRequest);
|
||||
}
|
||||
|
||||
private static List<(double At, AgentSessionEvent Event)> Load() =>
|
||||
File.ReadLines(Fixture).Select(line =>
|
||||
{
|
||||
using var doc = JsonDocument.Parse(line);
|
||||
return (doc.RootElement.GetProperty("t").GetDouble(), doc.RootElement.GetProperty("e").Deserialize<AgentSessionEvent>(VolcanicJson.Options)!);
|
||||
}).ToList();
|
||||
}
|
||||
File diff suppressed because one or more lines are too long
@@ -12,4 +12,7 @@
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="../../hosts/Volcanic.AgenticCore.Telegram/Volcanic.AgenticCore.Telegram.csproj" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<None Include="Fixtures/**" CopyToOutputDirectory="PreserveNewest" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
|
||||
Reference in New Issue
Block a user