)
const (
+ // telegramRetryDelay is the wait before the first retry of a failed
+ // call.
+ telegramRetryDelay = time.Second
+
+ // telegramRetryLimit is the longest wait between retries of a failed
+ // call.
+ telegramRetryLimit = time.Minute
+
// telegramAPIURL is the Telegram Bot API service.
telegramAPIURL = "https://api.telegram.org"
func newTelegramBot(api *TelegramAPI, owner int64, offsetPath string, logger *slog.Logger) *TelegramBot {
return &TelegramBot{
api: api,
+ voice: &VoiceTranscriber{},
owner: owner,
offsetPath: offsetPath,
logger: logger,
- minWait: time.Second,
- maxWait: time.Minute,
ready: make(chan struct{}, 1),
}
}
}
// TelegramBot connects the private chat of the owner with the agent. It polls
-// the Bot API for new messages and sends owner text and images to the agent. It
-// ignores every other update. Its sender sends the replies, the errors, the
-// notices of the bot, and the user messages of the other clients to the chat,
-// and shows the typing action while a turn runs. Only the replies have a
-// format: their Markdown goes as rich messages, which Telegram renders itself,
-// and as message entities only when Telegram refuses a rich message. The rest
-// is plain text. Tool calls do not go to the chat.
+// the Bot API for new messages and sends owner text, images, and voice notes to
+// the agent. It ignores every other update. Its sender sends the replies, the
+// errors, the notices of the bot, and the user messages of the other clients to
+// the chat, and shows the typing action while a turn runs. Only the replies
+// have a format: their Markdown goes as rich messages, which Telegram renders
+// itself, and as message entities only when Telegram refuses a rich message.
+// The rest is plain text. Tool calls do not go to the chat.
//
// The offset file holds the ID of the next update. It moves past an update
// only after the message of the update is durable in the log, so a restart
// does not lose a message. A crash between the flush and the offset write
// makes Telegram send the update again, and the message repeats once.
//
-// A failed Bot API call waits and tries again, from minWait up to maxWait, or
-// as long as the Bot API asks. It never stops the agent. A TelegramOutage logs
-// the first failure and the end of each run of failures, and the retries
-// between them log nothing.
+// A failed Bot API call waits and tries again, from telegramRetryDelay up to
+// telegramRetryLimit, or as long as the Bot API asks. It never stops the agent.
+// A TelegramOutage logs the first failure and the end of each run of failures,
+// and the retries between them log nothing.
type TelegramBot struct {
api *TelegramAPI
+ voice *VoiceTranscriber
owner int64
offsetPath string
logger *slog.Logger
- minWait time.Duration
- maxWait time.Duration
mutex sync.Mutex
queue []Event
back: "image receive back",
}
- imageWait := b.minWait
- wait := b.minWait
+ imageWait := telegramRetryDelay
+ wait := telegramRetryDelay
poll:
for ctx.Err() == nil {
outage.End()
- wait = b.minWait
+ wait = telegramRetryDelay
for _, update := range updates {
if ctx.Err() != nil {
imageOutage.End()
- imageWait = b.minWait
+ imageWait = telegramRetryDelay
offset = update.UpdateID + 1
err = telegramSaveOffset(b.offsetPath, offset)
}
// pause gives err to outage and waits the longer of wait and asked, or until
-// ctx ends. It returns the wait of the next failure: twice wait, up to maxWait.
+// ctx ends. It returns the wait of the next failure: twice wait, up to
+// telegramRetryLimit.
func (b *TelegramBot) pause(ctx context.Context, outage *TelegramOutage, err error, wait, asked time.Duration) time.Duration {
delay := max(wait, asked)
outage.Fail(err, delay)
case <-ctx.Done():
}
- return min(2*wait, b.maxWait)
+ return min(2*wait, telegramRetryLimit)
}
// handle runs one update. A text message of the owner in the private chat goes
// /stop interrupts the turn, the line /stop NAME stops the subagent NAME, and a
// line that starts with /whatsapp runs a WhatsApp command on whatsapp, whose
// answers go to the chat only. Photos and image documents enter the shared
-// image store with their caption. Other media keep the text-only answer. Every
-// other update is ignored, and its content is not logged.
+// image store with their caption. Voice notes keep their audio and local
+// transcript. Other media keep the text-only answer. Every other update is
+// ignored, and its content is not logged.
func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp, update TelegramUpdate) error {
message := update.Message
if message == nil || message.Chat.Type != "private" || message.Chat.ID != b.owner || message.From == nil || message.From.ID != b.owner {
name, stop := stopSubagentName(message.Text)
args, isWhatsApp := whatsappCommandArgs(message.Text)
switch {
+ case message.Voice != nil:
+ text := telegramVoiceText(ctx, b.api, l.agent.Voices(), b.voice, message)
+ _, err := l.Submit(telegramSource, text)
+ if err != nil {
+ return fmt.Errorf("submit Telegram voice: %w", err)
+ }
+
+ err = l.Flush(context.WithoutCancel(ctx))
+ if err != nil {
+ return fmt.Errorf("flush Telegram voice: %w", err)
+ }
case len(message.Photo) > 0 || message.Document != nil:
text, err := telegramImageText(ctx, b.api, l.agent.Images(), message)
if err != nil {
// later try can fix, until ctx ends. It drops the part after any other failure.
func (b *TelegramBot) deliver(ctx context.Context, part TelegramPart) {
outage := b.sendOutage()
- wait := b.minWait
+ wait := telegramRetryDelay
for {
err := b.api.SendMessage(ctx, b.owner, part.text, part.entities)
retry, asked := telegramRetryable(err)
// fix, until ctx ends. It drops the message after any other failure.
func (b *TelegramBot) deliverRich(ctx context.Context, markdown string) {
outage := b.sendOutage()
- wait := b.minWait
+ wait := telegramRetryDelay
for {
err := b.api.SendRichMessage(ctx, b.owner, markdown)
retry, asked := telegramRetryable(err)
projects / onechat / blobdiff