From 308e1fbf8efa15b47bc3e72759812c9cdd215391 Mon Sep 17 00:00:00 2001 From: EugeneTes Date: Fri, 25 Sep 2026 14:42:12 +0000 Subject: [PATCH] feat(telegram): telegram bot host with connection recovery --- README.md | 7 +- VolcanicAgenticCore.slnx | 4 + docs/telegram-bot.md | 75 +++++ docs/ui-integration.md | 3 +- .../BotSetupService.cs | 45 +++ .../Chats/ChatCommands.cs | 171 ++++++++++ .../Chats/ChatRegistry.cs | 135 ++++++++ .../Chats/TelegramUi.cs | 92 ++++++ .../Chats/UpdateRouter.cs | 95 ++++++ .../Polling/PollingService.cs | 153 +++++++++ .../Polling/ReconnectPolicy.cs | 27 ++ .../Volcanic.AgenticCore.Telegram/Program.cs | 55 ++++ .../Rendering/ChatRenderer.cs | 309 ++++++++++++++++++ .../Rendering/TelegramFormatter.cs | 187 +++++++++++ .../Rendering/TelegramSender.cs | 126 +++++++ .../TelegramOptions.cs | 62 ++++ .../Volcanic.AgenticCore.Telegram.csproj | 16 + .../appsettings.json | 14 + .../ChatTests.cs | 220 +++++++++++++ .../FormattingAndOptionsTests.cs | 87 +++++ .../GlobalUsings.cs | 1 + .../PollingServiceTests.cs | 139 ++++++++ .../Support/ChatHost.cs | 75 +++++ .../Support/FakeBotClient.cs | 107 ++++++ .../Support/Updates.cs | 65 ++++ ...Volcanic.AgenticCore.Telegram.Tests.csproj | 15 + 26 files changed, 2283 insertions(+), 2 deletions(-) create mode 100644 docs/telegram-bot.md create mode 100644 hosts/Volcanic.AgenticCore.Telegram/BotSetupService.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Chats/TelegramUi.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Polling/ReconnectPolicy.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Program.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Rendering/ChatRenderer.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramFormatter.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramSender.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/TelegramOptions.cs create mode 100644 hosts/Volcanic.AgenticCore.Telegram/Volcanic.AgenticCore.Telegram.csproj create mode 100644 hosts/Volcanic.AgenticCore.Telegram/appsettings.json create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/ChatTests.cs create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/FormattingAndOptionsTests.cs create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/GlobalUsings.cs create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/PollingServiceTests.cs create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/Support/FakeBotClient.cs create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/Support/Updates.cs create mode 100644 tests/Volcanic.AgenticCore.Telegram.Tests/Volcanic.AgenticCore.Telegram.Tests.csproj diff --git a/README.md b/README.md index c1c8e4b..31f3f97 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ A headless, extensible agent harness for .NET, modelled on [earendil-works/pi](https://github.com/earendil-works/pi). It keeps the parts of pi that make it *extensible* and *provider-agnostic* and leaves out the terminal UI, -so you can put any front end on top (a Telegram bot is the intended first one). +so you can put any front end on top. A Telegram bot ships as the first one. It also ships a C# port of [chem/pi-claude-code-provider](https://github.com/chem/pi-claude-code-provider): an extension that drives the locally installed `claude` CLI on a Claude subscription, so the core can use @@ -17,6 +17,7 @@ Claude Code as a model provider without an API key. | Anthropic Messages API provider | `extensions/Volcanic.AgenticCore.Extensions.Anthropic` | `ANTHROPIC_API_KEY` | | OpenAI-compatible provider | `extensions/Volcanic.AgenticCore.Extensions.OpenAI` | OpenAI, OpenRouter, Groq, Ollama, llama.cpp, vLLM | | Claude Code (subscription) provider | `extensions/Volcanic.AgenticCore.Extensions.ClaudeCode` (+ `.Bridge`) | Spawns `claude -p`, MCP "proposal" bridge | +| Telegram bot | `hosts/Volcanic.AgenticCore.Telegram` | Chat with the agent from Telegram; long polling, Claude Code, allow-list | | Reference headless host | `samples/Volcanic.AgenticCore.Sample.Headless` | Prompt in, JSON events out | | Tests | `tests/` | xunit; no test calls a paid API | @@ -34,6 +35,9 @@ ANTHROPIC_API_KEY=sk-ant-... dotnet run --project samples/Volcanic.AgenticCore.S # with a Claude subscription through the claude CLI dotnet run --project samples/Volcanic.AgenticCore.Sample.Headless -- --model claude-code/sonnet "List the files here" + +# the Telegram bot (see docs/telegram-bot.md) +TELEGRAM_BOT_TOKEN=123:abc TELEGRAM_ALLOWED_USER_IDS=111111111 dotnet run --project hosts/Volcanic.AgenticCore.Telegram ``` Hosting it yourself: @@ -62,6 +66,7 @@ await session.PromptAsync("Explain this repository"); - [Architecture](docs/architecture.md): projects, vertical slices, the agent loop and the event model. - [Writing extensions](docs/extensions.md): tools, commands, lifecycle hooks, providers, message injection, discovery. - [Providers and models](docs/providers.md): model keys, authentication, the built-in provider extensions. +- [Telegram bot](docs/telegram-bot.md): setup, commands, security, connection recovery. - [Building a UI (Telegram guide)](docs/ui-integration.md): how to put a chat front end on the core. - [Claude Code provider](docs/claude-code-provider.md): requirements, how the bridge works, limits, the doctor command. - Design spec and plan: `docs/superpowers/specs/`, `docs/superpowers/plans/`. diff --git a/VolcanicAgenticCore.slnx b/VolcanicAgenticCore.slnx index 87cc5d5..d35a7d9 100644 --- a/VolcanicAgenticCore.slnx +++ b/VolcanicAgenticCore.slnx @@ -5,6 +5,9 @@ + + + @@ -14,6 +17,7 @@ + diff --git a/docs/telegram-bot.md b/docs/telegram-bot.md new file mode 100644 index 0000000..0114fa3 --- /dev/null +++ b/docs/telegram-bot.md @@ -0,0 +1,75 @@ +# Telegram bot + +`hosts/Volcanic.AgenticCore.Telegram` lets you chat with the agent from Telegram. It uses long polling (no public +URL, no webhook), gives every private chat its own agent session and workspace, and runs the model through the +Claude Code provider (`claude-code/claude-opus-5-5`) on your Claude subscription. + +## Setup + +1. Create a bot with [@BotFather](https://t.me/BotFather) (`/newbot`) and copy the token. +2. Find your numeric Telegram user id, for example by messaging [@userinfobot](https://t.me/userinfobot). +3. Install Claude Code on the machine that runs the bot and log in (`claude`, then `/login`). Check with + `claude --version`; see [claude-code-provider.md](claude-code-provider.md). +4. Run: + +```bash +TELEGRAM_BOT_TOKEN=123456:ABC... \ +TELEGRAM_ALLOWED_USER_IDS=111111111 \ +dotnet run --project hosts/Volcanic.AgenticCore.Telegram +``` + +Open the bot in Telegram and say hello. + +## Configuration + +Environment variables, or the `Telegram` section of `appsettings.json` (`Telegram__Name` in the environment): + +| Setting | Default | Notes | +|---|---|---| +| `TELEGRAM_BOT_TOKEN` / `Telegram:BotToken` | — | Required | +| `TELEGRAM_ALLOWED_USER_IDS` / `Telegram:AllowedUserIds` | — | Required, comma separated. The bot refuses to start without it | +| `Telegram:WorkspaceRoot` | `~/.volcanic/telegram-workspaces` | Chat `` works in `/` | +| `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:ReconnectInitialDelay` / `ReconnectMaxDelay` | `1 s` / `60 s` | Backoff after lost connection | + +Sessions are stored as usual under `~/.volcanic/agent/sessions/telegram-/`; after a restart each chat resumes +its most recent conversation. + +## Using it + +Write normally. The reply streams into one message (split into several past Telegram's 4096 character limit); +each tool call shows as a line, `▶ bash: ls -la` becoming `✓ bash` or `✗ bash` with an excerpt of the error. +A message sent while the agent works steers the current run. Photos (with or without a caption) are passed to the model. + +| Command | | +|---|---| +| `/new` | Start a new conversation | +| `/stop` | Stop the current run | +| `/status` | Model, thinking level, session, context usage, workspace | +| `/sessions`, `/resume ` | List and resume earlier conversations (an id prefix is enough) | +| `/compact` | Summarise the conversation to free context | +| `/thinking [level]` | Show or set the thinking level (Claude Code `--effort`) | +| `/help` | Short help | + +Any other `/command` is passed to the session, so extension commands such as `/claude-code-doctor` work. +When an extension asks something, the question appears with buttons (or as a reply prompt for free text). + +## Security + +The agent has the built-in tools, including `bash`, and runs them as the bot's OS user. Only allow-listed user ids +are served (everyone else is ignored and logged), and only in private chats. Run the bot as a dedicated user or in a +container, and treat the workspace root as writable by anyone on the allow-list. + +## Connection loss + +Polling uses Telegram.Bot's built-in loop with a hardened error handler: + +- a failed `getUpdates` waits 1 s, 2 s, 4 s … up to 60 s (±20 % jitter) before the next attempt, instead of retrying + in a tight loop; the first successful poll resets the delay and logs `Telegram connection restored`; +- the update offset only moves past updates that were handed on, so reconnecting neither loses nor repeats messages; +- 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. diff --git a/docs/ui-integration.md b/docs/ui-integration.md index e8722e9..70f8419 100644 --- a/docs/ui-integration.md +++ b/docs/ui-integration.md @@ -7,7 +7,8 @@ The core has no user interface. A UI is a process that 3. subscribes to `AgentSessionEvent`s and renders them, 4. implements `IUserInterface` so extensions can ask the user questions. -This guide uses Telegram as the worked example. Everything transfers to Slack, Discord, a web socket +This guide uses Telegram as the worked example; the finished bot lives in `hosts/Volcanic.AgenticCore.Telegram` +(see [telegram-bot.md](telegram-bot.md)) and is the reference for everything below. Everything transfers to Slack, Discord, a web socket front end or a desktop app; only the rendering code changes. ## 1. One session per chat diff --git a/hosts/Volcanic.AgenticCore.Telegram/BotSetupService.cs b/hosts/Volcanic.AgenticCore.Telegram/BotSetupService.cs new file mode 100644 index 0000000..5df0f58 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/BotSetupService.cs @@ -0,0 +1,45 @@ +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Telegram.Bot; +using Volcanic.AgenticCore.Telegram.Chats; + +namespace Volcanic.AgenticCore.Telegram; + +/// +/// Best-effort start-up calls: log who the bot is and publish the command menu. Retried in the background until the +/// first success, so a host started without network still comes up. +/// +public sealed class BotSetupService(ITelegramBotClient bot, TelegramOptions options, ILogger logger) : BackgroundService +{ + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + var delay = options.ReconnectInitialDelay; + while (!stoppingToken.IsCancellationRequested) + { + try + { + var me = await bot.GetMe(stoppingToken).ConfigureAwait(false); + await bot.SetMyCommands(ChatCommands.Menu, cancellationToken: stoppingToken).ConfigureAwait(false); + logger.LogInformation("Bot @{Username} ({Id}) is ready", me.Username, me.Id); + return; + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + return; + } + catch (Exception ex) + { + logger.LogDebug(ex, "Bot setup failed; retrying in {Delay}", delay); + } + try + { + await Task.Delay(delay, stoppingToken).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + return; + } + delay = TimeSpan.FromTicks(Math.Min(delay.Ticks * 2, options.ReconnectMaxDelay.Ticks)); + } + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs b/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs new file mode 100644 index 0000000..d866e24 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs @@ -0,0 +1,171 @@ +using System.Globalization; +using System.Text; +using Telegram.Bot.Types; +using Volcanic.AgenticCore.Abstractions.Providers; +using Volcanic.AgenticCore.Telegram.Rendering; + +namespace Volcanic.AgenticCore.Telegram.Chats; + +/// The bot's own slash commands. Anything else starting with / goes to the session (extension commands). +public sealed class ChatCommands(TelegramSender sender) +{ + public static readonly IReadOnlyList Menu = + [ + new() { Command = "new", Description = "Start a new conversation" }, + new() { Command = "stop", Description = "Stop the current run" }, + new() { Command = "status", Description = "Model, session and context usage" }, + new() { Command = "sessions", Description = "Recent conversations" }, + new() { Command = "resume", Description = "Resume a conversation: /resume " }, + new() { Command = "compact", Description = "Summarise the conversation to free context" }, + new() { Command = "thinking", Description = "Show or set the thinking level" }, + new() { Command = "help", Description = "What this bot does" }, + ]; + + /// Split /name@bot args. Null when the text is not a command. + public static (string Name, string Args)? Parse(string text) + { + if (!text.StartsWith('/') || text.Length < 2) return null; + var space = text.IndexOfAny([' ', '\n']); + var head = space < 0 ? text[1..] : text[1..space]; + var args = space < 0 ? "" : text[(space + 1)..].Trim(); + var at = head.IndexOf('@'); + if (at >= 0) head = head[..at]; + return (head.ToLowerInvariant(), args); + } + + /// Run a bot command. False when is not one of ours. + public async Task TryHandleAsync(ChatContext chat, string text, CancellationToken cancellationToken) + { + if (Parse(text) is not { } command) return false; + var session = chat.Session; + switch (command.Name) + { + case "start": + case "help": + await ReplyAsync(chat, Help(chat), cancellationToken).ConfigureAwait(false); + return true; + + case "stop": + if (!session.IsStreaming && !session.IsCompacting) + { + await ReplyAsync(chat, "Nothing is running.", cancellationToken).ConfigureAwait(false); + return true; + } + await session.AbortAsync().ConfigureAwait(false); + return true; // the renderer reports "Stopped." + + case "new": + if (session.IsStreaming) await session.AbortAsync().ConfigureAwait(false); + var created = await session.NewSessionAsync(cancellationToken: cancellationToken).ConfigureAwait(false); + await ReplyAsync(chat, created.Cancelled ? "An extension cancelled the new session." : "🆕 New conversation.", cancellationToken).ConfigureAwait(false); + return true; + + case "status": + await ReplyAsync(chat, Status(chat), cancellationToken).ConfigureAwait(false); + return true; + + case "sessions": + await ReplyAsync(chat, await SessionsAsync(chat, cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false); + return true; + + case "resume": + await ReplyAsync(chat, await ResumeAsync(chat, command.Args, cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false); + return true; + + case "compact": + if (session.IsStreaming) + { + await ReplyAsync(chat, "Wait until the current run finishes, or /stop it.", cancellationToken).ConfigureAwait(false); + return true; + } + _ = Task.Run(async () => + { + try + { + await session.CompactAsync(cancellationToken: CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) + { + chat.Report($"Compaction failed: {ex.Message}"); + } + }, CancellationToken.None); + return true; // the renderer shows compaction progress + + case "thinking": + await ReplyAsync(chat, await ThinkingAsync(chat, command.Args, cancellationToken).ConfigureAwait(false), cancellationToken).ConfigureAwait(false); + return true; + + default: + return false; + } + } + + private Task ReplyAsync(ChatContext chat, string text, CancellationToken cancellationToken) => + sender.SendHtmlAsync(chat.ChatId, TelegramFormatter.ToHtml(text), text, cancellationToken: cancellationToken); + + private static string Help(ChatContext chat) => + $""" + I am a coding agent running **{chat.Session.Model?.Name ?? "an unconfigured model"}**. Write to me as you would to a colleague; I can read and change files and run shell commands in my workspace `{chat.Session.Cwd}`. + + While I am working, a new message steers the current run. + + /new – start a new conversation + /stop – stop the current run + /status – model, session and context usage + /sessions, /resume – earlier conversations + /compact – summarise the conversation to free context + /thinking [level] – show or set the thinking level + """; + + private static string Status(ChatContext chat) + { + var session = chat.Session; + var sb = new StringBuilder(); + sb.AppendLine(CultureInfo.InvariantCulture, $"**Model:** {session.Model?.Key ?? "none"}"); + sb.AppendLine(CultureInfo.InvariantCulture, $"**Thinking:** {session.ThinkingLevel.ToString().ToLowerInvariant()}"); + sb.AppendLine(CultureInfo.InvariantCulture, $"**Session:** `{session.SessionId}`{(session.SessionName is { } name ? $" ({name})" : "")}"); + sb.AppendLine(CultureInfo.InvariantCulture, $"**Messages:** {session.Messages.Count}"); + if (session.GetContextUsage() is { } usage) + sb.AppendLine(CultureInfo.InvariantCulture, $"**Context:** {(usage.Tokens is { } t ? t.ToString("N0", CultureInfo.InvariantCulture) : "?")} / {usage.ContextWindow.ToString("N0", CultureInfo.InvariantCulture)} tokens{(usage.Percent is { } p ? $" ({p:0}%)" : "")}"); + sb.AppendLine(CultureInfo.InvariantCulture, $"**Workspace:** `{session.Cwd}`"); + sb.Append(session.IsStreaming ? "**State:** running" : session.IsCompacting ? "**State:** compacting" : "**State:** idle"); + return sb.ToString(); + } + + private static async Task SessionsAsync(ChatContext chat, CancellationToken cancellationToken) + { + var sessions = await chat.Session.ListSessionsAsync(cancellationToken).ConfigureAwait(false); + if (sessions.Count == 0) return "No conversations yet."; + var sb = new StringBuilder("**Recent conversations** (newest first):\n"); + foreach (var s in sessions.Take(10)) + { + var title = s.Name ?? s.FirstUserText ?? "(empty)"; + if (title.Length > 60) title = title[..60] + "…"; + var current = s.SessionId == chat.Session.SessionId ? " ← current" : ""; + sb.Append(CultureInfo.InvariantCulture, $"\n`/resume {s.SessionId}`{current}\n{s.Modified:yyyy-MM-dd HH:mm} · {s.MessageCount} messages · {title}\n"); + } + return sb.ToString(); + } + + private static async Task ResumeAsync(ChatContext chat, string id, CancellationToken cancellationToken) + { + if (id.Length == 0) return "Usage: /resume (see /sessions)."; + var sessions = await chat.Session.ListSessionsAsync(cancellationToken).ConfigureAwait(false); + var matches = sessions.Where(s => s.SessionId.StartsWith(id, StringComparison.OrdinalIgnoreCase)).ToList(); + if (matches.Count != 1) return matches.Count == 0 ? $"No conversation `{id}`." : $"`{id}` matches several conversations; use more characters."; + if (chat.Session.IsStreaming) await chat.Session.AbortAsync().ConfigureAwait(false); + var result = await chat.Session.SwitchSessionAsync(matches[0].SessionId, cancellationToken).ConfigureAwait(false); + return result.Cancelled ? "An extension cancelled the switch." : $"Resumed `{matches[0].SessionId}` ({matches[0].MessageCount} messages)."; + } + + private static async Task ThinkingAsync(ChatContext chat, string arg, CancellationToken cancellationToken) + { + var session = chat.Session; + var available = string.Join(", ", session.AvailableThinkingLevels.Select(l => l.ToString().ToLowerInvariant())); + if (arg.Length == 0) return $"Thinking: **{session.ThinkingLevel.ToString().ToLowerInvariant()}**. Available: {available}."; + if (!Enum.TryParse(arg, ignoreCase: true, out var level) || !session.AvailableThinkingLevels.Contains(level)) + return $"Unknown level `{arg}`. Available: {available}."; + await session.SetThinkingLevelAsync(level, cancellationToken).ConfigureAwait(false); + return $"Thinking: **{session.ThinkingLevel.ToString().ToLowerInvariant()}**."; + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs b/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs new file mode 100644 index 0000000..395e365 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs @@ -0,0 +1,135 @@ +using System.Collections.Concurrent; +using Microsoft.Extensions.Logging; +using Volcanic.AgenticCore.Abstractions.Events; +using Volcanic.AgenticCore.Abstractions.Extensions.Events; +using Volcanic.AgenticCore.Abstractions.Messages; +using Volcanic.AgenticCore.Abstractions.Ui; +using Volcanic.AgenticCore.Features.Agent; +using Volcanic.AgenticCore.Hosting; +using Volcanic.AgenticCore.Telegram.Rendering; + +namespace Volcanic.AgenticCore.Telegram.Chats; + +/// Everything that belongs to one Telegram chat: its agent session, renderer and dialog surface. +public sealed class ChatContext(long chatId, AgentSession session, TelegramUi ui, ChatRenderer renderer, IDisposable subscription, ILogger logger) + : IAsyncDisposable +{ + public long ChatId => chatId; + public AgentSession Session => session; + public TelegramUi Ui => ui; + public ChatRenderer Renderer => renderer; + + /// The most recent background prompt; completes when that run has settled. + public Task LastPrompt { get; private set; } = Task.CompletedTask; + + /// + /// Send user input without waiting for the run (it can take minutes, and the polling loop must keep receiving so + /// /stop gets through). While a run is active the text steers it. + /// + public Task PromptInBackground(string text, IReadOnlyList? images = null) => LastPrompt = Task.Run(async () => + { + try + { + await session.PromptAsync(text, images, StreamingBehavior.Steer).ConfigureAwait(false); + } + catch (Exception ex) + { + logger.LogError(ex, "Prompt in chat {ChatId} failed", chatId); + Report($"The request failed: {ex.Message}"); + } + }); + + /// Show an error in the chat, in order with the rendered events. + public void Report(string message) => renderer.Post(new AgentSessionEvent.Notification(message, NotifyKind.Error)); + + public async ValueTask DisposeAsync() + { + subscription.Dispose(); + await session.DisposeAsync().ConfigureAwait(false); + await renderer.DisposeAsync().ConfigureAwait(false); + } +} + +/// +/// One per chat, created on first use. Each chat works in <WorkspaceRoot>/<chatId>, +/// stores its sessions under scope telegram-<chatId> and resumes the most recent one after a restart. +/// +public sealed class ChatRegistry(IAgentSessionFactory factory, TelegramSender sender, TelegramOptions options, ILoggerFactory loggerFactory) + : IAsyncDisposable +{ + private readonly ConcurrentDictionary>> _chats = new(); + private readonly ILogger _logger = loggerFactory.CreateLogger(); + + public static string ScopeFor(long chatId) => $"telegram-{chatId}"; + + public string WorkspaceFor(long chatId) => Path.Combine(options.WorkspaceRoot, chatId.ToString(System.Globalization.CultureInfo.InvariantCulture)); + + /// The chat's context if it was already created, without creating it. + public ChatContext? Find(long chatId) => + _chats.TryGetValue(chatId, out var lazy) && lazy.IsValueCreated && lazy.Value.IsCompletedSuccessfully ? lazy.Value.Result : null; + + public async Task GetAsync(long chatId, CancellationToken cancellationToken = default) + { + var lazy = _chats.GetOrAdd(chatId, id => new Lazy>(() => CreateAsync(id))); + try + { + return await lazy.Value.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception) when (lazy.Value.IsFaulted) + { + _chats.TryRemove(new KeyValuePair>>(chatId, lazy)); // let the next message try again + throw; + } + } + + private async Task CreateAsync(long chatId) + { + var cwd = WorkspaceFor(chatId); + Directory.CreateDirectory(cwd); + var scope = ScopeFor(chatId); + var last = (await factory.ListSessionsAsync(cwd, scope).ConfigureAwait(false)).FirstOrDefault(); + + var logger = loggerFactory.CreateLogger(); + var renderer = new ChatRenderer(chatId, sender, options, loggerFactory.CreateLogger()); + var ui = new TelegramUi(chatId, sender, renderer); + AgentSession session; + try + { + session = await factory.CreateAsync(new AgentSessionOptions + { + Cwd = cwd, + Scope = scope, + SessionId = last?.SessionId, + Model = options.Model, + Ui = ui, + }).ConfigureAwait(false); + } + catch + { + await renderer.DisposeAsync().ConfigureAwait(false); + throw; + } + + var subscription = session.Subscribe(renderer.Post); + _logger.LogInformation("Chat {ChatId}: {Action} session {SessionId} in {Cwd}", chatId, last is null ? "new" : "resumed", session.SessionId, cwd); + foreach (var error in session.ExtensionErrors) _logger.LogWarning("Chat {ChatId}: extension error {Source}: {Error}", chatId, error.Source, error.Error); + return new ChatContext(chatId, session, ui, renderer, subscription, logger); + } + + public async ValueTask DisposeAsync() + { + foreach (var lazy in _chats.Values) + { + if (!lazy.IsValueCreated) continue; + try + { + await (await lazy.Value.ConfigureAwait(false)).DisposeAsync().ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Disposing a chat failed"); + } + } + _chats.Clear(); + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Chats/TelegramUi.cs b/hosts/Volcanic.AgenticCore.Telegram/Chats/TelegramUi.cs new file mode 100644 index 0000000..b0d2182 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Chats/TelegramUi.cs @@ -0,0 +1,92 @@ +using System.Collections.Concurrent; +using Telegram.Bot.Types.ReplyMarkups; +using Volcanic.AgenticCore.Abstractions.Ui; +using Volcanic.AgenticCore.Telegram.Rendering; + +namespace Volcanic.AgenticCore.Telegram.Chats; + +/// +/// for one chat: choices become inline keyboards (callback data ui:<id>:<index>), +/// free-text questions a ForceReply prompt answered by the chat's next plain message. +/// +public sealed class TelegramUi(long chatId, TelegramSender sender, ChatRenderer renderer) : IUserInterface +{ + public const string CallbackPrefix = "ui:"; + private static readonly TimeSpan DefaultTimeout = TimeSpan.FromMinutes(10); + + private readonly ConcurrentDictionary> _choices = new(); + private TaskCompletionSource? _input; + + /// True while an question waits for its answer. + public bool IsWaitingForInput => Volatile.Read(ref _input) is { Task.IsCompleted: false }; + + public async Task SelectAsync(string title, IReadOnlyList options, UiRequestOptions? requestOptions, CancellationToken cancellationToken) + { + if (options.Count == 0) return null; + var id = Guid.NewGuid().ToString("N")[..8]; + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _choices[id] = tcs; + try + { + var keyboard = new InlineKeyboardMarkup(options.Select((option, i) => + new[] { InlineKeyboardButton.WithCallbackData(option, $"{CallbackPrefix}{id}:{i}") })); + var messageId = await sender.SendTextAsync(chatId, title, keyboard, cancellationToken).ConfigureAwait(false); + if (messageId is null) return null; + + var index = await WaitAsync(tcs, requestOptions, cancellationToken).ConfigureAwait(false); + var choice = int.TryParse(index, out var i) && i >= 0 && i < options.Count ? options[i] : null; + var outcome = choice is null ? "— no answer" : $"→ {choice}"; + await sender.EditHtmlAsync(chatId, messageId.Value, TelegramFormatter.Escape($"{title}\n{outcome}"), $"{title}\n{outcome}", CancellationToken.None) + .ConfigureAwait(false); + return choice; + } + finally + { + _choices.TryRemove(id, out _); + } + } + + public async Task ConfirmAsync(string title, string message, UiRequestOptions? requestOptions, CancellationToken cancellationToken) => + await SelectAsync($"{title}\n{message}", ["Yes", "No"], requestOptions, cancellationToken).ConfigureAwait(false) == "Yes"; + + public async Task InputAsync(string title, string? placeholder, UiRequestOptions? requestOptions, CancellationToken cancellationToken) + { + var tcs = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + Interlocked.Exchange(ref _input, tcs)?.TrySetResult(null); // one question at a time: a newer one cancels the older + try + { + var markup = new ForceReplyMarkup { InputFieldPlaceholder = placeholder }; + if (await sender.SendTextAsync(chatId, title, markup, cancellationToken).ConfigureAwait(false) is null) return null; + return await WaitAsync(tcs, requestOptions, cancellationToken).ConfigureAwait(false); + } + finally + { + Interlocked.CompareExchange(ref _input, null, tcs); + } + } + + public void Notify(string message, NotifyKind kind = NotifyKind.Info) => + renderer.Post(new Abstractions.Events.AgentSessionEvent.Notification(message, kind)); + + /// Status lines are not shown in Telegram; tool and retry progress is rendered from session events. + public void SetStatus(string key, string? text) { } + + /// Resolve a pending choice from callback data. False when it is not ours or no longer pending. + public bool TryResolveCallback(string data) + { + if (!data.StartsWith(CallbackPrefix, StringComparison.Ordinal)) return false; + var parts = data[CallbackPrefix.Length..].Split(':'); + return parts.Length == 2 && _choices.TryGetValue(parts[0], out var tcs) && tcs.TrySetResult(parts[1]); + } + + /// Answer a pending question. False when none is pending. + public bool TryResolveInput(string text) => Interlocked.Exchange(ref _input, null) is { } tcs && tcs.TrySetResult(text); + + private static async Task WaitAsync(TaskCompletionSource tcs, UiRequestOptions? requestOptions, CancellationToken cancellationToken) + { + using var timeout = new CancellationTokenSource(requestOptions?.Timeout ?? DefaultTimeout); + using var linked = CancellationTokenSource.CreateLinkedTokenSource(timeout.Token, cancellationToken); + await using var registration = linked.Token.Register(() => tcs.TrySetResult(null)).ConfigureAwait(false); + return await tcs.Task.ConfigureAwait(false); + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs b/hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs new file mode 100644 index 0000000..bc36236 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs @@ -0,0 +1,95 @@ +using Microsoft.Extensions.Logging; +using Telegram.Bot; +using Telegram.Bot.Types; +using Telegram.Bot.Types.Enums; +using ImageContent = Volcanic.AgenticCore.Abstractions.Messages.ImageContent; +using Volcanic.AgenticCore.Telegram.Polling; +using Volcanic.AgenticCore.Telegram.Rendering; + +namespace Volcanic.AgenticCore.Telegram.Chats; + +/// +/// Routes an update to its chat: only allow-listed users in private chats get through; button presses and answers go to +/// the chat's pending dialog, bot commands run directly, everything else becomes a prompt that runs in the background. +/// +public sealed class UpdateRouter( + TelegramOptions options, + ChatRegistry chats, + ChatCommands commands, + TelegramSender sender, + ILogger logger) : IUpdateRouter +{ + public async Task RouteAsync(Update update, CancellationToken cancellationToken) + { + switch (update) + { + case { CallbackQuery: { } query }: + await OnCallbackAsync(query, cancellationToken).ConfigureAwait(false); + break; + case { Message: { } message }: + await OnMessageAsync(message, cancellationToken).ConfigureAwait(false); + break; + } + } + + private async Task OnCallbackAsync(CallbackQuery query, CancellationToken cancellationToken) + { + if (!options.IsAllowed(query.From.Id)) + { + logger.LogWarning("Ignored a button press from user {UserId} (not allowed)", query.From.Id); + return; + } + var resolved = query.Message is { } message && query.Data is { } data + && chats.Find(message.Chat.Id) is { } chat && chat.Ui.TryResolveCallback(data); + await sender.AnswerCallbackAsync(query.Id, resolved ? null : "This question has expired.", cancellationToken).ConfigureAwait(false); + } + + private async Task OnMessageAsync(Message message, CancellationToken cancellationToken) + { + if (!options.IsAllowed(message.From?.Id)) + { + logger.LogWarning("Ignored a message from user {UserId} ({Username}) in chat {ChatId}: not in the allow-list", + message.From?.Id, message.From?.Username, message.Chat.Id); + return; + } + if (message.Chat.Type != ChatType.Private) + { + logger.LogInformation("Ignored a message in non-private chat {ChatId}", message.Chat.Id); + return; + } + + var text = message.Text ?? message.Caption ?? ""; + if (text.Length == 0 && message.Photo is not { Length: > 0 }) + { + await sender.SendTextAsync(message.Chat.Id, "I can read text and photos.", cancellationToken: cancellationToken).ConfigureAwait(false); + return; + } + + var chat = await chats.GetAsync(message.Chat.Id, cancellationToken).ConfigureAwait(false); + var isCommand = ChatCommands.Parse(text) is not null; + if (!isCommand && chat.Ui.IsWaitingForInput && chat.Ui.TryResolveInput(text)) return; + if (isCommand && await commands.TryHandleAsync(chat, text, cancellationToken).ConfigureAwait(false)) return; + + var images = await DownloadPhotoAsync(chat, message, cancellationToken).ConfigureAwait(false); + if (text.Length == 0 && images is null) return; // photo download failed and already reported + _ = chat.PromptInBackground(text, images); + } + + private async Task?> DownloadPhotoAsync(ChatContext chat, Message message, CancellationToken cancellationToken) + { + if (message.Photo is not { Length: > 0 } sizes) return null; + try + { + var largest = sizes.MaxBy(p => p.FileSize ?? (long)p.Width * p.Height)!; + using var stream = new MemoryStream(); + await sender.Bot.GetInfoAndDownloadFile(largest.FileId, stream, cancellationToken).ConfigureAwait(false); + return [new ImageContent(Convert.ToBase64String(stream.ToArray()), "image/jpeg")]; + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogWarning(ex, "Downloading a photo in chat {ChatId} failed", chat.ChatId); + chat.Report("I could not download that photo."); + return null; + } + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs b/hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs new file mode 100644 index 0000000..3ea1360 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs @@ -0,0 +1,153 @@ +using System.Net; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Telegram.Bot; +using Telegram.Bot.Args; +using Telegram.Bot.Exceptions; +using Telegram.Bot.Polling; +using Telegram.Bot.Requests; +using Telegram.Bot.Types; +using Telegram.Bot.Types.Enums; + +namespace Volcanic.AgenticCore.Telegram.Polling; + +/// Receives one update. Must return quickly: the polling loop waits for it before the next update. +public interface IUpdateRouter +{ + Task RouteAsync(Update update, CancellationToken cancellationToken); +} + +/// +/// Runs Telegram.Bot's built-in long-polling loop and makes it survive connection loss. +/// +/// The built-in loop polls again right after a failed getUpdates (a busy loop while offline) and ends for good +/// if the error handler throws. So the error handler here never throws and waits out a backoff delay, which is what +/// holds the loop back; a successful getUpdates response resets the backoff. 401 is the only fatal error. +/// If the loop still ends while the host runs, it is restarted from the last offset, so no update is lost or repeated. +/// +/// +public sealed class PollingService( + ITelegramBotClient bot, + IUpdateRouter router, + ReconnectPolicy policy, + IHostApplicationLifetime lifetime, + ILogger logger) : BackgroundService, IUpdateHandler +{ + internal static readonly UpdateType[] AllowedUpdates = [UpdateType.Message, UpdateType.CallbackQuery]; + + private int _offset; + + /// The next update id to request; everything below it has been handed to the router. + public int Offset => Volatile.Read(ref _offset); + + /// Set when polling stopped the application because retrying cannot help (a rejected token). + public Exception? FatalError { get; private set; } + + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + bot.OnApiResponseReceived += OnApiResponseReceived; + try + { + while (!stoppingToken.IsCancellationRequested) + { + try + { + logger.LogInformation("Polling Telegram for updates"); + await bot.ReceiveAsync(this, new ReceiverOptions { Offset = Offset, AllowedUpdates = AllowedUpdates }, stoppingToken) + .ConfigureAwait(false); + } + catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested) + { + return; + } + catch (FatalPollingException ex) + { + logger.LogCritical(ex.InnerException, "{Reason}. Stopping.", ex.Message); + FatalError = ex; + lifetime.StopApplication(); + return; + } + catch (Exception ex) + { + logger.LogError(ex, "The Telegram polling loop failed; restarting it"); + } + + if (stoppingToken.IsCancellationRequested) return; + if (!await DelayAsync(policy.NextDelay(), stoppingToken).ConfigureAwait(false)) return; + } + } + finally + { + bot.OnApiResponseReceived -= OnApiResponseReceived; + } + } + + public async Task HandleUpdateAsync(ITelegramBotClient botClient, Update update, CancellationToken cancellationToken) + { + Volatile.Write(ref _offset, update.Id + 1); + MarkConnected(); + try + { + await router.RouteAsync(update, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogError(ex, "Handling update {UpdateId} failed", update.Id); + } + } + + public async Task HandleErrorAsync(ITelegramBotClient botClient, Exception exception, HandleErrorSource source, CancellationToken cancellationToken) + { + if (source != HandleErrorSource.PollingError) + { + logger.LogError(exception, "Telegram update handling failed ({Source})", source); + return; + } + + if (exception is ApiRequestException { ErrorCode: (int)HttpStatusCode.Unauthorized }) + throw new FatalPollingException("Telegram rejected the bot token (401 Unauthorized)", exception); + + var delay = policy.NextDelay(); + var failures = policy.Failures; + if (exception is ApiRequestException { ErrorCode: (int)HttpStatusCode.Conflict }) + logger.LogWarning("Telegram reports another poller or a webhook for this bot (409): {Message}. Retrying in {Delay:0.#} s", + exception.Message, delay.TotalSeconds); + else if (failures == 1) + // A single failure is usually the long-poll timeout racing the HTTP timeout; only repeated ones matter. + logger.LogDebug(exception, "Polling Telegram failed; retrying in {Delay:0.#} s", delay.TotalSeconds); + else + logger.LogWarning("Telegram unreachable (attempt {Attempt}): {Message}. Retrying in {Delay:0.#} s", + failures, exception.Message, delay.TotalSeconds); + + await DelayAsync(delay, cancellationToken).ConfigureAwait(false); + } + + private ValueTask OnApiResponseReceived(ITelegramBotClient botClient, ApiResponseEventArgs args, CancellationToken cancellationToken) + { + if (args.ApiRequestEventArgs.Request is GetUpdatesRequest && args.ResponseMessage.IsSuccessStatusCode) MarkConnected(); + return ValueTask.CompletedTask; + } + + private void MarkConnected() + { + var failures = policy.Reset(); + if (failures > 1) logger.LogInformation("Telegram connection restored after {Failures} failed attempts", failures); + } + + /// False when the host is stopping. + private static async Task DelayAsync(TimeSpan delay, CancellationToken cancellationToken) + { + try + { + await Task.Delay(delay, cancellationToken).ConfigureAwait(false); + return true; + } + catch (OperationCanceledException) + { + return false; + } + } +} + +/// A polling error retrying cannot fix. +public sealed class FatalPollingException(string message, Exception inner) : Exception(message, inner); diff --git a/hosts/Volcanic.AgenticCore.Telegram/Polling/ReconnectPolicy.cs b/hosts/Volcanic.AgenticCore.Telegram/Polling/ReconnectPolicy.cs new file mode 100644 index 0000000..c9e10b9 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Polling/ReconnectPolicy.cs @@ -0,0 +1,27 @@ +namespace Volcanic.AgenticCore.Telegram.Polling; + +/// +/// Exponential backoff with jitter for consecutive polling failures. Thread-safe: failures are counted by the polling +/// loop, the reset comes from the HTTP response callback. +/// +public sealed class ReconnectPolicy(TimeSpan initialDelay, TimeSpan maxDelay, double jitter = 0.2, Random? random = null) +{ + private readonly Random _random = random ?? Random.Shared; + private int _failures; + + /// Consecutive failures since the last success. + public int Failures => Volatile.Read(ref _failures); + + /// Record a failure and return how long to wait before the next attempt. + public TimeSpan NextDelay() + { + var failures = Interlocked.Increment(ref _failures); + var exponent = Math.Min(failures - 1, 30); + var ms = Math.Min(maxDelay.TotalMilliseconds, initialDelay.TotalMilliseconds * Math.Pow(2, exponent)); + var factor = 1 + ((_random.NextDouble() * 2) - 1) * jitter; + return TimeSpan.FromMilliseconds(Math.Max(0, ms * factor)); + } + + /// Record a success. Returns the number of failures it ends (0 when the connection was healthy). + public int Reset() => Interlocked.Exchange(ref _failures, 0); +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Program.cs b/hosts/Volcanic.AgenticCore.Telegram/Program.cs new file mode 100644 index 0000000..3ae4c51 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Program.cs @@ -0,0 +1,55 @@ +// Telegram front end for the agent: long polling, one agent session per private chat, Claude Code as the model. +// +// TELEGRAM_BOT_TOKEN=123:abc TELEGRAM_ALLOWED_USER_IDS=111111 dotnet run --project hosts/Volcanic.AgenticCore.Telegram + +using Microsoft.Extensions.Configuration; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Telegram.Bot; +using Volcanic.AgenticCore.Extensions.ClaudeCode; +using Volcanic.AgenticCore.Hosting; +using Volcanic.AgenticCore.Telegram; +using Volcanic.AgenticCore.Telegram.Chats; +using Volcanic.AgenticCore.Telegram.Polling; +using Volcanic.AgenticCore.Telegram.Rendering; + +var builder = Host.CreateApplicationBuilder(args); + +var options = builder.Configuration.GetSection(TelegramOptions.SectionName).Get() ?? new TelegramOptions(); +options.ApplyEnvironment(Environment.GetEnvironmentVariable); +var errors = options.Validate(); +if (errors.Count > 0) +{ + foreach (var error in errors) Console.Error.WriteLine(error); + return 2; +} + +builder.Services.AddSingleton(options); +builder.Services.AddAgenticCore(o => +{ + o.DefaultModel = options.Model; + o.AddExtension(); +}); +builder.Services.AddSingleton(_ => new TelegramBotClient(options.BotToken)); +builder.Services.AddSingleton(new ReconnectPolicy(options.ReconnectInitialDelay, options.ReconnectMaxDelay)); +builder.Services.AddSingleton(); +builder.Services.AddSingleton(); +builder.Services.AddSingleton(); +builder.Services.AddSingleton(); +builder.Services.AddSingleton(); +builder.Services.AddHostedService(sp => sp.GetRequiredService()); +builder.Services.AddHostedService(); + +using var host = builder.Build(); + +var logger = host.Services.GetRequiredService().CreateLogger("Volcanic.AgenticCore.Telegram"); +if (ClaudeInstallation.Locate(Environment.GetEnvironmentVariable) is { } claude) + logger.LogInformation("Using Claude Code at {Path}, model {Model}", claude, options.Model); +else + logger.LogWarning("Claude Code (claude) was not found on PATH; requests will fail until it is installed and logged in. Send /claude-code-doctor to the bot for details."); +logger.LogInformation("Allowed users: {Users}. Workspaces under {Root}", string.Join(", ", options.AllowedUserIds), options.WorkspaceRoot); + +var polling = host.Services.GetRequiredService(); +await host.RunAsync(); // disposes the host +return polling.FatalError is null ? 0 : 1; diff --git a/hosts/Volcanic.AgenticCore.Telegram/Rendering/ChatRenderer.cs b/hosts/Volcanic.AgenticCore.Telegram/Rendering/ChatRenderer.cs new file mode 100644 index 0000000..2d0eff2 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Rendering/ChatRenderer.cs @@ -0,0 +1,309 @@ +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: streamed text edits one message (split into several past the size limit) at most every +/// , tool calls become status lines, and "typing…" repeats while the agent runs. +/// +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 _events = Channel.CreateUnbounded(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 _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); + } + + /// Queue an event; safe to call from any thread (this is the session subscription). + public void Post(AgentSessionEvent e) + { + Interlocked.Increment(ref _pending); + if (!_events.Writer.TryWrite(e)) Interlocked.Decrement(ref _pending); + } + + /// Completes once every queued event is rendered. For tests. + 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($"▶ {TelegramFormatter.Escape(start.ToolName)}{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 + ? $"✗ {TelegramFormatter.Escape(end.ToolName)}{tool.Args}{FormatExcerpt(end.Result)}" + : $"✓ {TelegramFormatter.Escape(end.ToolName)}{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 SendLineAsync(string text, CancellationToken ct) => + _sender.SendHtmlAsync(_chatId, TelegramFormatter.Escape(text), text, cancellationToken: ct); + + private Task 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(); + } + + /// ": ls -la" for bash, the most telling argument otherwise. + 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 $": {TelegramFormatter.Escape(Shorten(summary, 200))}"; + } + + private static string FormatExcerpt(ToolResult result) + { + var text = string.Concat(result.Content.OfType().Select(c => c.Text)).Trim(); + return text.Length == 0 ? "" : $"\n
{TelegramFormatter.Escape(Shorten(text, ToolExcerptLength))}
"; + } + + 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(); + } +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramFormatter.cs b/hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramFormatter.cs new file mode 100644 index 0000000..2b0d497 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramFormatter.cs @@ -0,0 +1,187 @@ +using System.Text; +using System.Text.RegularExpressions; + +namespace Volcanic.AgenticCore.Telegram.Rendering; + +/// +/// Turns the model's Markdown into Telegram HTML and splits long text into message-sized chunks. +/// The output is always well formed: everything is escaped and tags are only emitted in matched pairs, so partial +/// (still streaming) Markdown renders too. +/// +public static partial class TelegramFormatter +{ + /// Telegram's limit for one message, in UTF-16 code units after entity parsing. + public const int MessageLimit = 4096; + + /// Raw text per chunk; leaves room for the HTML tags and escapes the conversion adds. + public const int ChunkLimit = 3500; + + private const string Fence = "```"; + + /// Escape the three characters Telegram HTML requires; everything else (emoji included) stays as is. + public static string Escape(string text) => text.Replace("&", "&").Replace("<", "<").Replace(">", ">"); + + /// Convert Markdown to Telegram HTML. + public static string ToHtml(string markdown) + { + var html = new StringBuilder(); + var lines = markdown.Replace("\r\n", "\n").Split('\n'); + var inCode = false; + string? language = null; + var code = new StringBuilder(); + + void CloseCode() + { + var body = Escape(code.ToString().TrimEnd('\n')); + html.Append(language is { Length: > 0 } + ? $"
{body}
" + : $"
{body}
"); + code.Clear(); + } + + for (var i = 0; i < lines.Length; i++) + { + var line = lines[i]; + if (line.TrimStart().StartsWith(Fence, StringComparison.Ordinal)) + { + if (inCode) + { + CloseCode(); + inCode = false; + } + else + { + inCode = true; + language = line.TrimStart()[Fence.Length..].Trim(); + } + if (i < lines.Length - 1 && !inCode) html.Append('\n'); + continue; + } + + if (inCode) + { + code.Append(line).Append('\n'); + continue; + } + + html.Append(FormatLine(line)); + if (i < lines.Length - 1) html.Append('\n'); + } + + if (inCode) CloseCode(); // unterminated fence while streaming: show what we have as code + return html.ToString(); + } + + private static string FormatLine(string line) + { + if (HeadingRegex().Match(line) is { Success: true } heading) return $"{FormatInline(heading.Groups[1].Value)}"; + if (BulletRegex().Match(line) is { Success: true } bullet) return $"{bullet.Groups[1].Value}• {FormatInline(bullet.Groups[2].Value)}"; + return FormatInline(line); + } + + /// Inline code, bold, italics, strikethrough and links. Code spans are cut out first so their content stays literal. + private static string FormatInline(string text) + { + var result = new StringBuilder(); + var last = 0; + foreach (Match m in InlineCodeRegex().Matches(text)) + { + result.Append(FormatEmphasis(text[last..m.Index])); + result.Append("").Append(Escape(m.Groups[1].Value)).Append(""); + last = m.Index + m.Length; + } + result.Append(FormatEmphasis(text[last..])); + return result.ToString(); + } + + private static string FormatEmphasis(string text) + { + var s = Escape(text); + s = LinkRegex().Replace(s, m => $"{m.Groups[1].Value}"); + s = BoldRegex().Replace(s, "$2"); + s = StrikeRegex().Replace(s, "$1"); + s = ItalicRegex().Replace(s, "$1$3"); + return s; + } + + /// + /// Split raw Markdown into chunks of at most characters, preferring paragraph, line and + /// word boundaries. A code fence cut by a split is closed at the end of one chunk and reopened in the next. + /// + public static IReadOnlyList Split(string text, int limit = ChunkLimit) + { + if (text.Length <= limit) return [text]; + var chunks = new List(); + var rest = text; + string? reopen = null; + while (rest.Length > 0) + { + var prefix = reopen is null ? "" : reopen + "\n"; + var budget = limit - prefix.Length - (Fence.Length + 1); // room to close a fence + if (prefix.Length + rest.Length <= limit) + { + chunks.Add(prefix + rest); + break; + } + + var (cut, skip) = FindCut(rest, budget); + var piece = rest[..cut]; + rest = rest[(cut + skip)..].TrimStart('\n'); + + var openFence = OpenFence(prefix + piece); + var chunk = prefix + piece.TrimEnd('\n'); + if (openFence is not null) chunk += "\n" + Fence; + chunks.Add(chunk); + reopen = openFence; + } + return chunks; + } + + /// Where the chunk ends and how many separator characters to drop after it. + private static (int Cut, int Skip) FindCut(string text, int budget) + { + budget = Math.Max(1, Math.Min(budget, text.Length)); + var window = text[..budget]; + foreach (var separator in new[] { "\n\n", "\n", " " }) + { + var at = window.LastIndexOf(separator, StringComparison.Ordinal); + if (at > budget / 2) return (at, separator.Length); + } + // Do not split a surrogate pair. + return (char.IsHighSurrogate(text[budget - 1]) ? budget - 1 : budget, 0); + } + + /// The opening fence line (with language) when ends inside a code block. + private static string? OpenFence(string text) + { + string? open = null; + foreach (var line in text.Split('\n')) + { + var trimmed = line.TrimStart(); + if (!trimmed.StartsWith(Fence, StringComparison.Ordinal)) continue; + open = open is null ? trimmed : null; + } + return open; + } + + [GeneratedRegex(@"^\s{0,3}#{1,6}\s+(.+?)\s*#*\s*$")] + private static partial Regex HeadingRegex(); + + [GeneratedRegex(@"^(\s*)[-*+]\s+(.*)$")] + private static partial Regex BulletRegex(); + + [GeneratedRegex(@"`([^`\n]+)`")] + private static partial Regex InlineCodeRegex(); + + [GeneratedRegex(@"\[([^\]\n]+)\]\((https?://[^\s)""]+)\)")] + private static partial Regex LinkRegex(); + + [GeneratedRegex(@"(\*\*|__)(?=\S)(.+?)(?<=\S)\1")] + private static partial Regex BoldRegex(); + + [GeneratedRegex(@"~~(?=\S)(.+?)(?<=\S)~~")] + private static partial Regex StrikeRegex(); + + [GeneratedRegex(@"(^|[^\w*])([*_])(?=\S)([^*_\n]+?)(?<=\S)\2(?![\w*])")] + private static partial Regex ItalicRegex(); +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramSender.cs b/hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramSender.cs new file mode 100644 index 0000000..87bbee5 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramSender.cs @@ -0,0 +1,126 @@ +using Microsoft.Extensions.Logging; +using Telegram.Bot; +using Telegram.Bot.Exceptions; +using Telegram.Bot.Types; +using Telegram.Bot.Types.Enums; +using Telegram.Bot.Types.ReplyMarkups; + +namespace Volcanic.AgenticCore.Telegram.Rendering; + +/// +/// Sends and edits chat messages without ever throwing (except on cancellation): a failed call is logged and reported +/// as null/false. HTML Telegram rejects is resent as plain text; "message is not modified" counts as success. +/// +public sealed class TelegramSender(ITelegramBotClient bot, ILogger logger) +{ + public ITelegramBotClient Bot => bot; + + /// Send HTML; returns the message id, or null when sending failed. + public async Task SendHtmlAsync(long chatId, string html, string plain, ReplyMarkup? replyMarkup = null, CancellationToken cancellationToken = default) + { + try + { + var message = await bot.SendMessage(chatId, html, parseMode: ParseMode.Html, replyMarkup: replyMarkup, + linkPreviewOptions: NoPreview, cancellationToken: cancellationToken).ConfigureAwait(false); + return message.Id; + } + catch (ApiRequestException ex) when (IsBadMarkup(ex)) + { + return await SendTextAsync(chatId, plain, replyMarkup, cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogWarning(ex, "Sending a message to chat {ChatId} failed", chatId); + return null; + } + } + + /// Send plain text; returns the message id, or null when sending failed. + public async Task SendTextAsync(long chatId, string text, ReplyMarkup? replyMarkup = null, CancellationToken cancellationToken = default) + { + try + { + var message = await bot.SendMessage(chatId, Truncate(text), replyMarkup: replyMarkup, + linkPreviewOptions: NoPreview, cancellationToken: cancellationToken).ConfigureAwait(false); + return message.Id; + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogWarning(ex, "Sending a message to chat {ChatId} failed", chatId); + return null; + } + } + + /// Replace a message's text with HTML (plain fallback). Returns false when the edit failed. + public async Task EditHtmlAsync(long chatId, int messageId, string html, string plain, CancellationToken cancellationToken = default) + { + try + { + await bot.EditMessageText(chatId, messageId, html, parseMode: ParseMode.Html, + linkPreviewOptions: NoPreview, cancellationToken: cancellationToken).ConfigureAwait(false); + return true; + } + catch (ApiRequestException ex) when (IsNotModified(ex)) + { + return true; + } + catch (ApiRequestException ex) when (IsBadMarkup(ex)) + { + try + { + await bot.EditMessageText(chatId, messageId, Truncate(plain), linkPreviewOptions: NoPreview, + cancellationToken: cancellationToken).ConfigureAwait(false); + return true; + } + catch (ApiRequestException inner) when (IsNotModified(inner)) + { + return true; + } + catch (Exception inner) when (inner is not OperationCanceledException) + { + logger.LogWarning(inner, "Editing message {MessageId} in chat {ChatId} failed", messageId, chatId); + return false; + } + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogWarning(ex, "Editing message {MessageId} in chat {ChatId} failed", messageId, chatId); + return false; + } + } + + public async Task SendTypingAsync(long chatId, CancellationToken cancellationToken = default) + { + try + { + await bot.SendChatAction(chatId, ChatAction.Typing, cancellationToken: cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogDebug(ex, "Sending the typing action to chat {ChatId} failed", chatId); + } + } + + public async Task AnswerCallbackAsync(string callbackQueryId, string? text = null, CancellationToken cancellationToken = default) + { + try + { + await bot.AnswerCallbackQuery(callbackQueryId, text, cancellationToken: cancellationToken).ConfigureAwait(false); + } + catch (Exception ex) when (ex is not OperationCanceledException) + { + logger.LogDebug(ex, "Answering callback query {Id} failed", callbackQueryId); + } + } + + private static readonly LinkPreviewOptions NoPreview = new() { IsDisabled = true }; + + private static string Truncate(string text) => + text.Length <= TelegramFormatter.MessageLimit ? text : text[..(TelegramFormatter.MessageLimit - 1)] + "…"; + + private static bool IsBadMarkup(ApiRequestException ex) => + ex.ErrorCode == 400 && ex.Message.Contains("parse entities", StringComparison.OrdinalIgnoreCase); + + private static bool IsNotModified(ApiRequestException ex) => + ex.ErrorCode == 400 && ex.Message.Contains("message is not modified", StringComparison.OrdinalIgnoreCase); +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/TelegramOptions.cs b/hosts/Volcanic.AgenticCore.Telegram/TelegramOptions.cs new file mode 100644 index 0000000..c6250bf --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/TelegramOptions.cs @@ -0,0 +1,62 @@ +namespace Volcanic.AgenticCore.Telegram; + +/// The Telegram configuration section. +public sealed class TelegramOptions +{ + public const string SectionName = "Telegram"; + public const string DefaultModel = "claude-code/claude-opus-5-5"; + + /// Bot token from @BotFather. Also read from TELEGRAM_BOT_TOKEN. + public string BotToken { get; set; } = ""; + + /// Telegram user ids allowed to talk to the bot. Required: the agent runs shell commands. + public List AllowedUserIds { get; set; } = []; + + /// Each chat works in <WorkspaceRoot>/<chatId>. + public string WorkspaceRoot { get; set; } = Path.Combine( + Environment.GetFolderPath(Environment.SpecialFolder.UserProfile), ".volcanic", "telegram-workspaces"); + + /// provider/model-id every chat uses. + public string Model { get; set; } = DefaultModel; + + /// Minimum time between two edits of a streaming reply (Telegram rate-limits edits). + public TimeSpan EditInterval { get; set; } = TimeSpan.FromSeconds(1.5); + + /// How often the "typing…" action is repeated while the agent runs (Telegram shows it for 5 s). + public TimeSpan TypingInterval { get; set; } = TimeSpan.FromSeconds(4); + + /// First delay after a failed poll; doubles per consecutive failure up to . + public TimeSpan ReconnectInitialDelay { get; set; } = TimeSpan.FromSeconds(1); + + public TimeSpan ReconnectMaxDelay { get; set; } = TimeSpan.FromSeconds(60); + + /// Reads the environment-variable shortcuts the README documents, when the section leaves them empty. + public void ApplyEnvironment(Func getEnv) + { + if (string.IsNullOrWhiteSpace(BotToken)) BotToken = getEnv("TELEGRAM_BOT_TOKEN") ?? ""; + if (AllowedUserIds.Count == 0 && getEnv("TELEGRAM_ALLOWED_USER_IDS") is { Length: > 0 } ids) + { + foreach (var part in ids.Split([',', ';', ' '], StringSplitOptions.RemoveEmptyEntries)) + if (long.TryParse(part, out var id)) AllowedUserIds.Add(id); + } + } + + /// Configuration problems that must stop the host, empty when valid. + public IReadOnlyList Validate() + { + var errors = new List(); + if (string.IsNullOrWhiteSpace(BotToken)) + errors.Add("Telegram bot token is missing: set TELEGRAM_BOT_TOKEN or Telegram:BotToken."); + if (AllowedUserIds.Count == 0) + errors.Add("No allowed users: set TELEGRAM_ALLOWED_USER_IDS (comma separated) or Telegram:AllowedUserIds. " + + "The bot runs shell commands, so it refuses to start open to everyone."); + if (string.IsNullOrWhiteSpace(WorkspaceRoot)) errors.Add("Telegram:WorkspaceRoot is empty."); + if (string.IsNullOrWhiteSpace(Model) || !Model.Contains('/')) errors.Add("Telegram:Model must be 'provider/model-id'."); + if (EditInterval <= TimeSpan.Zero || TypingInterval <= TimeSpan.Zero) errors.Add("Telegram intervals must be positive."); + if (ReconnectInitialDelay <= TimeSpan.Zero || ReconnectMaxDelay < ReconnectInitialDelay) + errors.Add("Telegram reconnect delays must be positive with max >= initial."); + return errors; + } + + public bool IsAllowed(long? userId) => userId is { } id && AllowedUserIds.Contains(id); +} diff --git a/hosts/Volcanic.AgenticCore.Telegram/Volcanic.AgenticCore.Telegram.csproj b/hosts/Volcanic.AgenticCore.Telegram/Volcanic.AgenticCore.Telegram.csproj new file mode 100644 index 0000000..4bfbbf7 --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/Volcanic.AgenticCore.Telegram.csproj @@ -0,0 +1,16 @@ + + + Exe + volcanic-agentic-core-telegram + + + + + + + + + + + + diff --git a/hosts/Volcanic.AgenticCore.Telegram/appsettings.json b/hosts/Volcanic.AgenticCore.Telegram/appsettings.json new file mode 100644 index 0000000..89b1bfd --- /dev/null +++ b/hosts/Volcanic.AgenticCore.Telegram/appsettings.json @@ -0,0 +1,14 @@ +{ + "Logging": { + "LogLevel": { + "Default": "Information", + "Microsoft.Hosting.Lifetime": "Warning", + "System.Net.Http": "Warning" + } + }, + "Telegram": { + "BotToken": "", + "AllowedUserIds": [], + "Model": "claude-code/claude-opus-5-5" + } +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/ChatTests.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/ChatTests.cs new file mode 100644 index 0000000..71b8b03 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/ChatTests.cs @@ -0,0 +1,220 @@ +using Telegram.Bot.Requests; +using Telegram.Bot.Types.Enums; +using Telegram.Bot.Types.ReplyMarkups; +using Volcanic.AgenticCore.Telegram.Chats; +using Volcanic.AgenticCore.Telegram.Tests.Support; +using Volcanic.AgenticCore.Testing; + +namespace Volcanic.AgenticCore.Telegram.Tests; + +public sealed class ChatTests +{ + [Fact] + public async Task A_prompt_streams_into_one_reply_message() + { + await using var host = new ChatHost(); + host.Faux.Enqueue(FauxTurn.Say("Hello **there**, how can I help?")); + + await host.SendAsync("hi"); + await host.SettleAsync(); + + var reply = Assert.Single(host.Bot.Sent); + Assert.Equal(ParseMode.Html, reply.ParseMode); + Assert.Equal("Hello there, how can I help?", host.Bot.CurrentTexts.Values.Single()); + Assert.Contains(host.Bot.Requests, r => r is SendChatActionRequest); + Assert.Equal("hi", host.Faux.Requests.Single().Messages.Last().Text); + } + + [Fact] + public async Task Messages_from_users_outside_the_allow_list_are_ignored() + { + await using var host = new ChatHost(); + + await host.SendAsync("let me in", userId: Updates.Stranger); + + Assert.Empty(host.Bot.Requests); + Assert.Null(host.Chats.Find(Updates.Stranger)); + Assert.Empty(host.Faux.Requests); + } + + [Fact] + public async Task Group_chats_are_ignored() + { + await using var host = new ChatHost(); + + await host.Router.RouteAsync(Updates.Text("hi all", chatId: -100, chatType: ChatType.Group), CancellationToken.None); + + Assert.Empty(host.Bot.Requests); + Assert.Empty(host.Faux.Requests); + } + + [Fact] + public async Task A_long_reply_is_split_across_messages() + { + await using var host = new ChatHost(); + var words = string.Join(" ", Enumerable.Range(0, 2000).Select(i => $"word{i}")); + host.Faux.Enqueue(FauxTurn.Say(words)); + + await host.SendAsync("talk a lot"); + await host.SettleAsync(); + + var texts = host.Texts.ToList(); + Assert.True(texts.Count >= 4, $"expected several messages, got {texts.Count}"); + Assert.All(texts, t => Assert.True(t.Length <= 4096)); + Assert.Equal(words, string.Join(" ", texts)); + } + + [Fact] + public async Task Tool_calls_show_a_status_line_that_is_completed() + { + await using var host = new ChatHost(); + host.Faux.Enqueue(FauxTurn.Tool("ls", new { path = "." }), FauxTurn.Say("Nothing here.")); + + await host.SendAsync("what is here?"); + await host.SettleAsync(); + + var texts = host.Texts.ToList(); + Assert.Contains("✓ ls: .", texts); + Assert.Equal("Nothing here.", texts.Last()); + Assert.Contains(host.Bot.Sent, s => s.Text.StartsWith("▶ ls", StringComparison.Ordinal)); + } + + [Fact] + public async Task Stop_aborts_a_running_prompt() + { + await using var host = new ChatHost(); + host.Faux.Enqueue(FauxTurn.Say("too late").WithDelay(TimeSpan.FromSeconds(30))); + + await host.SendAsync("take your time"); + var chat = await host.ChatAsync(); + await Wait.UntilAsync(() => chat.Session.IsStreaming, "the run to start"); + + await host.SendAsync("/stop"); + await host.SettleAsync(); + + Assert.Contains("⏹ Stopped.", host.Texts); + Assert.DoesNotContain(host.Texts, t => t.Contains("too late", StringComparison.Ordinal)); + } + + [Fact] + public async Task A_message_while_running_steers_instead_of_failing() + { + await using var host = new ChatHost(); + host.Faux.Enqueue(FauxTurn.Tool("ls", new { }).WithDelay(TimeSpan.FromMilliseconds(300)), FauxTurn.Say("Adjusted.")); + + await host.SendAsync("list files"); + var chat = await host.ChatAsync(); + await Wait.UntilAsync(() => chat.Session.IsStreaming, "the run to start"); + await host.SendAsync("actually, also say hi"); + await host.SettleAsync(); + + Assert.DoesNotContain(host.Texts, t => t.Contains("failed", StringComparison.OrdinalIgnoreCase)); + Assert.Contains(host.Faux.Requests.Last().Messages, m => m.Text == "actually, also say hi"); + } + + [Fact] + public async Task New_starts_a_fresh_session() + { + await using var host = new ChatHost(); + host.Faux.Enqueue(FauxTurn.Say("one")); + await host.SendAsync("first"); + await host.SettleAsync(); + var chat = await host.ChatAsync(); + var firstId = chat.Session.SessionId; + + await host.SendAsync("/new"); + + Assert.NotEqual(firstId, chat.Session.SessionId); + Assert.Empty(chat.Session.Messages); + Assert.Contains("🆕 New conversation.", host.Texts); + } + + [Fact] + public async Task Status_reports_the_model() + { + await using var host = new ChatHost(); + + await host.SendAsync("/status@fake_bot"); + + Assert.Contains(host.Texts, t => t.Contains(FauxProvider.ModelKey, StringComparison.Ordinal)); + Assert.Empty(host.Faux.Requests); + } + + [Fact] + public async Task Unknown_commands_go_to_the_session() + { + await using var host = new ChatHost(); + host.Faux.Enqueue(FauxTurn.Say("ok")); + + await host.SendAsync("/whatever please"); + await host.SettleAsync(); + + Assert.Equal("/whatever please", host.Faux.Requests.Single().Messages.Last().Text); + } + + [Fact] + public async Task Select_is_answered_with_an_inline_keyboard_button() + { + await using var host = new ChatHost(); + await host.SendAsync("/status"); // creates the chat + var chat = await host.ChatAsync(); + + var answer = chat.Ui.SelectAsync("Pick one", ["red", "green"], null, CancellationToken.None); + await Wait.UntilAsync(() => host.Bot.Sent.Any(s => s.ReplyMarkup is InlineKeyboardMarkup), "the keyboard"); + var keyboard = (InlineKeyboardMarkup)host.Bot.Sent.Last(s => s.ReplyMarkup is InlineKeyboardMarkup).ReplyMarkup!; + var green = keyboard.InlineKeyboard.SelectMany(row => row).Single(b => b.Text == "green"); + + await host.Router.RouteAsync(Updates.Callback(green.CallbackData!, Updates.AllowedUser), CancellationToken.None); + + Assert.Equal("green", await answer.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Contains(host.Bot.Requests, r => r is AnswerCallbackQueryRequest); + Assert.Contains("Pick one\n→ green", host.Texts); + } + + [Fact] + public async Task Input_is_answered_by_the_next_plain_message() + { + await using var host = new ChatHost(); + await host.SendAsync("/status"); + var chat = await host.ChatAsync(); + + var answer = chat.Ui.InputAsync("Your name?", "name", null, CancellationToken.None); + await Wait.UntilAsync(() => chat.Ui.IsWaitingForInput, "the question"); + await host.SendAsync("Ada"); + + Assert.Equal("Ada", await answer.WaitAsync(TimeSpan.FromSeconds(5))); + Assert.Empty(host.Faux.Requests); // the answer is not a prompt + } + + [Fact] + public async Task Buttons_from_strangers_are_ignored() + { + await using var host = new ChatHost(); + await host.SendAsync("/status"); + var chat = await host.ChatAsync(); + var answer = chat.Ui.SelectAsync("Pick", ["a"], new(TimeSpan.FromMilliseconds(300)), CancellationToken.None); + await Wait.UntilAsync(() => host.Bot.Sent.Any(s => s.ReplyMarkup is InlineKeyboardMarkup), "the keyboard"); + var data = ((InlineKeyboardMarkup)host.Bot.Sent.Last().ReplyMarkup!).InlineKeyboard.First().First().CallbackData!; + + await host.Router.RouteAsync(Updates.Callback(data, Updates.AllowedUser, userId: Updates.Stranger), CancellationToken.None); + + Assert.Null(await answer.WaitAsync(TimeSpan.FromSeconds(5))); + } + + [Fact] + public async Task Telegram_send_failures_do_not_break_rendering() + { + await using var host = new ChatHost(); + var failures = 0; + host.Bot.OnSend = r => + { + if (r is SendChatActionRequest && Interlocked.Increment(ref failures) < 100) throw FakeBotClient.NetworkDown(); + }; + host.Faux.Enqueue(FauxTurn.Say("still here")); + + await host.SendAsync("hello?"); + await host.SettleAsync(); + + Assert.Contains("still here", host.Texts); + } +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/FormattingAndOptionsTests.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/FormattingAndOptionsTests.cs new file mode 100644 index 0000000..448c579 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/FormattingAndOptionsTests.cs @@ -0,0 +1,87 @@ +using Volcanic.AgenticCore.Telegram.Chats; +using Volcanic.AgenticCore.Telegram.Rendering; + +namespace Volcanic.AgenticCore.Telegram.Tests; + +public sealed class FormattingAndOptionsTests +{ + [Theory] + [InlineData("a < b & c > d", "a < b & c > d")] + [InlineData("**bold** and *it* and _it_", "bold and it and it")] + [InlineData("use `x < 1` here", "use x < 1 here")] + [InlineData("`**not bold**`", "**not bold**")] + [InlineData("snake_case_name stays", "snake_case_name stays")] + [InlineData("# Title", "Title")] + [InlineData("- item", "• item")] + [InlineData("[docs](https://example.com/a?b=1&c=2)", "docs")] + [InlineData("~~gone~~", "gone")] + [InlineData("🆕 Привет \"quoted\" it's", "🆕 Привет \"quoted\" it's")] + public void Markdown_becomes_telegram_html(string markdown, string html) => + Assert.Equal(html, TelegramFormatter.ToHtml(markdown)); + + [Fact] + public void Code_blocks_keep_their_content_literal() + { + var html = TelegramFormatter.ToHtml("Run:\n```bash\necho \"\" **x**\n```\nDone."); + Assert.Equal("Run:\n
echo \"<hi>\" **x**
\nDone.", html); + } + + [Fact] + public void An_unterminated_code_block_while_streaming_is_still_well_formed() + { + Assert.Equal("
partial <
", TelegramFormatter.ToHtml("```\npartial <")); + } + + [Fact] + public void Split_prefers_paragraphs_and_reopens_cut_code_blocks() + { + var code = string.Join("\n", Enumerable.Range(0, 40).Select(i => $"line {i}")); + var text = "intro\n\n```python\n" + code + "\n```\nafter"; + + var chunks = TelegramFormatter.Split(text, limit: 120); + + Assert.True(chunks.Count > 2); + Assert.All(chunks, c => Assert.True(c.Length <= 120, $"{c.Length}: {c}")); + Assert.All(chunks, c => Assert.Equal(0, c.Split('\n').Count(l => l.StartsWith("```", StringComparison.Ordinal)) % 2)); + Assert.StartsWith("```python\n", chunks[1]); + Assert.Equal(text.Replace("```python\n", "").Replace("```\n", "").Replace("\n", ""), + string.Concat(chunks.Select(c => string.Join("", c.Split('\n').Where(l => !l.StartsWith("```", StringComparison.Ordinal)))))); + } + + [Fact] + public void Short_text_is_one_chunk() => Assert.Equal(["hi"], TelegramFormatter.Split("hi")); + + [Theory] + [InlineData("/stop", "stop", "")] + [InlineData("/resume@my_bot abc", "resume", "abc")] + [InlineData("/Thinking high ", "thinking", "high")] + public void Commands_parse(string text, string name, string args) => Assert.Equal((name, args), ChatCommands.Parse(text)); + + [Theory] + [InlineData("hello")] + [InlineData("/")] + public void Non_commands_do_not_parse(string text) => Assert.Null(ChatCommands.Parse(text)); + + [Fact] + public void An_empty_allow_list_is_rejected() + { + var errors = new TelegramOptions { BotToken = "t" }.Validate(); + Assert.Contains(errors, e => e.Contains("allowed users", StringComparison.OrdinalIgnoreCase)); + } + + [Fact] + public void Environment_shortcuts_fill_the_token_and_users() + { + var env = new Dictionary { ["TELEGRAM_BOT_TOKEN"] = "123:abc", ["TELEGRAM_ALLOWED_USER_IDS"] = "1, 2;3" }; + var options = new TelegramOptions(); + options.ApplyEnvironment(k => env.GetValueOrDefault(k)); + + Assert.Equal("123:abc", options.BotToken); + Assert.Equal([1L, 2L, 3L], options.AllowedUserIds); + Assert.Empty(options.Validate()); + Assert.Equal("claude-code/claude-opus-5-5", options.Model); + Assert.True(options.IsAllowed(2)); + Assert.False(options.IsAllowed(4)); + Assert.False(options.IsAllowed(null)); + } +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/GlobalUsings.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/GlobalUsings.cs new file mode 100644 index 0000000..c802f44 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/GlobalUsings.cs @@ -0,0 +1 @@ +global using Xunit; diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/PollingServiceTests.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/PollingServiceTests.cs new file mode 100644 index 0000000..51659ca --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/PollingServiceTests.cs @@ -0,0 +1,139 @@ +using System.Diagnostics; +using Microsoft.Extensions.Logging.Abstractions; +using Telegram.Bot.Types; +using Volcanic.AgenticCore.Telegram.Polling; +using Volcanic.AgenticCore.Telegram.Tests.Support; + +namespace Volcanic.AgenticCore.Telegram.Tests; + +public sealed class PollingServiceTests +{ + private sealed class RecordingRouter(Func? onRoute = null) : IUpdateRouter + { + private readonly List _ids = []; + public IReadOnlyList Ids { get { lock (_ids) return _ids.ToList(); } } + + public async Task RouteAsync(Update update, CancellationToken cancellationToken) + { + lock (_ids) _ids.Add(update.Id); + if (onRoute is not null) await onRoute(update); + } + } + + private static (PollingService Service, ReconnectPolicy Policy, FakeLifetime Lifetime) Create( + FakeBotClient bot, IUpdateRouter router, int initialMs = 5, int maxMs = 20) + { + var policy = new ReconnectPolicy(TimeSpan.FromMilliseconds(initialMs), TimeSpan.FromMilliseconds(maxMs), jitter: 0); + var lifetime = new FakeLifetime(); + return (new PollingService(bot, router, policy, lifetime, NullLogger.Instance), policy, lifetime); + } + + [Fact] + public async Task Recovers_from_connection_loss_and_delivers_every_update_once_in_order() + { + var bot = new FakeBotClient() + .EnqueueFailure(FakeBotClient.NetworkDown()) + .EnqueueFailure(FakeBotClient.NetworkDown()) + .EnqueueUpdates(Updates.WithId(1), Updates.WithId(2)) + .EnqueueFailure(FakeBotClient.NetworkDown()) + .EnqueueFailure(FakeBotClient.NetworkDown()) + .EnqueueFailure(FakeBotClient.NetworkDown()) + .EnqueueUpdates(Updates.WithId(3)); + var router = new RecordingRouter(); + var (service, policy, lifetime) = Create(bot, router); + + await service.StartAsync(CancellationToken.None); + await Wait.UntilAsync(() => router.Ids.Count == 3, "three updates"); + await Wait.UntilAsync(() => bot.PollOffsets.Count == 8, "the poll after the last update"); + await service.StopAsync(CancellationToken.None); + + Assert.Equal([1, 2, 3], router.Ids); + // After a failure the same offset is asked again; after updates the offset moves past them. + Assert.Equal([0, 0, 0, 3, 3, 3, 3, 4], bot.PollOffsets); + Assert.Equal(0, policy.Failures); + Assert.False(lifetime.StopRequested); + } + + [Fact] + public async Task Backs_off_between_failed_polls_instead_of_spinning() + { + var bot = new FakeBotClient(); + for (var i = 0; i < 100; i++) bot.EnqueueFailure(FakeBotClient.NetworkDown()); + var (service, policy, _) = Create(bot, new RecordingRouter(), initialMs: 40, maxMs: 80); + + var clock = Stopwatch.StartNew(); + await service.StartAsync(CancellationToken.None); + await Wait.UntilAsync(() => bot.PollOffsets.Count >= 4, "four polls"); + var elapsed = clock.Elapsed; + await service.StopAsync(CancellationToken.None); + + // Delays 40 + 80 + 80 ms between the four polls (no jitter). + Assert.True(elapsed >= TimeSpan.FromMilliseconds(190), $"four polls took only {elapsed.TotalMilliseconds} ms"); + Assert.True(policy.Failures >= 3); + } + + [Fact] + public async Task Failures_after_a_success_start_again_from_the_initial_delay() + { + var policy = new ReconnectPolicy(TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(8), jitter: 0); + Assert.Equal([1, 2, 4, 8, 8], Enumerable.Range(0, 5).Select(_ => policy.NextDelay().TotalSeconds)); + Assert.Equal(5, policy.Reset()); + Assert.Equal(1, policy.NextDelay().TotalSeconds); + } + + [Fact] + public void Jitter_stays_within_bounds() + { + var policy = new ReconnectPolicy(TimeSpan.FromSeconds(10), TimeSpan.FromSeconds(10), jitter: 0.2); + for (var i = 0; i < 200; i++) + { + var delay = policy.NextDelay().TotalSeconds; + Assert.InRange(delay, 8, 12); + } + } + + [Fact] + public async Task A_rejected_token_stops_the_application() + { + var bot = new FakeBotClient().EnqueueFailure(FakeBotClient.Unauthorized()); + var (service, _, lifetime) = Create(bot, new RecordingRouter()); + + await service.StartAsync(CancellationToken.None); + await Wait.UntilAsync(() => lifetime.StopRequested, "StopApplication"); + await service.StopAsync(CancellationToken.None); + + Assert.Single(bot.PollOffsets); + Assert.NotNull(service.FatalError); + } + + [Fact] + public async Task A_failing_handler_does_not_stop_polling() + { + var bot = new FakeBotClient() + .EnqueueUpdates(Updates.WithId(10)) + .EnqueueUpdates(Updates.WithId(11)); + var router = new RecordingRouter(u => u.Id == 10 ? throw new InvalidOperationException("boom") : Task.CompletedTask); + var (service, _, _) = Create(bot, router); + + await service.StartAsync(CancellationToken.None); + await Wait.UntilAsync(() => router.Ids.Count == 2, "both updates"); + await service.StopAsync(CancellationToken.None); + + Assert.Equal([10, 11], router.Ids); + Assert.Equal(12, service.Offset); + } + + [Fact] + public async Task Stops_promptly_while_waiting_out_a_long_backoff() + { + var bot = new FakeBotClient().EnqueueFailure(FakeBotClient.NetworkDown()).EnqueueFailure(FakeBotClient.NetworkDown()); + var (service, _, _) = Create(bot, new RecordingRouter(), initialMs: 60_000, maxMs: 60_000); + + await service.StartAsync(CancellationToken.None); + await Wait.UntilAsync(() => bot.PollOffsets.Count == 1, "the first poll"); + var clock = Stopwatch.StartNew(); + await service.StopAsync(CancellationToken.None); + + Assert.True(clock.Elapsed < TimeSpan.FromSeconds(5)); + } +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs new file mode 100644 index 0000000..3b37246 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/ChatHost.cs @@ -0,0 +1,75 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using Volcanic.AgenticCore.Features.Agent; +using Volcanic.AgenticCore.Features.Extensions; +using Volcanic.AgenticCore.Features.Sessions; +using Volcanic.AgenticCore.Hosting; +using Volcanic.AgenticCore.Telegram.Chats; +using Volcanic.AgenticCore.Telegram.Rendering; +using Volcanic.AgenticCore.Testing; + +namespace Volcanic.AgenticCore.Telegram.Tests.Support; + +/// The bot's chat side wired to a and the . +public sealed class ChatHost : IAsyncDisposable +{ + public FakeBotClient Bot { get; } = new(); + public FauxProvider Faux { get; } = new(); + public TelegramOptions Options { get; } + public ChatRegistry Chats { get; } + public UpdateRouter Router { get; } + private readonly ServiceProvider _services; + private readonly string _root; + + public ChatHost() + { + _root = Path.Combine(Path.GetTempPath(), "volcanic-telegram", Guid.NewGuid().ToString("N")); + Options = new TelegramOptions + { + BotToken = "test", + AllowedUserIds = [Updates.AllowedUser], + WorkspaceRoot = Path.Combine(_root, "workspaces"), + Model = FauxProvider.ModelKey, + EditInterval = TimeSpan.FromMilliseconds(20), + TypingInterval = TimeSpan.FromMilliseconds(50), + }; + var store = new InMemorySessionStore(); + var services = new ServiceCollection(); + services.AddAgenticCore(o => + { + o.AgentDir = Path.Combine(_root, "agent"); + o.DiscoverExtensionAssemblies = false; + o.DefaultModel = FauxProvider.ModelKey; + o.Extensions.Add(ExtensionRegistration.FromInstance(Faux.Extension)); + o.SessionStoreFactory = _ => store; + }); + _services = services.BuildServiceProvider(); + ILoggerFactory loggers = NullLoggerFactory.Instance; + var sender = new TelegramSender(Bot, NullLogger.Instance); + Chats = new ChatRegistry(_services.GetRequiredService(), sender, Options, loggers); + Router = new UpdateRouter(Options, Chats, new ChatCommands(sender), sender, NullLogger.Instance); + } + + public Task SendAsync(string text, long userId = Updates.AllowedUser) => Router.RouteAsync(Updates.Text(text, userId), CancellationToken.None); + + public async Task ChatAsync() => await Chats.GetAsync(Updates.AllowedUser); + + /// Wait until the chat's run has settled and everything it produced is rendered. + public async Task SettleAsync() + { + var chat = await ChatAsync(); + await chat.LastPrompt.WaitAsync(TimeSpan.FromSeconds(10)); + await Wait.UntilAsync(() => !chat.Session.IsStreaming, "the run to settle"); + await chat.Renderer.DrainAsync().WaitAsync(TimeSpan.FromSeconds(5)); + } + + public IEnumerable Texts => Bot.CurrentTexts.OrderBy(kv => kv.Key).Select(kv => kv.Value); + + public async ValueTask DisposeAsync() + { + await Chats.DisposeAsync(); + await _services.DisposeAsync(); + try { Directory.Delete(_root, true); } catch (IOException) { } + } +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/FakeBotClient.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/FakeBotClient.cs new file mode 100644 index 0000000..f5ba113 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/FakeBotClient.cs @@ -0,0 +1,107 @@ +using System.Collections.Concurrent; +using System.Net; +using Telegram.Bot; +using Telegram.Bot.Args; +using Telegram.Bot.Exceptions; +using Telegram.Bot.Requests; +using Telegram.Bot.Requests.Abstractions; +using Telegram.Bot.Types; + +namespace Volcanic.AgenticCore.Telegram.Tests.Support; + +/// +/// In-memory : records every request, answers sends with fake messages and plays a +/// script of getUpdates results. When the script is exhausted a poll blocks until cancelled, like an idle long poll. +/// +public sealed class FakeBotClient : ITelegramBotClient +{ + private readonly ConcurrentQueue> _polls = new(); + private readonly List _requests = []; + private int _nextMessageId = 100; + + public bool LocalBotServer => false; + public long BotId => 42; + public TimeSpan Timeout { get; set; } = TimeSpan.FromSeconds(50); + public IExceptionParser ExceptionsParser { get; set; } = new DefaultExceptionParser(); + + public event AsyncEventHandler? OnMakingApiRequest; + public event AsyncEventHandler? OnApiResponseReceived; + + /// Called for every non-poll request before it is answered; throw to simulate a failure. + public Action? OnSend { get; set; } + + public IReadOnlyList Requests { get { lock (_requests) return _requests.ToList(); } } + /// The offset of every getUpdates call (the library reuses one request object, so it is captured per call). + public IReadOnlyList PollOffsets { get { lock (_pollOffsets) return _pollOffsets.ToList(); } } + private readonly List _pollOffsets = []; + public IReadOnlyList Sent => Requests.OfType().ToList(); + public IReadOnlyList Edits => Requests.OfType().ToList(); + + /// The text each sent message shows now (after edits), by message id. + public IReadOnlyDictionary CurrentTexts => new Dictionary(_texts); + + private readonly ConcurrentDictionary _texts = new(); + + public FakeBotClient EnqueueUpdates(params Update[] updates) + { + _polls.Enqueue(() => updates); + return this; + } + + public FakeBotClient EnqueueFailure(Exception exception) + { + _polls.Enqueue(() => throw exception); + return this; + } + + public static RequestException NetworkDown() => + new("Bot API Service Failure: HttpRequestException: Name or service not known", new HttpRequestException("Name or service not known")); + + public static ApiRequestException Unauthorized() => new("Unauthorized", (int)HttpStatusCode.Unauthorized); + + public async Task SendRequest(IRequest request, CancellationToken cancellationToken = default) + { + lock (_requests) _requests.Add(request); + if (OnMakingApiRequest is not null) await OnMakingApiRequest(this, new ApiRequestEventArgs(request), cancellationToken); + + if (request is GetUpdatesRequest getUpdates) + { + lock (_pollOffsets) _pollOffsets.Add(getUpdates.Offset); + if (!_polls.TryDequeue(out var poll)) + { + await Task.Delay(System.Threading.Timeout.Infinite, cancellationToken); + throw new OperationCanceledException(cancellationToken); + } + var updates = poll(); // may throw: a failed poll + if (OnApiResponseReceived is not null) + { + using var response = new HttpResponseMessage(HttpStatusCode.OK); + await OnApiResponseReceived(this, new ApiResponseEventArgs(response, new ApiRequestEventArgs(request)), cancellationToken); + } + return (TResponse)(object)updates; + } + + OnSend?.Invoke(request); + object result; + switch (request) + { + case SendMessageRequest send: + var id = Interlocked.Increment(ref _nextMessageId); + _texts[id] = send.Text; + result = new Message { Id = id, Chat = new Chat { Id = send.ChatId.Identifier ?? 0 }, Text = send.Text }; + break; + case EditMessageTextRequest edit: + _texts[edit.MessageId] = edit.Text!; + result = new Message { Id = edit.MessageId, Text = edit.Text }; + break; + default: + result = request is GetMeRequest ? new User { Id = BotId, IsBot = true, FirstName = "Fake", Username = "fake_bot" } : true; + break; + } + return typeof(TResponse) == typeof(bool) ? (TResponse)(object)true : (TResponse)result; + } + + public Task TestApi(CancellationToken cancellationToken = default) => Task.FromResult(true); + public Task DownloadFile(string filePath, Stream destination, CancellationToken cancellationToken = default) => Task.CompletedTask; + public Task DownloadFile(TGFile file, Stream destination, CancellationToken cancellationToken = default) => Task.CompletedTask; +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/Support/Updates.cs b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/Updates.cs new file mode 100644 index 0000000..da325a4 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/Support/Updates.cs @@ -0,0 +1,65 @@ +using Microsoft.Extensions.Hosting; +using Telegram.Bot.Types; +using Telegram.Bot.Types.Enums; + +namespace Volcanic.AgenticCore.Telegram.Tests.Support; + +public static class Updates +{ + public const long AllowedUser = 1001; + public const long Stranger = 666; + + private static int _nextUpdateId; + private static int _nextMessageId = 1; + + public static Update Text(string text, long userId = AllowedUser, long? chatId = null, ChatType chatType = ChatType.Private) => new() + { + Id = Interlocked.Increment(ref _nextUpdateId), + Message = new Message + { + Id = Interlocked.Increment(ref _nextMessageId), + From = new User { Id = userId, FirstName = "Test" }, + Chat = new Chat { Id = chatId ?? userId, Type = chatType }, + Text = text, + Date = DateTime.UtcNow, + }, + }; + + public static Update WithId(int id) => new() { Id = id, Message = Text("#" + id).Message }; + + public static Update Callback(string data, long chatId, long userId = AllowedUser) => new() + { + Id = Interlocked.Increment(ref _nextUpdateId), + CallbackQuery = new CallbackQuery + { + Id = Guid.NewGuid().ToString("N"), + From = new User { Id = userId, FirstName = "Test" }, + Message = new Message { Id = 1, Chat = new Chat { Id = chatId, Type = ChatType.Private }, Date = DateTime.UtcNow }, + Data = data, + ChatInstance = "ci", + }, + }; +} + +public static class Wait +{ + public static async Task UntilAsync(Func condition, string what, int timeoutMs = 5000) + { + var deadline = DateTime.UtcNow.AddMilliseconds(timeoutMs); + while (!condition()) + { + if (DateTime.UtcNow > deadline) throw new TimeoutException($"Timed out waiting for {what}"); + await Task.Delay(10); + } + } +} + +public sealed class FakeLifetime : IHostApplicationLifetime +{ + private readonly CancellationTokenSource _stopping = new(); + public CancellationToken ApplicationStarted => CancellationToken.None; + public CancellationToken ApplicationStopping => _stopping.Token; + public CancellationToken ApplicationStopped => CancellationToken.None; + public bool StopRequested => _stopping.IsCancellationRequested; + public void StopApplication() => _stopping.Cancel(); +} diff --git a/tests/Volcanic.AgenticCore.Telegram.Tests/Volcanic.AgenticCore.Telegram.Tests.csproj b/tests/Volcanic.AgenticCore.Telegram.Tests/Volcanic.AgenticCore.Telegram.Tests.csproj new file mode 100644 index 0000000..a373da4 --- /dev/null +++ b/tests/Volcanic.AgenticCore.Telegram.Tests/Volcanic.AgenticCore.Telegram.Tests.csproj @@ -0,0 +1,15 @@ + + + false + true + false + + + + + + + + + +