admin/VolcanicAgenticCore
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(telegram): address review findings

Browse Source 
- A speech or download timeout (TaskCanceledException nobody asked for) is a failure
  with the user notice, not a silent shutdown; OpenAiSpeechToText maps it to
  "the request timed out."; UpdateRouter only treats the caller's cancellation as shutdown.
- send: WorkspacePaths resolves the real path component by component (realpath),
  closing the directory-link-inside-target bypass; `.` reports a directory.
- Image documents are attached inline only when their magic bytes are JPEG/PNG/GIF/WebP,
  under the detected type.
- Audio goes to STT as an ASCII audio.<ext> name.
- MCP Browser section mentions send only when it is active.
- RunInOrderAsync takes a cancellation token; queued work is skipped once disposal gives up;
  SendReplyAsync stays private behind ReplyInOrderAsync.
- Speech HttpClient handler recycles pooled connections; BuiltInTools.Send();
  IExtensionHost.Channel is abstract; WorkspacePaths in its own file.
- Tests: timeouts, symlink chains (Linux), sniffing, channel delivery failures,
  queued-work cancellation, gated voice ordering. Spec and docs updated.
This commit is contained in:
EugeneTes
2026-09-26 22:35:20 +02:00
parent c5ecd017a0
commit 6ab98a8155
25 changed files with 536 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
hosts/Volcanic.AgenticCore.Telegram/Chats/TelegramChatChannel.cs
View File

@@ -17,11 +17,7 @@ public sealed class TelegramChatChannel(long chatId, TelegramSender sender, Chat
public async Task<ChatDelivery> SendTextAsync(string markdown, CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(markdown);
var (first, complete) = await renderer.RunInOrderAsync(async ct =>
{
using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, cancellationToken);
return await renderer.SendReplyAsync(markdown, linked.Token).ConfigureAwait(false);
}).ConfigureAwait(false);
var (first, complete) = await renderer.ReplyInOrderAsync(markdown, cancellationToken).ConfigureAwait(false);
if (!complete) throw new ChatDeliveryException("Telegram did not accept the message.");
return new ChatDelivery(null, first?.ToString(CultureInfo.InvariantCulture));
}
@@ -31,18 +27,17 @@ public sealed class TelegramChatChannel(long chatId, TelegramSender sender, Chat
ArgumentNullException.ThrowIfNull(file);
return renderer.RunInOrderAsync(async ct =>
{
using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, cancellationToken);
var asPhoto = file.Kind == ChatFileKind.Photo || (file.Kind == ChatFileKind.Auto && LooksLikePhoto(file.Path));
var result = await sender.SendFileAsync(chatId, file.Path, asPhoto, file.Caption, linked.Token).ConfigureAwait(false);
var result = await sender.SendFileAsync(chatId, file.Path, asPhoto, file.Caption, ct).ConfigureAwait(false);
if (result.MessageId is { } id) return Delivered(asPhoto, id);
if (asPhoto && file.Kind == ChatFileKind.Auto && result.PhotoRejected)
{
// Telegram refuses some images as photos (dimensions, aspect ratio): send the file as a document instead.
result = await sender.SendFileAsync(chatId, file.Path, asPhoto: false, file.Caption, linked.Token).ConfigureAwait(false);
result = await sender.SendFileAsync(chatId, file.Path, asPhoto: false, file.Caption, ct).ConfigureAwait(false);
if (result.MessageId is { } documentId) return Delivered(false, documentId);
}
throw new ChatDeliveryException($"Telegram refused the file: {result.Error}");
});
}, cancellationToken);
}
private bool LooksLikePhoto(string path) =>

2
hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs
View File

@@ -92,7 +92,7 @@ public sealed class UpdateRouter(
{
lease.Complete(await composer.ComposeAsync(chat, message, cancellationToken).ConfigureAwait(false));
}
catch (OperationCanceledException)
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
// shutting down: the lease completes empty
}

2
hosts/Volcanic.AgenticCore.Telegram/Inbound/InboundComposer.cs
View File

