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

fix(telegram): address review findings

Browse Source 
- /resume discards the inbox once the id names one conversation, before any await
- /stop discards input still waiting in the window
- a forwarded message is never a command or a pending answer
- ChatInbox.DrainAsync is an internal test helper
- tests: /resume and /stop drop buffered text, forwarded /new is content,
  a failing flush doesn't stop later batches; no wall-clock sleep in the drain test
- spec and user docs: timestamp exception for extension commands, command
  ordering, InputDebounce=0 wording
This commit is contained in:
EugeneTes
2026-09-26 20:15:57 +02:00
parent 0c1778e741
commit 2ec941bf2a
7 changed files with 104 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
docs/superpowers/specs/2026-09-26-telegram-message-context-design.md
View File

@@ -37,7 +37,7 @@ spec ports that behaviour. Header wording is kept identical so both bots read th
| Topic | Decision |
|---|---|
| Header format | One English line per header in square brackets, before the body: reply header first, then forward header, then the message text. Same wording as the reference. The batch's local timestamp line (next row) comes before all of them. |
| Local timestamp | Every prompt the host sends to the agent starts with one timestamp line in the chat's zone: `[Sat 2026-09-26 14:03 Europe/Berlin]` (invariant culture, 24 h, IANA id). **One per flushed batch**, not per part, before the first part's reply/forward headers. The time is the first part's `Message.Date` (when the user sent it, not when the batch flushed or a transcript finished), converted to the zone. It's cache-safe: it's in the appended user message, not in the system prompt prefix. Steering messages get it too, since they're prompts. |
| Local timestamp | Every prompt the host sends to the agent starts with one timestamp line in the chat's zone (except a `/command` the bot doesn't handle itself, which goes to the session as typed; see *What bypasses the inbox*): `[Sat 2026-09-26 14:03 Europe/Berlin]` (invariant culture, 24 h, IANA id). **One per flushed batch**, not per part, before the first part's reply/forward headers. The time is the first part's `Message.Date` (when the user sent it, not when the batch flushed or a transcript finished), converted to the zone. It's cache-safe: it's in the appended user message, not in the system prompt prefix. Steering messages get it too, since they're prompts. |
| Where the timestamp is built | **Host-side** (Telegram), not in the core. The core doesn't know when a user message was sent, and a headless caller may not want a header; the zone it uses is the same one the core session carries (`ChatSettingsStore.TimeZoneFor(chatId)`, which also feeds `AgentSessionOptions.TimeZone`, scheduler spec §7–§8), so prompt date, header and `bash`'s `TZ` always agree. Scheduled prompts and watcher matches get the same header from `TelegramJobDelivery` (scheduler spec §6). |
| Reply content | `quote.text` if present (reply with quote), else the replied message's `text` → `caption` → media descriptor. Capped at **500** Unicode scalars (first 499 + `…`). Quote entities are dropped. |
| Who was replied to | `ReplyToMessage.From.Id == bot.BotId` → "your message". Any other `From` → "your own earlier message" (private chats only, so it's the user). No `From` → "the earlier message". The bot id comes from `ITelegramBotClient.BotId` (parsed from the token), so no `getMe` is needed. |
@@ -49,7 +49,7 @@ spec ports that behaviour. Header wording is kept identical so both bots read th
| Race handling | The reference's pending counter, as leases. A lease is taken **synchronously in update order** while routing. The window timer is only armed when no lease is open. Parts merge in lease order, which is Telegram order, not completion order. This improves on the reference, which merges in completion order. |
| Hung preparation | A lease open longer than **2 min** (`ChatInbox` constructor default, not user config) stops holding the batch. The batch flushes without it, and the late part becomes a prompt of its own. |
| Slow work off the poll loop | Downloads (and the sibling's transcription) run in a background task per message, holding the lease. `RouteAsync` returns right after taking the lease, so `/stop` still gets through. Today the photo download is awaited inline. |
| What bypasses the inbox | Bot commands, callback queries, and plain text that answers a pending `TelegramUi` input. Unchanged rules, checked before the inbox, and only for messages with no attachment: a message with an attachment is never a command or a pending answer, and its caption is prompt text (attachments spec). `/new` and `/resume` discard the buffer (open leases still finish, but their parts are dropped). A `/command` the bot doesn't handle itself goes to the session directly, without headers or the window, so extension commands (`/claude-code-doctor`, `/claude-code-login`) still parse. `AgentSession.PromptAsync` only recognises a command at the start of the text. |
| What bypasses the inbox | Bot commands, callback queries, and plain text that answers a pending `TelegramUi` input. Unchanged rules, checked before the inbox, and only for messages with no attachment that weren't forwarded: a message with an attachment is never a command or a pending answer, and its caption is prompt text (attachments spec); a forwarded message is content, so a forwarded `/new …` goes to the agent with its `[Forwarded from …]` label. `/new`, `/stop` and a `/resume` that names exactly one conversation discard the buffer first, before any await, so a buffered message can't flush into the old conversation (open leases still finish, but their parts are dropped). A `/resume` whose switch an extension cancels has already dropped the buffer. A `/command` the bot doesn't handle itself goes to the session directly, without headers or the window, so extension commands (`/claude-code-doctor`, `/claude-code-login`) still parse. `AgentSession.PromptAsync` only recognises a command at the start of the text. Such a command doesn't wait for the batch in the inbox, so it can overtake messages sent just before it; that is accepted, since extension commands don't depend on them. |
| Delivery | A flush calls `ChatContext.PromptInBackground(text, images)`, which is unchanged. While a run is active the merged prompt steers it (`StreamingBehavior.Steer`), as today. |
| 👀 reaction | At flush time, if `session.IsStreaming`, every message in the batch gets a 👀 reaction (`setMessageReaction`), best-effort. An album has at most 10 items. It is not sent when the batch starts a new run. |
| 🟢 Online | Once per process, after `BotSetupService`'s first successful `getMe` + `setMyCommands`, it sends `🟢 Online.` to the owner (`AllowedUserIds[0]`). Best-effort: a failure (for example 403 because the owner never started the bot) is logged and not retried. |
@@ -161,8 +161,8 @@ public sealed class ChatInbox(Func<IReadOnlyList<InboundPart>, Task> onFlush, Ti
TimeProvider time, TimeSpan? maxHold = null, ILogger? logger = null) : IAsyncDisposable
{
public InboundLease Begin(int messageId); // reserves the next slot; cancels an armed timer
public void Discard(); // /new, /resume: drop buffered and in-flight parts (generation bump)
public Task DrainAsync(); // completes when nothing is buffered, no lease is open and the last flush was handed over
public void Discard(); // /new, /stop, /resume: drop buffered and in-flight parts (generation bump)
internal Task DrainAsync(); // test helper: completes when nothing is buffered, no lease is open and the last flush was handed over
}
public sealed class InboundLease : IDisposable
@@ -217,8 +217,8 @@ _ = Task.Run(async () =>
```
The pending-input rule adds one condition: a message with `ForwardOrigin` never answers a pending question. A
forwarded message is content, not an answer. `ChatCommands` calls `chat.Inbox.Discard()` on `/new` and a successful
`/resume`.
forwarded message is content, not an answer, and never a command either. `ChatCommands` calls `chat.Inbox.Discard()`
on `/new`, `/stop` and a `/resume` that names exactly one conversation, before its first await.
### `ChatContext` / `ChatRegistry` changes
@@ -309,6 +309,7 @@ signature, and a bot id taken from `BotId` when `From.IsBot` is false.
- `Discard` with an open lease: the late completion is dropped;
- double `Dispose` is a no-op;
- `maxHold` detaches a hung lease, and its late part flushes as its own batch;
- a throwing `onFlush` doesn't stop the next batch, and `DrainAsync` still completes;
- album header collapsing in the renderer.
**`MessageContextChatTests`** (through `ChatHost` + `FauxProvider`, asserting on `Faux.Requests[..].Messages.Last()`):
@@ -320,7 +321,8 @@ signature, and a bot id taken from `BotId` when `From.IsBot` is false.
- a three-photo album with one caption is one request with three images and one forward label;
- a message sent during a scripted long run (a `FauxTurn` tool that waits on a gate) gets a 👀
`SetMessageReactionRequest` and appears as a steering message. A message sent while idle gets no reaction;
- `/new` during the window drops the buffered text;
- `/new`, `/stop` and `/resume <id>` during the window drop the buffered text;
- a forwarded `/new …` reaches the model with its forward label and starts no new conversation;
- a forwarded text while `TelegramUi` waits for input goes to the agent, not to the question;
- `/stop` routed while a photo download is blocked (an `OnSend` gate on `GetFileRequest`) is handled right away,
which proves the poll loop isn't blocked.

6
docs/telegram-bot.md
View File

@@ -31,7 +31,7 @@ Environment variables, or the `Telegram` section of `appsettings.json` (`Telegra
| `Telegram:WorkspaceRoot` | `~/.volcanic/telegram-workspaces` | Chat `<id>` works in `<root>/<id>`; put `AGENTS.md`, `PERSONA.md`, `MEMORY.md` there |
| `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 the "Working..." status message |
| `Telegram:InputDebounce` | `00:00:00.6` | How long the bot waits after a message for more before sending them to the agent as one request. `0` sends each message on its own. Must not be negative |
| `Telegram:InputDebounce` | `00:00:00.6` | How long the bot waits after a message for more before sending them to the agent as one request. `0` disables the wait; messages still being read together (e.g. a photo still downloading and a text after it) are still sent as one request. Must not be negative |
| `Telegram:ReconnectInitialDelay` / `ReconnectMaxDelay` | `1 s` / `60 s` | Backoff after lost connection |
| `PROCESS_ENV_ALLOW` | — | Comma-separated names or globs let through to `bash`, scheduled scripts and MCP servers despite the deny-list (e.g. `GH_TOKEN` for `gh`). Added to `ProcessEnvironment:Allow` |
| `ProcessEnvironment:Deny` / `ProcessEnvironment:Allow` | built-in deny-list | Added to the defaults (`*_API_KEY`, `*_TOKEN`, `*_SECRET`, `*_PASSWORD`, their `Section__Key` forms and the named provider credentials), never replacing them. Allowing a name that is denied by name stops startup |
@@ -66,7 +66,9 @@ Along with your text, the agent sees:
`[Sat 2026-09-26 14:03 Europe/Berlin]`, once per request.
Messages sent within about 0.6 s of each other (`Telegram:InputDebounce`), such as a forward with your comment or
an album, reach the agent as one request. Every message waits that long before it's sent.
an album, reach the agent as one request. Every message waits that long before it's sent. Commands that an extension
handles (for example `/claude-code-doctor`) are sent at once, as typed, with no time header. `/new`, `/stop` and
`/resume` drop a message that is still waiting. A forwarded message is never taken as a command.
After a start the bot sends `🟢 Online.` to the owner (the first id in `AllowedUserIds`), once. If the owner never
started a chat with the bot, Telegram refuses the message; the bot logs that and carries on.

3
hosts/Volcanic.AgenticCore.Telegram/Chats/ChatCommands.cs
View File

@@ -46,6 +46,7 @@ public sealed class ChatCommands(TelegramSender sender)
return true;
case "stop":
chat.Inbox.Discard(); // what is still waiting in the window would start the run the user just stopped
if (!session.IsStreaming && !session.IsCompacting)
{
await ReplyAsync(chat, "Nothing is running.", cancellationToken).ConfigureAwait(false);
@@ -156,9 +157,9 @@ public sealed class ChatCommands(TelegramSender sender)
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.";
chat.Inbox.Discard(); // what was typed before /resume belongs to neither conversation; drop it before it can flush
if (chat.Session.IsStreaming) await chat.Session.AbortAsync().ConfigureAwait(false);
var result = await chat.Session.SwitchSessionAsync(matches[0].SessionId, cancellationToken).ConfigureAwait(false);
if (!result.Cancelled) chat.Inbox.Discard(); // what was typed before /resume belongs to the old conversation
return result.Cancelled ? "An extension cancelled the switch." : $"Resumed `{matches[0].SessionId}` ({matches[0].MessageCount} messages).";
}

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

@@ -68,8 +68,9 @@ public sealed class UpdateRouter(
var chat = await chats.GetAsync(message.Chat.Id, cancellationToken).ConfigureAwait(false);
var text = message.Text ?? message.Caption ?? "";
// Commands and pending answers are checked before the inbox, and only for messages without an attachment.
if (!InboundComposer.HasAttachment(message))
// Commands and pending answers are checked before the inbox, and only for messages without an attachment that
// weren't forwarded: a forwarded message is content, never a command or an answer.
if (!InboundComposer.HasAttachment(message) && message.ForwardOrigin is null)
{
if (ChatCommands.Parse(text) is not null)
{
@@ -78,8 +79,7 @@ public sealed class UpdateRouter(
_ = chat.PromptInBackground(text);
return;
}
// A forwarded message is content, not an answer.
if (message.ForwardOrigin is null && chat.Ui.IsWaitingForInput && chat.Ui.TryResolveInput(text)) return;
if (chat.Ui.IsWaitingForInput && chat.Ui.TryResolveInput(text)) return;
}
// Take the lease now, in update order; compose (downloads) in the background so polling isn't blocked.

6
hosts/Volcanic.AgenticCore.Telegram/Inbound/ChatInbox.cs
View File

@@ -77,14 +77,14 @@ public sealed class ChatInbox : IAsyncDisposable
}
}
/// <summary>Drop buffered parts and in-flight leases (<c>/new</c>, <c>/resume</c>). Leases already open complete into nothing.</summary>
/// <summary>Drop buffered parts and in-flight leases (<c>/new</c>, <c>/stop</c>, <c>/resume</c>). Leases already open complete into nothing.</summary>
public void Discard()
{
lock (_lock) DropLocked();
}
/// <summary>Completes when nothing is buffered, no lease is open and the last flush was handed over.</summary>
public async Task DrainAsync()
/// <summary>Completes when nothing is buffered, no lease is open and the last flush was handed over (test helper; polls).</summary>
internal async Task DrainAsync()
{
while (true)
{

25
tests/Volcanic.AgenticCore.Telegram.Tests/ChatInboxTests.cs
View File

@@ -202,12 +202,33 @@ public sealed class ChatInboxTests
inbox.Begin(1).Complete(Text(1));
var drain = inbox.DrainAsync();
await Task.Delay(50);
Assert.False(drain.IsCompleted);
Assert.False(drain.IsCompleted); // the part is still buffered
_time.Advance(Window);
await drain.WaitAsync(TimeSpan.FromSeconds(5));
}
[Fact]
public async Task A_failing_flush_does_not_stop_later_batches()
{
var calls = 0;
await using var inbox = new ChatInbox(parts =>
{
if (Interlocked.Increment(ref calls) == 1) throw new InvalidOperationException("boom");
lock (_flushes) _flushes.Add(parts);
return Task.CompletedTask;
}, Window, _time);
inbox.Begin(1).Complete(Text(1));
_time.Advance(Window);
await inbox.DrainAsync().WaitAsync(TimeSpan.FromSeconds(5)); // the flushing counter is back to zero
Assert.Equal(1, calls);
inbox.Begin(2).Complete(Text(2));
_time.Advance(Window);
await inbox.DrainAsync().WaitAsync(TimeSpan.FromSeconds(5));
Assert.Equal(["2"], Batches);
}
[Fact]
public async Task Dispose_drops_buffered_parts()
{

59
tests/Volcanic.AgenticCore.Telegram.Tests/MessageContextChatTests.cs
View File

@@ -202,6 +202,65 @@ public sealed class MessageContextChatTests
Assert.Contains("🆕 New conversation.", host.Texts);
}
[Fact]
public async Task Resume_during_the_window_drops_the_buffered_text()
{
var time = new FakeTimeProvider(DateTimeOffset.UtcNow);
await using var host = new ChatHost(time: time);
host.Faux.Enqueue(FauxTurn.Say("Hi."));
await host.SendAsync("hello");
await FlushAsync(host, time, parts: 1);
await host.SettleAsync();
var chat = await host.ChatAsync();
var earlier = chat.Session.SessionId;
await host.SendAsync("/new");
await host.SendAsync("forget me");
await Wait.UntilAsync(() => chat.Inbox.Buffered == 1 && chat.Inbox.OpenLeases == 0, "the buffered text");
await host.SendAsync($"/resume {earlier}");
Assert.Equal(0, chat.Inbox.Buffered);
time.Advance(host.Options.InputDebounce * 2);
await host.SettleAsync();
Assert.Single(host.Faux.Requests);
Assert.Equal(earlier, chat.Session.SessionId);
Assert.Contains(host.Texts, t => t.StartsWith("Resumed", StringComparison.Ordinal) && t.Contains(earlier, StringComparison.Ordinal));
}
[Fact]
public async Task Stop_during_the_window_drops_the_buffered_text()
{
var time = new FakeTimeProvider(DateTimeOffset.UtcNow);
await using var host = new ChatHost(time: time);
await host.SendAsync("never mind");
var chat = await host.ChatAsync();
await Wait.UntilAsync(() => chat.Inbox.Buffered == 1 && chat.Inbox.OpenLeases == 0, "the buffered text");
await host.SendAsync("/stop");
Assert.Equal(0, chat.Inbox.Buffered);
time.Advance(host.Options.InputDebounce * 2);
await host.SettleAsync();
Assert.Empty(host.Faux.Requests);
}
[Fact]
public async Task A_forwarded_command_is_content_not_a_command()
{
await using var host = new ChatHost();
host.Faux.Enqueue(FauxTurn.Say("Noted."));
await host.SendAsync("/status"); // creates the chat
var chat = await host.ChatAsync();
var session = chat.Session.SessionId;
await host.RouteAsync(Updates.Forward("/new beginnings", Updates.Channel("Daily News")));
await host.SettleAsync();
Assert.Equal(session, chat.Session.SessionId);
Assert.DoesNotContain("🆕 New conversation.", host.Texts);
Assert.EndsWith("\n[Forwarded from Daily News]\n/new beginnings", Prompt(host), StringComparison.Ordinal);
}
[Fact]
public async Task A_forwarded_text_while_input_is_pending_goes_to_the_agent()
{