Compare commits
3 Commits
692992bb35
...
308e1fbf8e
| Author | SHA1 | Date | |
|---|---|---|---|
|
308e1fbf8e | ||
|
|
2321906719 | ||
|
|
8a97a9890f |
@@ -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/`.
|
||||
|
||||
@@ -5,6 +5,9 @@
|
||||
<Project Path="extensions/Volcanic.AgenticCore.Extensions.ClaudeCode/Volcanic.AgenticCore.Extensions.ClaudeCode.csproj" />
|
||||
<Project Path="extensions/Volcanic.AgenticCore.Extensions.OpenAI/Volcanic.AgenticCore.Extensions.OpenAI.csproj" />
|
||||
</Folder>
|
||||
<Folder Name="/hosts/">
|
||||
<Project Path="hosts/Volcanic.AgenticCore.Telegram/Volcanic.AgenticCore.Telegram.csproj" />
|
||||
</Folder>
|
||||
<Folder Name="/samples/">
|
||||
<Project Path="samples/Volcanic.AgenticCore.Sample.Headless/Volcanic.AgenticCore.Sample.Headless.csproj" />
|
||||
</Folder>
|
||||
@@ -14,6 +17,7 @@
|
||||
</Folder>
|
||||
<Folder Name="/tests/">
|
||||
<Project Path="tests/Volcanic.AgenticCore.Extensions.Tests/Volcanic.AgenticCore.Extensions.Tests.csproj" />
|
||||
<Project Path="tests/Volcanic.AgenticCore.Telegram.Tests/Volcanic.AgenticCore.Telegram.Tests.csproj" />
|
||||
<Project Path="tests/Volcanic.AgenticCore.Tests/Volcanic.AgenticCore.Tests.csproj" />
|
||||
</Folder>
|
||||
</Solution>
|
||||
|
||||
@@ -20,7 +20,7 @@ Register it and pick a model:
|
||||
|
||||
```csharp
|
||||
o.AddExtension<ClaudeCodeProviderExtension>();
|
||||
o.DefaultModel = "claude-code/sonnet"; // or opus, haiku, fable
|
||||
o.DefaultModel = "claude-code/sonnet"; // or opus, haiku, fable, or the pinned claude-opus-5-5
|
||||
```
|
||||
|
||||
Run `/claude-code-doctor` in a session to check the installation, auth, required CLI flags and the bridge
|
||||
|
||||
91
docs/superpowers/specs/2026-09-25-telegram-host-design.md
Normal file
91
docs/superpowers/specs/2026-09-25-telegram-host-design.md
Normal file
@@ -0,0 +1,91 @@
|
||||
# Telegram host — design
|
||||
|
||||
Date: 2026-09-25. Status: approved in chat.
|
||||
|
||||
## Goal
|
||||
|
||||
Chat with the agent through a Telegram bot. A new host project in this repo, run with `dotnet run`, long polling
|
||||
(no public URL). Implements the contract described in `docs/ui-integration.md`.
|
||||
|
||||
## Decisions
|
||||
|
||||
| Topic | Decision |
|
||||
|---|---|
|
||||
| Location | `hosts/Volcanic.AgenticCore.Telegram` (console app, Generic Host) + `tests/Volcanic.AgenticCore.Telegram.Tests` |
|
||||
| Library | `Telegram.Bot` 22.x |
|
||||
| Polling | The library's built-in loop (`ReceiveAsync` with an `IUpdateHandler`), hardened for connection loss (below) |
|
||||
| Provider | `ClaudeCodeProviderExtension` only |
|
||||
| Model | `claude-code/claude-opus-5-5`, a pinned entry added to `ClaudeCodeModels` (the CLI accepts full ids in `--model`) |
|
||||
| Tools | All built-in tools; each chat works in its own directory `<WorkspaceRoot>/<chatId>` |
|
||||
| Access | Mandatory user-id allow-list; empty list = refuse to start; other users are ignored silently |
|
||||
| Chats | Private chats only |
|
||||
|
||||
Out of scope: webhooks, Docker/Coolify, voice and documents, group chats, several bot tokens, a custom polling loop.
|
||||
|
||||
## Configuration
|
||||
|
||||
`appsettings.json` section `Telegram`, overridable by environment variables (`Telegram__...`):
|
||||
|
||||
- `BotToken` (also read from `TELEGRAM_BOT_TOKEN`), required.
|
||||
- `AllowedUserIds`, required, non-empty (`Telegram__AllowedUserIds__0=123`, or `TELEGRAM_ALLOWED_USER_IDS=1,2`).
|
||||
- `WorkspaceRoot`, default `~/.volcanic/telegram-workspaces`.
|
||||
- `Model`, default `claude-code/claude-opus-5-5`.
|
||||
|
||||
At start the host logs whether the `claude` CLI was found; a missing CLI is a warning (the doctor command explains).
|
||||
|
||||
## Polling and connection recovery
|
||||
|
||||
Findings from the Telegram.Bot 22.x source (`TelegramBotClientExtensions.Polling.cs`):
|
||||
|
||||
1. After a `getUpdates` failure the loop calls `HandleErrorAsync(PollingError)` and then polls again **immediately**:
|
||||
without a delay it spins while the network is down.
|
||||
2. If `HandleErrorAsync` throws anything but `OperationCanceledException`, the exception leaves `ReceiveAsync`
|
||||
and polling is over.
|
||||
3. The long-poll timeout equals `HttpClient.Timeout`, so an idle poll occasionally times out client side
|
||||
("Bot API Request timed out"). Harmless.
|
||||
4. The offset advances before an update's handler runs and handlers run one at a time.
|
||||
5. `429` with `retry_after <= 60` is retried inside the client.
|
||||
|
||||
Design (`PollingService : BackgroundService`, `BotUpdateHandler : IUpdateHandler`, `ReconnectPolicy`):
|
||||
|
||||
- `HandleErrorAsync(PollingError)` never throws. It asks `ReconnectPolicy` for the next delay and awaits it, which is
|
||||
what holds the built-in loop back: 1 s, 2 s, 4 s … capped at 60 s, ±20 % jitter.
|
||||
- `401 Unauthorized` (token revoked or wrong) is fatal: log and stop the application.
|
||||
- `409 Conflict` (another instance polling) is logged as such and backed off like any other error.
|
||||
- The first failure is logged at debug level (covers the idle timeout race), later ones as warnings with
|
||||
"Telegram unreachable (attempt n), retrying in x s".
|
||||
- `OnApiResponseReceived` for a successful `getUpdates` resets the policy and, if it was failing, logs
|
||||
"Telegram connection restored after n failed attempts".
|
||||
- `PollingService` wraps `ReceiveAsync` in an outer loop: if it ever returns or throws while the host is still
|
||||
running, it is restarted after the policy delay with the last offset kept, so no update is lost or repeated.
|
||||
- `HandleUpdateAsync` returns fast: prompts run in background tasks, so the loop keeps receiving
|
||||
(`/stop` arrives while a run is active).
|
||||
|
||||
## Components
|
||||
|
||||
- `TelegramOptions` + validation.
|
||||
- `BotUpdateHandler`: allow-list, then route: callback query → `TelegramUi`; reply to a pending input → `TelegramUi`;
|
||||
bot command → `ChatCommands`; text/photo → prompt in the background. Errors are reported to the chat.
|
||||
- `ChatSessions`: one `AgentSession` per chat, lazy; resumes the most recent session in scope `telegram-<chatId>`.
|
||||
- `ChatRenderer`: subscribes once per session, events into a channel, one render task per chat.
|
||||
Placeholder message on assistant `MessageStart`; text deltas edit it at most every 1.5 s; final edit on `MessageEnd`;
|
||||
text over 4096 chars continues in new messages; typing action every 4 s while running; tool lines
|
||||
`▶ bash: <args>` → `✓ bash` / `✗ bash: <excerpt>`; retry, compaction, errors and notifications as short lines.
|
||||
Telegram send/edit failures are logged and never stop the render task.
|
||||
- `TelegramFormatter`: Markdown → Telegram HTML (code blocks, inline code, bold, italics, links; everything else
|
||||
escaped); a send that Telegram rejects as bad markup is retried as plain text. Splitting at 4096.
|
||||
- `TelegramUi : IUserInterface`: inline keyboards for select/confirm, ForceReply for input, timeouts.
|
||||
- `ChatCommands`: `/start`, `/help`, `/new`, `/stop`, `/status`, `/sessions`, `/resume <id>`, `/compact`,
|
||||
`/thinking <level>`. Other `/commands` go to `PromptAsync` (extension commands).
|
||||
|
||||
## Testing
|
||||
|
||||
xunit, a fake `ITelegramBotClient` that records requests and scripts `getUpdates` responses, `FauxProvider` as model.
|
||||
|
||||
- Polling survives thrown network errors and keeps delivering every update exactly once in order; delays grow and reset.
|
||||
- A 401 stops the application.
|
||||
- A throwing handler does not stop polling.
|
||||
- Unknown users are ignored.
|
||||
- A prompt streams into an edited message; long text is split; `/stop` aborts a running prompt.
|
||||
- `TelegramUi` select round-trip through a callback query.
|
||||
- Formatter escaping and splitting.
|
||||
75
docs/telegram-bot.md
Normal file
75
docs/telegram-bot.md
Normal file
@@ -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 `<id>` works in `<root>/<id>` |
|
||||
| `Telegram:Model` | `claude-code/claude-opus-5-5` | Any `provider/model` the Claude Code provider serves |
|
||||
| `Telegram:EditInterval` | `00:00:01.5` | Minimum time between edits of a streaming reply |
|
||||
| `Telegram:ReconnectInitialDelay` / `ReconnectMaxDelay` | `1 s` / `60 s` | Backoff after lost connection |
|
||||
|
||||
Sessions are stored as usual under `~/.volcanic/agent/sessions/telegram-<chatId>/`; after a restart each chat resumes
|
||||
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 <id>` | 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.
|
||||
@@ -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
|
||||
|
||||
@@ -30,8 +30,10 @@ public static class ClaudeCodeModels
|
||||
// Haiku has no effort control; Claude Code still owns its thinking default.
|
||||
public static readonly Model Haiku = Create("haiku", "Claude Code Haiku", 200_000, 32_000, reasoning: false);
|
||||
public static readonly Model Fable = Create("fable", "Claude Code Fable", 1_000_000, 64_000, reasoning: true);
|
||||
// A full model id pins the version; the aliases above follow whatever Claude Code maps them to.
|
||||
public static readonly Model Opus55 = Create("claude-opus-5-5", "Claude Code Opus 5.5", 1_000_000, 128_000, reasoning: true);
|
||||
|
||||
public static IReadOnlyList<Model> All { get; } = [Sonnet, Opus, Haiku, Fable];
|
||||
public static IReadOnlyList<Model> All { get; } = [Sonnet, Opus, Haiku, Fable, Opus55];
|
||||
|
||||
/// <summary>The <c>--effort</c> value for a request, or null to omit the flag.</summary>
|
||||
public static string? EffortFor(Model model, ThinkingLevel? level)
|
||||
|
||||
45
hosts/Volcanic.AgenticCore.Telegram/BotSetupService.cs
Normal file
45
hosts/Volcanic.AgenticCore.Telegram/BotSetupService.cs
Normal file
@@ -0,0 +1,45 @@
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Telegram.Bot;
|
||||
using Volcanic.AgenticCore.Telegram.Chats;
|
||||
|
||||
namespace Volcanic.AgenticCore.Telegram;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public sealed class BotSetupService(ITelegramBotClient bot, TelegramOptions options, ILogger<BotSetupService> 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));
|
||||
}
|
||||
}
|
||||
}
|
||||
171
hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs
Normal file
171
hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs
Normal file
@@ -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;
|
||||
|
||||
/// <summary>The bot's own slash commands. Anything else starting with <c>/</c> goes to the session (extension commands).</summary>
|
||||
public sealed class ChatCommands(TelegramSender sender)
|
||||
{
|
||||
public static readonly IReadOnlyList<BotCommand> 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 <id>" },
|
||||
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" },
|
||||
];
|
||||
|
||||
/// <summary>Split <c>/name@bot args</c>. Null when the text is not a command.</summary>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>Run a bot command. False when <paramref name="text"/> is not one of ours.</summary>
|
||||
public async Task<bool> 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 <id> – 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<string> 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<string> ResumeAsync(ChatContext chat, string id, CancellationToken cancellationToken)
|
||||
{
|
||||
if (id.Length == 0) return "Usage: /resume <id> (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<string> 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<ThinkingLevel>(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()}**.";
|
||||
}
|
||||
}
|
||||
135
hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs
Normal file
135
hosts/Volcanic.AgenticCore.Telegram/Chats/ChatRegistry.cs
Normal file
@@ -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;
|
||||
|
||||
/// <summary>Everything that belongs to one Telegram chat: its agent session, renderer and dialog surface.</summary>
|
||||
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;
|
||||
|
||||
/// <summary>The most recent background prompt; completes when that run has settled.</summary>
|
||||
public Task LastPrompt { get; private set; } = Task.CompletedTask;
|
||||
|
||||
/// <summary>
|
||||
/// Send user input without waiting for the run (it can take minutes, and the polling loop must keep receiving so
|
||||
/// <c>/stop</c> gets through). While a run is active the text steers it.
|
||||
/// </summary>
|
||||
public Task PromptInBackground(string text, IReadOnlyList<ImageContent>? 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}");
|
||||
}
|
||||
});
|
||||
|
||||
/// <summary>Show an error in the chat, in order with the rendered events.</summary>
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// One <see cref="ChatContext"/> per chat, created on first use. Each chat works in <c><WorkspaceRoot>/<chatId></c>,
|
||||
/// stores its sessions under scope <c>telegram-<chatId></c> and resumes the most recent one after a restart.
|
||||
/// </summary>
|
||||
public sealed class ChatRegistry(IAgentSessionFactory factory, TelegramSender sender, TelegramOptions options, ILoggerFactory loggerFactory)
|
||||
: IAsyncDisposable
|
||||
{
|
||||
private readonly ConcurrentDictionary<long, Lazy<Task<ChatContext>>> _chats = new();
|
||||
private readonly ILogger _logger = loggerFactory.CreateLogger<ChatRegistry>();
|
||||
|
||||
public static string ScopeFor(long chatId) => $"telegram-{chatId}";
|
||||
|
||||
public string WorkspaceFor(long chatId) => Path.Combine(options.WorkspaceRoot, chatId.ToString(System.Globalization.CultureInfo.InvariantCulture));
|
||||
|
||||
/// <summary>The chat's context if it was already created, without creating it.</summary>
|
||||
public ChatContext? Find(long chatId) =>
|
||||
_chats.TryGetValue(chatId, out var lazy) && lazy.IsValueCreated && lazy.Value.IsCompletedSuccessfully ? lazy.Value.Result : null;
|
||||
|
||||
public async Task<ChatContext> GetAsync(long chatId, CancellationToken cancellationToken = default)
|
||||
{
|
||||
var lazy = _chats.GetOrAdd(chatId, id => new Lazy<Task<ChatContext>>(() => CreateAsync(id)));
|
||||
try
|
||||
{
|
||||
return await lazy.Value.WaitAsync(cancellationToken).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception) when (lazy.Value.IsFaulted)
|
||||
{
|
||||
_chats.TryRemove(new KeyValuePair<long, Lazy<Task<ChatContext>>>(chatId, lazy)); // let the next message try again
|
||||
throw;
|
||||
}
|
||||
}
|
||||
|
||||
private async Task<ChatContext> 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<ChatContext>();
|
||||
var renderer = new ChatRenderer(chatId, sender, options, loggerFactory.CreateLogger<ChatRenderer>());
|
||||
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();
|
||||
}
|
||||
}
|
||||
92
hosts/Volcanic.AgenticCore.Telegram/Chats/TelegramUi.cs
Normal file
92
hosts/Volcanic.AgenticCore.Telegram/Chats/TelegramUi.cs
Normal file
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// <see cref="IUserInterface"/> for one chat: choices become inline keyboards (callback data <c>ui:<id>:<index></c>),
|
||||
/// free-text questions a ForceReply prompt answered by the chat's next plain message.
|
||||
/// </summary>
|
||||
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<string, TaskCompletionSource<string?>> _choices = new();
|
||||
private TaskCompletionSource<string?>? _input;
|
||||
|
||||
/// <summary>True while an <see cref="InputAsync"/> question waits for its answer.</summary>
|
||||
public bool IsWaitingForInput => Volatile.Read(ref _input) is { Task.IsCompleted: false };
|
||||
|
||||
public async Task<string?> SelectAsync(string title, IReadOnlyList<string> options, UiRequestOptions? requestOptions, CancellationToken cancellationToken)
|
||||
{
|
||||
if (options.Count == 0) return null;
|
||||
var id = Guid.NewGuid().ToString("N")[..8];
|
||||
var tcs = new TaskCompletionSource<string?>(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<bool> ConfirmAsync(string title, string message, UiRequestOptions? requestOptions, CancellationToken cancellationToken) =>
|
||||
await SelectAsync($"{title}\n{message}", ["Yes", "No"], requestOptions, cancellationToken).ConfigureAwait(false) == "Yes";
|
||||
|
||||
public async Task<string?> InputAsync(string title, string? placeholder, UiRequestOptions? requestOptions, CancellationToken cancellationToken)
|
||||
{
|
||||
var tcs = new TaskCompletionSource<string?>(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));
|
||||
|
||||
/// <summary>Status lines are not shown in Telegram; tool and retry progress is rendered from session events.</summary>
|
||||
public void SetStatus(string key, string? text) { }
|
||||
|
||||
/// <summary>Resolve a pending choice from callback data. False when it is not ours or no longer pending.</summary>
|
||||
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]);
|
||||
}
|
||||
|
||||
/// <summary>Answer a pending <see cref="InputAsync"/> question. False when none is pending.</summary>
|
||||
public bool TryResolveInput(string text) => Interlocked.Exchange(ref _input, null) is { } tcs && tcs.TrySetResult(text);
|
||||
|
||||
private static async Task<string?> WaitAsync(TaskCompletionSource<string?> 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);
|
||||
}
|
||||
}
|
||||
95
hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs
Normal file
95
hosts/Volcanic.AgenticCore.Telegram/Chats/UpdateRouter.cs
Normal file
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public sealed class UpdateRouter(
|
||||
TelegramOptions options,
|
||||
ChatRegistry chats,
|
||||
ChatCommands commands,
|
||||
TelegramSender sender,
|
||||
ILogger<UpdateRouter> 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<IReadOnlyList<ImageContent>?> 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
153
hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs
Normal file
153
hosts/Volcanic.AgenticCore.Telegram/Polling/PollingService.cs
Normal file
@@ -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;
|
||||
|
||||
/// <summary>Receives one update. Must return quickly: the polling loop waits for it before the next update.</summary>
|
||||
public interface IUpdateRouter
|
||||
{
|
||||
Task RouteAsync(Update update, CancellationToken cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Runs Telegram.Bot's built-in long-polling loop and makes it survive connection loss.
|
||||
/// <para>
|
||||
/// The built-in loop polls again right after a failed <c>getUpdates</c> (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 <c>getUpdates</c> response resets the backoff. <c>401</c> 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.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
public sealed class PollingService(
|
||||
ITelegramBotClient bot,
|
||||
IUpdateRouter router,
|
||||
ReconnectPolicy policy,
|
||||
IHostApplicationLifetime lifetime,
|
||||
ILogger<PollingService> logger) : BackgroundService, IUpdateHandler
|
||||
{
|
||||
internal static readonly UpdateType[] AllowedUpdates = [UpdateType.Message, UpdateType.CallbackQuery];
|
||||
|
||||
private int _offset;
|
||||
|
||||
/// <summary>The next update id to request; everything below it has been handed to the router.</summary>
|
||||
public int Offset => Volatile.Read(ref _offset);
|
||||
|
||||
/// <summary>Set when polling stopped the application because retrying cannot help (a rejected token).</summary>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <returns>False when the host is stopping.</returns>
|
||||
private static async Task<bool> DelayAsync(TimeSpan delay, CancellationToken cancellationToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
|
||||
return true;
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>A polling error retrying cannot fix.</summary>
|
||||
public sealed class FatalPollingException(string message, Exception inner) : Exception(message, inner);
|
||||
@@ -0,0 +1,27 @@
|
||||
namespace Volcanic.AgenticCore.Telegram.Polling;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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;
|
||||
|
||||
/// <summary>Consecutive failures since the last success.</summary>
|
||||
public int Failures => Volatile.Read(ref _failures);
|
||||
|
||||
/// <summary>Record a failure and return how long to wait before the next attempt.</summary>
|
||||
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));
|
||||
}
|
||||
|
||||
/// <summary>Record a success. Returns the number of failures it ends (0 when the connection was healthy).</summary>
|
||||
public int Reset() => Interlocked.Exchange(ref _failures, 0);
|
||||
}
|
||||
55
hosts/Volcanic.AgenticCore.Telegram/Program.cs
Normal file
55
hosts/Volcanic.AgenticCore.Telegram/Program.cs
Normal file
@@ -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<TelegramOptions>() ?? 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<ClaudeCodeProviderExtension>();
|
||||
});
|
||||
builder.Services.AddSingleton<ITelegramBotClient>(_ => new TelegramBotClient(options.BotToken));
|
||||
builder.Services.AddSingleton(new ReconnectPolicy(options.ReconnectInitialDelay, options.ReconnectMaxDelay));
|
||||
builder.Services.AddSingleton<TelegramSender>();
|
||||
builder.Services.AddSingleton<ChatRegistry>();
|
||||
builder.Services.AddSingleton<ChatCommands>();
|
||||
builder.Services.AddSingleton<IUpdateRouter, UpdateRouter>();
|
||||
builder.Services.AddSingleton<PollingService>();
|
||||
builder.Services.AddHostedService(sp => sp.GetRequiredService<PollingService>());
|
||||
builder.Services.AddHostedService<BotSetupService>();
|
||||
|
||||
using var host = builder.Build();
|
||||
|
||||
var logger = host.Services.GetRequiredService<ILoggerFactory>().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<PollingService>();
|
||||
await host.RunAsync(); // disposes the host
|
||||
return polling.FatalError is null ? 0 : 1;
|
||||
309
hosts/Volcanic.AgenticCore.Telegram/Rendering/ChatRenderer.cs
Normal file
309
hosts/Volcanic.AgenticCore.Telegram/Rendering/ChatRenderer.cs
Normal file
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Renders one chat's session events. Session handlers run on the agent loop, so events are queued into a channel and
|
||||
/// rendered by a single task: streamed text edits one message (split into several past the size limit) at most every
|
||||
/// <see cref="TelegramOptions.EditInterval"/>, tool calls become status lines, and "typing…" repeats while the agent runs.
|
||||
/// </summary>
|
||||
public sealed class ChatRenderer : IAsyncDisposable
|
||||
{
|
||||
private const int ToolExcerptLength = 300;
|
||||
|
||||
private readonly long _chatId;
|
||||
private readonly TelegramSender _sender;
|
||||
private readonly TelegramOptions _options;
|
||||
private readonly ILogger _logger;
|
||||
private readonly TimeProvider _time;
|
||||
private readonly Channel<AgentSessionEvent> _events = Channel.CreateUnbounded<AgentSessionEvent>(new UnboundedChannelOptions { SingleReader = true });
|
||||
private readonly CancellationTokenSource _stop = new();
|
||||
private readonly Task _loop;
|
||||
|
||||
// Streaming reply state; touched only by the render loop.
|
||||
private readonly List<(int MessageId, string Sent)> _replyMessages = [];
|
||||
private readonly Dictionary<string, (int MessageId, string Args)> _toolMessages = [];
|
||||
private string _replyText = "";
|
||||
private bool _replyDirty;
|
||||
private DateTimeOffset _lastFlush = DateTimeOffset.MinValue;
|
||||
private CancellationTokenSource? _typing;
|
||||
private int _pending;
|
||||
|
||||
public ChatRenderer(long chatId, TelegramSender sender, TelegramOptions options, ILogger logger, TimeProvider? time = null)
|
||||
{
|
||||
_chatId = chatId;
|
||||
_sender = sender;
|
||||
_options = options;
|
||||
_logger = logger;
|
||||
_time = time ?? TimeProvider.System;
|
||||
_loop = Task.Run(RunAsync);
|
||||
}
|
||||
|
||||
/// <summary>Queue an event; safe to call from any thread (this is the session subscription).</summary>
|
||||
public void Post(AgentSessionEvent e)
|
||||
{
|
||||
Interlocked.Increment(ref _pending);
|
||||
if (!_events.Writer.TryWrite(e)) Interlocked.Decrement(ref _pending);
|
||||
}
|
||||
|
||||
/// <summary>Completes once every queued event is rendered. For tests.</summary>
|
||||
internal async Task DrainAsync()
|
||||
{
|
||||
while (Volatile.Read(ref _pending) > 0 || _replyDirty) await Task.Delay(10).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
private async Task RunAsync()
|
||||
{
|
||||
var reader = _events.Reader;
|
||||
var ct = _stop.Token;
|
||||
try
|
||||
{
|
||||
while (true)
|
||||
{
|
||||
if (_replyDirty)
|
||||
{
|
||||
var wait = _lastFlush + _options.EditInterval - _time.GetUtcNow();
|
||||
if (wait <= TimeSpan.Zero)
|
||||
{
|
||||
await FlushReplyAsync(ct).ConfigureAwait(false);
|
||||
continue;
|
||||
}
|
||||
using var timeout = CancellationTokenSource.CreateLinkedTokenSource(ct);
|
||||
timeout.CancelAfter(wait);
|
||||
try
|
||||
{
|
||||
if (!await reader.WaitToReadAsync(timeout.Token).ConfigureAwait(false)) break;
|
||||
}
|
||||
catch (OperationCanceledException) when (!ct.IsCancellationRequested)
|
||||
{
|
||||
continue; // edit interval elapsed: flush on the next iteration
|
||||
}
|
||||
}
|
||||
else if (!await reader.WaitToReadAsync(ct).ConfigureAwait(false))
|
||||
{
|
||||
break;
|
||||
}
|
||||
|
||||
while (reader.TryRead(out var e))
|
||||
{
|
||||
try
|
||||
{
|
||||
await RenderAsync(e, ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
_logger.LogWarning(ex, "Rendering {Event} for chat {ChatId} failed", e.GetType().Name, _chatId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
Interlocked.Decrement(ref _pending);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (_replyDirty) await FlushReplyAsync(ct).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
}
|
||||
finally
|
||||
{
|
||||
StopTyping();
|
||||
}
|
||||
}
|
||||
|
||||
private async Task RenderAsync(AgentSessionEvent e, CancellationToken ct)
|
||||
{
|
||||
switch (e)
|
||||
{
|
||||
case AgentStart:
|
||||
StartTyping();
|
||||
break;
|
||||
case AgentSettled:
|
||||
StopTyping();
|
||||
break;
|
||||
|
||||
case MessageStart { Message: AssistantMessage }:
|
||||
ResetReply();
|
||||
break;
|
||||
case MessageUpdate { AssistantMessageEvent: AssistantMessageEvent.TextDelta } update:
|
||||
_replyText = update.Message.Text;
|
||||
_replyDirty = true;
|
||||
break;
|
||||
case MessageEnd { Message: AssistantMessage message }:
|
||||
_replyText = message.Text;
|
||||
_replyDirty = true;
|
||||
await FlushReplyAsync(ct).ConfigureAwait(false);
|
||||
if (message.StopReason == StopReason.Error)
|
||||
await SendLineAsync($"❌ {message.ErrorMessage ?? "The model request failed."}", ct).ConfigureAwait(false);
|
||||
else if (message.StopReason == StopReason.Aborted)
|
||||
await SendLineAsync("⏹ Stopped.", ct).ConfigureAwait(false);
|
||||
ResetReply();
|
||||
break;
|
||||
|
||||
case ToolExecutionStart start:
|
||||
// Keep the reply above the tool line in the chat.
|
||||
if (_replyDirty) await FlushReplyAsync(ct).ConfigureAwait(false);
|
||||
var args = FormatArgs(start.Args);
|
||||
if (await SendHtmlLineAsync($"▶ <b>{TelegramFormatter.Escape(start.ToolName)}</b>{args}", ct).ConfigureAwait(false) is { } id)
|
||||
_toolMessages[start.ToolCallId] = (id, args);
|
||||
break;
|
||||
case ToolExecutionEnd end:
|
||||
var started = _toolMessages.Remove(end.ToolCallId, out var tool);
|
||||
var line = end.IsError
|
||||
? $"✗ <b>{TelegramFormatter.Escape(end.ToolName)}</b>{tool.Args}{FormatExcerpt(end.Result)}"
|
||||
: $"✓ <b>{TelegramFormatter.Escape(end.ToolName)}</b>{tool.Args}";
|
||||
if (started)
|
||||
await _sender.EditHtmlAsync(_chatId, tool.MessageId, line, StripTags(line), ct).ConfigureAwait(false);
|
||||
else
|
||||
await SendHtmlLineAsync(line, ct).ConfigureAwait(false);
|
||||
break;
|
||||
|
||||
case AutoRetryStart retry:
|
||||
await SendLineAsync($"⏳ {retry.ErrorMessage} — retrying in {retry.Delay.TotalSeconds:0} s ({retry.Attempt}/{retry.MaxAttempts})", ct).ConfigureAwait(false);
|
||||
break;
|
||||
case AutoRetryEnd { Success: false } retryEnd:
|
||||
await SendLineAsync($"❌ Giving up after {retryEnd.Attempt} attempts: {retryEnd.FinalError}", ct).ConfigureAwait(false);
|
||||
break;
|
||||
case CompactionStart:
|
||||
await SendLineAsync("🗜 Compacting the conversation…", ct).ConfigureAwait(false);
|
||||
break;
|
||||
case CompactionEnd { ErrorMessage: { } error }:
|
||||
await SendLineAsync($"❌ Compaction failed: {error}", ct).ConfigureAwait(false);
|
||||
break;
|
||||
case CompactionEnd { Result: { } result }:
|
||||
await SendLineAsync($"🗜 Compacted: ~{result.TokensBefore:N0} → ~{result.EstimatedTokensAfter:N0} tokens.", ct).ConfigureAwait(false);
|
||||
break;
|
||||
case Notification notification:
|
||||
await SendLineAsync($"{Icon(notification.Kind)} {notification.Message}", ct).ConfigureAwait(false);
|
||||
break;
|
||||
case ExtensionError error:
|
||||
_logger.LogWarning("Extension {Extension} failed on {Event}: {Error}", error.ExtensionName, error.Event, error.Error);
|
||||
await SendLineAsync($"⚠️ {error.ExtensionName}: {error.Error}", ct).ConfigureAwait(false);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private async Task FlushReplyAsync(CancellationToken ct)
|
||||
{
|
||||
_replyDirty = false;
|
||||
_lastFlush = _time.GetUtcNow();
|
||||
if (string.IsNullOrWhiteSpace(_replyText)) return;
|
||||
|
||||
var chunks = TelegramFormatter.Split(_replyText);
|
||||
for (var i = 0; i < chunks.Count; i++)
|
||||
{
|
||||
var chunk = chunks[i];
|
||||
var html = TelegramFormatter.ToHtml(chunk);
|
||||
if (i < _replyMessages.Count)
|
||||
{
|
||||
if (_replyMessages[i].Sent == chunk) continue;
|
||||
if (await _sender.EditHtmlAsync(_chatId, _replyMessages[i].MessageId, html, chunk, ct).ConfigureAwait(false))
|
||||
_replyMessages[i] = (_replyMessages[i].MessageId, chunk);
|
||||
}
|
||||
else if (await _sender.SendHtmlAsync(_chatId, html, chunk, cancellationToken: ct).ConfigureAwait(false) is { } id)
|
||||
{
|
||||
_replyMessages.Add((id, chunk));
|
||||
}
|
||||
else
|
||||
{
|
||||
break; // try the rest on the next flush
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void ResetReply()
|
||||
{
|
||||
_replyMessages.Clear();
|
||||
_replyText = "";
|
||||
_replyDirty = false;
|
||||
_lastFlush = DateTimeOffset.MinValue;
|
||||
}
|
||||
|
||||
private Task<int?> SendLineAsync(string text, CancellationToken ct) =>
|
||||
_sender.SendHtmlAsync(_chatId, TelegramFormatter.Escape(text), text, cancellationToken: ct);
|
||||
|
||||
private Task<int?> SendHtmlLineAsync(string html, CancellationToken ct) =>
|
||||
_sender.SendHtmlAsync(_chatId, html, StripTags(html), cancellationToken: ct);
|
||||
|
||||
private void StartTyping()
|
||||
{
|
||||
if (_typing is not null) return;
|
||||
var cts = CancellationTokenSource.CreateLinkedTokenSource(_stop.Token);
|
||||
_typing = cts;
|
||||
_ = Task.Run(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
while (!cts.IsCancellationRequested)
|
||||
{
|
||||
await _sender.SendTypingAsync(_chatId, cts.Token).ConfigureAwait(false);
|
||||
await Task.Delay(_options.TypingInterval, cts.Token).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private void StopTyping()
|
||||
{
|
||||
var typing = Interlocked.Exchange(ref _typing, null);
|
||||
typing?.Cancel();
|
||||
typing?.Dispose();
|
||||
}
|
||||
|
||||
/// <summary>": <code>ls -la</code>" for bash, the most telling argument otherwise.</summary>
|
||||
internal static string FormatArgs(JsonElement args)
|
||||
{
|
||||
if (args.ValueKind != JsonValueKind.Object) return "";
|
||||
string? summary = null;
|
||||
foreach (var key in new[] { "command", "path", "pattern", "file_path" })
|
||||
{
|
||||
if (args.TryGetProperty(key, out var value) && value.ValueKind == JsonValueKind.String)
|
||||
{
|
||||
summary = value.GetString();
|
||||
break;
|
||||
}
|
||||
}
|
||||
summary ??= args.EnumerateObject().Any() ? args.GetRawText() : null;
|
||||
if (string.IsNullOrWhiteSpace(summary)) return "";
|
||||
return $": <code>{TelegramFormatter.Escape(Shorten(summary, 200))}</code>";
|
||||
}
|
||||
|
||||
private static string FormatExcerpt(ToolResult result)
|
||||
{
|
||||
var text = string.Concat(result.Content.OfType<TextContent>().Select(c => c.Text)).Trim();
|
||||
return text.Length == 0 ? "" : $"\n<pre>{TelegramFormatter.Escape(Shorten(text, ToolExcerptLength))}</pre>";
|
||||
}
|
||||
|
||||
private static string Shorten(string text, int max) => text.Length <= max ? text : text[..max] + "…";
|
||||
|
||||
private static string StripTags(string html) =>
|
||||
System.Net.WebUtility.HtmlDecode(System.Text.RegularExpressions.Regex.Replace(html, "<[^>]+>", ""));
|
||||
|
||||
private static string Icon(NotifyKind kind) => kind switch { NotifyKind.Warning => "⚠️", NotifyKind.Error => "❌", _ => "ℹ️" };
|
||||
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
_events.Writer.TryComplete();
|
||||
try
|
||||
{
|
||||
await _loop.WaitAsync(TimeSpan.FromSeconds(5)).ConfigureAwait(false);
|
||||
}
|
||||
catch (TimeoutException)
|
||||
{
|
||||
}
|
||||
await _stop.CancelAsync().ConfigureAwait(false);
|
||||
StopTyping();
|
||||
_stop.Dispose();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,187 @@
|
||||
using System.Text;
|
||||
using System.Text.RegularExpressions;
|
||||
|
||||
namespace Volcanic.AgenticCore.Telegram.Rendering;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public static partial class TelegramFormatter
|
||||
{
|
||||
/// <summary>Telegram's limit for one message, in UTF-16 code units after entity parsing.</summary>
|
||||
public const int MessageLimit = 4096;
|
||||
|
||||
/// <summary>Raw text per chunk; leaves room for the HTML tags and escapes the conversion adds.</summary>
|
||||
public const int ChunkLimit = 3500;
|
||||
|
||||
private const string Fence = "```";
|
||||
|
||||
/// <summary>Escape the three characters Telegram HTML requires; everything else (emoji included) stays as is.</summary>
|
||||
public static string Escape(string text) => text.Replace("&", "&").Replace("<", "<").Replace(">", ">");
|
||||
|
||||
/// <summary>Convert Markdown to Telegram HTML.</summary>
|
||||
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 }
|
||||
? $"<pre><code class=\"language-{Escape(language)}\">{body}</code></pre>"
|
||||
: $"<pre>{body}</pre>");
|
||||
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 $"<b>{FormatInline(heading.Groups[1].Value)}</b>";
|
||||
if (BulletRegex().Match(line) is { Success: true } bullet) return $"{bullet.Groups[1].Value}• {FormatInline(bullet.Groups[2].Value)}";
|
||||
return FormatInline(line);
|
||||
}
|
||||
|
||||
/// <summary>Inline code, bold, italics, strikethrough and links. Code spans are cut out first so their content stays literal.</summary>
|
||||
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("<code>").Append(Escape(m.Groups[1].Value)).Append("</code>");
|
||||
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 => $"<a href=\"{m.Groups[2].Value}\">{m.Groups[1].Value}</a>");
|
||||
s = BoldRegex().Replace(s, "<b>$2</b>");
|
||||
s = StrikeRegex().Replace(s, "<s>$1</s>");
|
||||
s = ItalicRegex().Replace(s, "$1<i>$3</i>");
|
||||
return s;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Split raw Markdown into chunks of at most <paramref name="limit"/> 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.
|
||||
/// </summary>
|
||||
public static IReadOnlyList<string> Split(string text, int limit = ChunkLimit)
|
||||
{
|
||||
if (text.Length <= limit) return [text];
|
||||
var chunks = new List<string>();
|
||||
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;
|
||||
}
|
||||
|
||||
/// <returns>Where the chunk ends and how many separator characters to drop after it.</returns>
|
||||
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);
|
||||
}
|
||||
|
||||
/// <summary>The opening fence line (with language) when <paramref name="text"/> ends inside a code block.</summary>
|
||||
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();
|
||||
}
|
||||
126
hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramSender.cs
Normal file
126
hosts/Volcanic.AgenticCore.Telegram/Rendering/TelegramSender.cs
Normal file
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
public sealed class TelegramSender(ITelegramBotClient bot, ILogger<TelegramSender> logger)
|
||||
{
|
||||
public ITelegramBotClient Bot => bot;
|
||||
|
||||
/// <summary>Send HTML; returns the message id, or null when sending failed.</summary>
|
||||
public async Task<int?> 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;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Send plain text; returns the message id, or null when sending failed.</summary>
|
||||
public async Task<int?> 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;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Replace a message's text with HTML (plain fallback). Returns false when the edit failed.</summary>
|
||||
public async Task<bool> 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);
|
||||
}
|
||||
62
hosts/Volcanic.AgenticCore.Telegram/TelegramOptions.cs
Normal file
62
hosts/Volcanic.AgenticCore.Telegram/TelegramOptions.cs
Normal file
@@ -0,0 +1,62 @@
|
||||
namespace Volcanic.AgenticCore.Telegram;
|
||||
|
||||
/// <summary>The <c>Telegram</c> configuration section.</summary>
|
||||
public sealed class TelegramOptions
|
||||
{
|
||||
public const string SectionName = "Telegram";
|
||||
public const string DefaultModel = "claude-code/claude-opus-5-5";
|
||||
|
||||
/// <summary>Bot token from @BotFather. Also read from <c>TELEGRAM_BOT_TOKEN</c>.</summary>
|
||||
public string BotToken { get; set; } = "";
|
||||
|
||||
/// <summary>Telegram user ids allowed to talk to the bot. Required: the agent runs shell commands.</summary>
|
||||
public List<long> AllowedUserIds { get; set; } = [];
|
||||
|
||||
/// <summary>Each chat works in <c><WorkspaceRoot>/<chatId></c>.</summary>
|
||||
public string WorkspaceRoot { get; set; } = Path.Combine(
|
||||
Environment.GetFolderPath(Environment.SpecialFolder.UserProfile), ".volcanic", "telegram-workspaces");
|
||||
|
||||
/// <summary><c>provider/model-id</c> every chat uses.</summary>
|
||||
public string Model { get; set; } = DefaultModel;
|
||||
|
||||
/// <summary>Minimum time between two edits of a streaming reply (Telegram rate-limits edits).</summary>
|
||||
public TimeSpan EditInterval { get; set; } = TimeSpan.FromSeconds(1.5);
|
||||
|
||||
/// <summary>How often the "typing…" action is repeated while the agent runs (Telegram shows it for 5 s).</summary>
|
||||
public TimeSpan TypingInterval { get; set; } = TimeSpan.FromSeconds(4);
|
||||
|
||||
/// <summary>First delay after a failed poll; doubles per consecutive failure up to <see cref="ReconnectMaxDelay"/>.</summary>
|
||||
public TimeSpan ReconnectInitialDelay { get; set; } = TimeSpan.FromSeconds(1);
|
||||
|
||||
public TimeSpan ReconnectMaxDelay { get; set; } = TimeSpan.FromSeconds(60);
|
||||
|
||||
/// <summary>Reads the environment-variable shortcuts the README documents, when the section leaves them empty.</summary>
|
||||
public void ApplyEnvironment(Func<string, string?> 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);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Configuration problems that must stop the host, empty when valid.</summary>
|
||||
public IReadOnlyList<string> Validate()
|
||||
{
|
||||
var errors = new List<string>();
|
||||
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);
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
<PropertyGroup>
|
||||
<OutputType>Exe</OutputType>
|
||||
<UserSecretsId>volcanic-agentic-core-telegram</UserSecretsId>
|
||||
</PropertyGroup>
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="../../src/Volcanic.AgenticCore/Volcanic.AgenticCore.csproj" />
|
||||
<ProjectReference Include="../../extensions/Volcanic.AgenticCore.Extensions.ClaudeCode/Volcanic.AgenticCore.Extensions.ClaudeCode.csproj" />
|
||||
<PackageReference Include="Microsoft.Extensions.Hosting" Version="10.0.0" />
|
||||
<PackageReference Include="Telegram.Bot" Version="22.10.3.1" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<None Update="appsettings.json" CopyToOutputDirectory="PreserveNewest" />
|
||||
<InternalsVisibleTo Include="Volcanic.AgenticCore.Telegram.Tests" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
14
hosts/Volcanic.AgenticCore.Telegram/appsettings.json
Normal file
14
hosts/Volcanic.AgenticCore.Telegram/appsettings.json
Normal file
@@ -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"
|
||||
}
|
||||
}
|
||||
@@ -105,7 +105,7 @@ public class ProviderExtensionTests
|
||||
Assert.Equal("claude-code-headless", provider.Api);
|
||||
Assert.Equal("claude-code://local", provider.BaseUrl);
|
||||
Assert.False(provider.RequiresApiKey);
|
||||
Assert.Equal(["sonnet", "opus", "haiku", "fable"], provider.Models.Select(m => m.Id).ToArray());
|
||||
Assert.Equal(["sonnet", "opus", "haiku", "fable", "claude-opus-5-5"], provider.Models.Select(m => m.Id).ToArray());
|
||||
Assert.All(provider.Models, m => Assert.Equal(ModelCost.Zero, m.Cost));
|
||||
Assert.Equal(1_000_000, provider.Models[0].ContextWindow);
|
||||
Assert.Equal(64_000, provider.Models[0].MaxTokens);
|
||||
|
||||
220
tests/Volcanic.AgenticCore.Telegram.Tests/ChatTests.cs
Normal file
220
tests/Volcanic.AgenticCore.Telegram.Tests/ChatTests.cs
Normal file
@@ -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 <b>there</b>, 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("✓ <b>ls</b>: <code>.</code>", texts);
|
||||
Assert.Equal("Nothing here.", texts.Last());
|
||||
Assert.Contains(host.Bot.Sent, s => s.Text.StartsWith("▶ <b>ls</b>", 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);
|
||||
}
|
||||
}
|
||||
@@ -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_", "<b>bold</b> and <i>it</i> and <i>it</i>")]
|
||||
[InlineData("use `x < 1` here", "use <code>x < 1</code> here")]
|
||||
[InlineData("`**not bold**`", "<code>**not bold**</code>")]
|
||||
[InlineData("snake_case_name stays", "snake_case_name stays")]
|
||||
[InlineData("# Title", "<b>Title</b>")]
|
||||
[InlineData("- item", "• item")]
|
||||
[InlineData("[docs](https://example.com/a?b=1&c=2)", "<a href=\"https://example.com/a?b=1&c=2\">docs</a>")]
|
||||
[InlineData("~~gone~~", "<s>gone</s>")]
|
||||
[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 \"<hi>\" **x**\n```\nDone.");
|
||||
Assert.Equal("Run:\n<pre><code class=\"language-bash\">echo \"<hi>\" **x**</code></pre>\nDone.", html);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void An_unterminated_code_block_while_streaming_is_still_well_formed()
|
||||
{
|
||||
Assert.Equal("<pre>partial <</pre>", 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<string, string> { ["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));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
global using Xunit;
|
||||
139
tests/Volcanic.AgenticCore.Telegram.Tests/PollingServiceTests.cs
Normal file
139
tests/Volcanic.AgenticCore.Telegram.Tests/PollingServiceTests.cs
Normal file
@@ -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<Update, Task>? onRoute = null) : IUpdateRouter
|
||||
{
|
||||
private readonly List<int> _ids = [];
|
||||
public IReadOnlyList<int> 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<PollingService>.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));
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>The bot's chat side wired to a <see cref="FakeBotClient"/> and the <see cref="FauxProvider"/>.</summary>
|
||||
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<TelegramSender>.Instance);
|
||||
Chats = new ChatRegistry(_services.GetRequiredService<IAgentSessionFactory>(), sender, Options, loggers);
|
||||
Router = new UpdateRouter(Options, Chats, new ChatCommands(sender), sender, NullLogger<UpdateRouter>.Instance);
|
||||
}
|
||||
|
||||
public Task SendAsync(string text, long userId = Updates.AllowedUser) => Router.RouteAsync(Updates.Text(text, userId), CancellationToken.None);
|
||||
|
||||
public async Task<ChatContext> ChatAsync() => await Chats.GetAsync(Updates.AllowedUser);
|
||||
|
||||
/// <summary>Wait until the chat's run has settled and everything it produced is rendered.</summary>
|
||||
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<string> 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) { }
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// In-memory <see cref="ITelegramBotClient"/>: records every request, answers sends with fake messages and plays a
|
||||
/// script of <c>getUpdates</c> results. When the script is exhausted a poll blocks until cancelled, like an idle long poll.
|
||||
/// </summary>
|
||||
public sealed class FakeBotClient : ITelegramBotClient
|
||||
{
|
||||
private readonly ConcurrentQueue<Func<Update[]>> _polls = new();
|
||||
private readonly List<IRequest> _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<ApiRequestEventArgs>? OnMakingApiRequest;
|
||||
public event AsyncEventHandler<ApiResponseEventArgs>? OnApiResponseReceived;
|
||||
|
||||
/// <summary>Called for every non-poll request before it is answered; throw to simulate a failure.</summary>
|
||||
public Action<IRequest>? OnSend { get; set; }
|
||||
|
||||
public IReadOnlyList<IRequest> Requests { get { lock (_requests) return _requests.ToList(); } }
|
||||
/// <summary>The offset of every <c>getUpdates</c> call (the library reuses one request object, so it is captured per call).</summary>
|
||||
public IReadOnlyList<int?> PollOffsets { get { lock (_pollOffsets) return _pollOffsets.ToList(); } }
|
||||
private readonly List<int?> _pollOffsets = [];
|
||||
public IReadOnlyList<SendMessageRequest> Sent => Requests.OfType<SendMessageRequest>().ToList();
|
||||
public IReadOnlyList<EditMessageTextRequest> Edits => Requests.OfType<EditMessageTextRequest>().ToList();
|
||||
|
||||
/// <summary>The text each sent message shows now (after edits), by message id.</summary>
|
||||
public IReadOnlyDictionary<int, string> CurrentTexts => new Dictionary<int, string>(_texts);
|
||||
|
||||
private readonly ConcurrentDictionary<int, string> _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<TResponse> SendRequest<TResponse>(IRequest<TResponse> 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<bool> 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;
|
||||
}
|
||||
65
tests/Volcanic.AgenticCore.Telegram.Tests/Support/Updates.cs
Normal file
65
tests/Volcanic.AgenticCore.Telegram.Tests/Support/Updates.cs
Normal file
@@ -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<bool> 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();
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
<PropertyGroup>
|
||||
<IsPackable>false</IsPackable>
|
||||
<IsTestProject>true</IsTestProject>
|
||||
<TreatWarningsAsErrors>false</TreatWarningsAsErrors>
|
||||
</PropertyGroup>
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.14.1" />
|
||||
<PackageReference Include="xunit" Version="2.9.3" />
|
||||
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.1" />
|
||||
</ItemGroup>
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="../../hosts/Volcanic.AgenticCore.Telegram/Volcanic.AgenticCore.Telegram.csproj" />
|
||||
</ItemGroup>
|
||||
</Project>
|
||||
Reference in New Issue
Block a user