@@ -81,7 +81,7 @@ public sealed class InboundComposer(TelegramSender sender, InboundMedia media, I
await media.DownloadAsync(largest.FileId, stream, cancellationToken).ConfigureAwait(false);
draft.Images.Add(new ImageContent(Convert.ToBase64String(stream.ToArray()), "image/jpeg"));
}
catch (Exception ex) when (ex is not OperationCanceledException)
catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
{
logger.LogWarning(ex, "Downloading a photo in chat {ChatId} failed", chat.ChatId);
chat.Report("I could not download that photo.");

47
hosts/Volcanic.AgenticCore.Telegram/Inbound/InboundMedia.cs
View File

@@ -29,6 +29,9 @@ public sealed class InboundMedia(TelegramSender sender, TelegramOptions options,
/// <summary>Suffix of a <c>[File received: …]</c> line for an image too large to attach inline.</summary>
public const string TooLargeInlineNote = " (too large to show inline; downscale it with bash if you need to see it)";
/// <summary>Suffix of a <c>[File received: …]</c> line for an image document whose bytes aren't a known image.</summary>
public const string NotAnImageNote = " (not shown inline: the contents are not a JPEG, PNG, WebP or GIF image)";
private static readonly HashSet<string> InlineImageTypes = new(StringComparer.OrdinalIgnoreCase) { "image/jpeg", "image/png", "image/webp", "image/gif" };
/// <summary>The notice for a file over the 20 MB download limit; <c>over 20 MB</c> when the size is unknown.</summary>
@@ -67,7 +70,7 @@ public sealed class InboundMedia(TelegramSender sender, TelegramOptions options,
await using (stream.ConfigureAwait(false))
await DownloadAsync(fileId, stream, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException)
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
TryDelete(path);
throw;
@@ -88,8 +91,12 @@ public sealed class InboundMedia(TelegramSender sender, TelegramOptions options,
{
if (bytes <= options.InlineImageMaxBytes)
{
// Documents aren't re-encoded by Telegram, so the declared type can be wrong (HEIC named .png, a
// truncated file). A provider rejects such an image, and the persisted prompt would break every later
// turn: attach only bytes that really are one of the four types, under the type they show.
var data = await File.ReadAllBytesAsync(path, cancellationToken).ConfigureAwait(false);
draft.Images.Add(new ImageContent(Convert.ToBase64String(data), mime.ToLowerInvariant()));
if (SniffImageType(data) is { } actual) draft.Images.Add(new ImageContent(Convert.ToBase64String(data), actual));
else note = NotAnImageNote;
}
else
{
@@ -109,7 +116,7 @@ public sealed class InboundMedia(TelegramSender sender, TelegramOptions options,
var (fileId, size, name, mime, isVoice) = message switch
{
{ Voice: { } v } => (v.FileId, v.FileSize, "voice.ogg", "audio/ogg", true), // Telegram's .oga is rejected by OpenAI
{ Audio: { } a } => (a.FileId, a.FileSize, a.FileName ?? "audio", a.MimeType ?? "audio/mpeg", false),
{ Audio: { } a } => (a.FileId, a.FileSize, SpeechFileName(a.FileName, a.MimeType), a.MimeType ?? "audio/mpeg", false),
_ => ((string?)null, (long?)null, "", "", false),
};
if (fileId is null) return;
@@ -137,7 +144,7 @@ public sealed class InboundMedia(TelegramSender sender, TelegramOptions options,
{
await DownloadAsync(fileId, buffer, cancellationToken).ConfigureAwait(false);
}
catch (Exception ex) when (ex is not OperationCanceledException)
catch (Exception ex) when (!cancellationToken.IsCancellationRequested) // a client timeout is a failure, not a shutdown
{
logger.LogWarning(ex, "Downloading a voice message in chat {ChatId} failed", chat.ChatId);
await chat.Renderer.SayAsync(ex is ApiRequestException api && IsTooBig(api) ? TooBigNotice(size) : DownloadFailedNotice).ConfigureAwait(false);
@@ -152,7 +159,7 @@ public sealed class InboundMedia(TelegramSender sender, TelegramOptions options,
{
transcript = (await speech.TranscribeAsync(audio, name, mime, cancellationToken).ConfigureAwait(false)).Trim();
}
catch (Exception ex) when (ex is not OperationCanceledException)
catch (Exception ex) when (!cancellationToken.IsCancellationRequested) // a client timeout is a failure, not a shutdown
{
logger.LogWarning(ex, "Transcribing a voice message in chat {ChatId} failed", chat.ChatId);
await chat.Renderer.SayAsync($"Could not transcribe the voice message: {ex.Message}").ConfigureAwait(false);
@@ -167,6 +174,36 @@ public sealed class InboundMedia(TelegramSender sender, TelegramOptions options,
draft.Body.Add($"[Voice message] {transcript}");
}
/// <summary>
/// The name an audio file is sent to speech to text under: OpenAI picks the format from the extension, so it's an
/// ASCII <c>audio.&lt;ext&gt;</c> from the original extension when OpenAI knows it, else from the MIME type.
/// </summary>
internal static string SpeechFileName(string? fileName, string? mimeType)
{
var ext = Path.GetExtension(fileName ?? "").ToLowerInvariant();
if (ext is ".flac" or ".m4a" or ".mp3" or ".mp4" or ".mpeg" or ".mpga" or ".ogg" or ".wav" or ".webm") return "audio" + ext;
if (ext is ".oga" or ".opus") return "audio.ogg";
return "audio" + (mimeType?.ToLowerInvariant() switch
{
"audio/mp4" or "audio/m4a" or "audio/x-m4a" => ".m4a",
"audio/ogg" or "audio/opus" => ".ogg",
"audio/wav" or "audio/x-wav" or "audio/wave" => ".wav",
"audio/flac" or "audio/x-flac" => ".flac",
"audio/webm" => ".webm",
_ => ".mp3", // audio/mpeg, Telegram's default for audio
});
}
/// <summary>The image type the leading bytes show (JPEG, PNG, GIF or WebP), or null.</summary>
internal static string? SniffImageType(ReadOnlySpan<byte> data) => data switch
{
[0xFF, 0xD8, 0xFF, ..] => "image/jpeg",
[0x89, 0x50, 0x4E, 0x47, 0x0D, 0x0A, 0x1A, 0x0A, ..] => "image/png",
[0x47, 0x49, 0x46, 0x38, ..] => "image/gif",
[0x52, 0x49, 0x46, 0x46, _, _, _, _, 0x57, 0x45, 0x42, 0x50, ..] => "image/webp",
_ => null,
};
private void TryDelete(string path)
{
try { File.Delete(path); }

6
hosts/Volcanic.AgenticCore.Telegram/Program.cs
View File

@@ -42,7 +42,11 @@ builder.Services.AddSingleton(options);
builder.Services.AddSingleton(mcp);
builder.Services.AddSingleton(options.Speech);
if (options.Speech.IsEnabled) // no key, no ISpeechToText: voice notes get the setup notice
builder.Services.AddHttpClient<ISpeechToText, OpenAiSpeechToText>(client => client.Timeout = TimeSpan.FromMinutes(2));
// InboundMedia (a singleton) keeps this typed client for the process lifetime, so the factory never rotates its
// handler: let the handler recycle its pooled connections instead, so DNS changes are still picked up.
builder.Services.AddHttpClient<ISpeechToText, OpenAiSpeechToText>(client => client.Timeout = TimeSpan.FromMinutes(2))
.ConfigurePrimaryHttpMessageHandler(() => new SocketsHttpHandler { PooledConnectionLifetime = TimeSpan.FromMinutes(5) })
.SetHandlerLifetime(Timeout.InfiniteTimeSpan);
if (calendar is not null) builder.Services.AddSingleton(calendar); // ChatRegistry and BotSetupService take it optionally
builder.Services.AddSingleton<OwnerLoginChannel>();
// The extension is constructed per session through ActivatorUtilities, which injects this. LoginUi resolves

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

@@ -41,7 +41,8 @@ public sealed class ChatRenderer : IAsyncDisposable
private readonly CancellationTokenSource _stop = new();
private readonly Task _loop;
private static readonly TimeSpan[] ReplyRetryDelays = [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(3)];
/// <summary>Delays before resending a reply chunk that failed (1 s, 3 s). Settable for tests.</summary>
internal IReadOnlyList<TimeSpan> ReplyRetryDelays { get; set; } = [TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(3)];
private DateTimeOffset _lastFlush = DateTimeOffset.MinValue;
private CancellationTokenSource? _typing;
@@ -83,23 +84,49 @@ public sealed class ChatRenderer : IAsyncDisposable
/// <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.
/// <paramref name="cancellationToken"/> 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.
/// </summary>
public Task<T> RunInOrderAsync<T>(Func<CancellationToken, Task<T>> work)
public Task<T> RunInOrderAsync<T>(Func<CancellationToken, Task<T>> work, CancellationToken cancellationToken = default)
{
ArgumentNullException.ThrowIfNull(work);
if (cancellationToken.IsCancellationRequested) return Task.FromCanceled<T>(cancellationToken);
var done = new TaskCompletionSource<T>(TaskCreationOptions.RunContinuationsAsynchronously);
var registration = cancellationToken.Register(() => done.TrySetCanceled(cancellationToken));
var item = new WorkItem(async ct =>
{
try { done.TrySetResult(await work(ct).ConfigureAwait(false)); }
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); }
}, () => done.TrySetCanceled());
if (!Enqueue(item)) done.TrySetCanceled();
finally { await registration.DisposeAsync().ConfigureAwait(false); }
}, () =>
{
registration.Dispose();
done.TrySetCanceled();
});
if (!Enqueue(item))
{
registration.Dispose();
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));
public Task<int?> SayAsync(string text, CancellationToken cancellationToken = default) =>
RunInOrderAsync(ct => _sender.SendTextAsync(_chatId, text, cancellationToken: ct), cancellationToken);
/// <summary>
/// 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.
/// </summary>
internal Task<(int? FirstId, bool Complete)> ReplyInOrderAsync(string markdown, CancellationToken cancellationToken) =>
RunInOrderAsync(ct => SendReplyAsync(markdown, ct), cancellationToken);
/// <summary>Completes once every queued event is rendered. For tests.</summary>
internal async Task DrainAsync()
@@ -148,6 +175,9 @@ public sealed class ChatRenderer : IAsyncDisposable
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();
@@ -254,7 +284,7 @@ public sealed class ChatRenderer : IAsyncDisposable
/// 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)
private async Task<(int? FirstId, bool Complete)> SendReplyAsync(string text, CancellationToken ct)
{
if (string.IsNullOrWhiteSpace(text)) return (null, true);
int? first = null;

25
hosts/Volcanic.AgenticCore.Telegram/Speech/OpenAiSpeechToText.cs
View File

@@ -21,8 +21,29 @@ public sealed class OpenAiSpeechToText(HttpClient http, SpeechOptions options) :
using var request = new HttpRequestMessage(HttpMethod.Post, options.BaseUrl.TrimEnd('/') + "/audio/transcriptions") { Content = form };
request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", options.ApiKey);
using var response = await http.SendAsync(request, cancellationToken).ConfigureAwait(false);
var body = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
HttpResponseMessage response;
string body;
try
{
response = await http.SendAsync(request, cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException ex) when (!cancellationToken.IsCancellationRequested) // HttpClient.Timeout
{
throw new SpeechToTextException(Prefix + "the request timed out.", ex);
}
catch (HttpRequestException ex)
{
throw new SpeechToTextException(Prefix + Hide(ex.Message), ex);
}
using var _ = response;
try
{
body = await response.Content.ReadAsStringAsync(cancellationToken).ConfigureAwait(false);
}
catch (OperationCanceledException ex) when (!cancellationToken.IsCancellationRequested)
{
throw new SpeechToTextException(Prefix + "the request timed out.", ex);
}
if (!response.IsSuccessStatusCode)
{
var hidden = Hide(body); // hide the key before cutting, so half a key can't leak