X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/48d7bce28b18ecef4c52c98821675605464cecb9..d7bc0a6ac452bfc646610fdfd7c8eccb5c55218f:/telegram.go diff --git a/telegram.go b/telegram.go index f3b3bb2..a024e82 100644 --- a/telegram.go +++ b/telegram.go @@ -58,6 +58,66 @@ const ( telegramTextOnly = "Only text messages are supported." ) +// TelegramConfig holds the settings of telegram.json: the bot token and the +// user ID of the owner. The token is a secret, so the file must have mode 0600. +type TelegramConfig struct { + Token string `json:"token"` + Owner int64 `json:"owner"` +} + +// TelegramBot connects the private chat of the owner with the agent. It polls +// 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 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 + + mutex sync.Mutex + queue []Event + ready chan struct{} +} + +// TelegramOutage is one run of failures of a kind of Bot API call. The first +// failure logs loss, its cause, and the wait of the next try. The retries log +// nothing. The next answer of Telegram logs back and the length of the outage. +// loss and back say what the user loses and gets again. +type TelegramOutage struct { + logger *slog.Logger + loss string + back string + start time.Time +} + +// TelegramPart is the text of one message to the chat, at most +// telegramMessageUnits long, and its entities. +type TelegramPart struct { + text string + entities []TelegramEntity +} + +// TelegramOffsetFile is the content of the offset file. +type TelegramOffsetFile struct { + Offset int64 `json:"offset"` +} + // loadTelegramConfig reads telegram.json in dir. It returns false when the file // does not exist. An error never holds the token. func loadTelegramConfig(dir string) (TelegramConfig, bool, error) { @@ -108,6 +168,7 @@ func loadTelegramConfig(dir string) (TelegramConfig, bool, error) { func newTelegramBot(api *TelegramAPI, owner int64, offsetPath string, logger *slog.Logger) *TelegramBot { return &TelegramBot{ api: api, + voice: &VoiceTranscriber{}, owner: owner, offsetPath: offsetPath, logger: logger, @@ -115,13 +176,6 @@ func newTelegramBot(api *TelegramAPI, owner int64, offsetPath string, logger *sl } } -// TelegramConfig holds the settings of telegram.json: the bot token and the -// user ID of the owner. The token is a secret, so the file must have mode 0600. -type TelegramConfig struct { - Token string `json:"token"` - Owner int64 `json:"owner"` -} - // validate makes sure that the token has the form of a bot token and that the // owner is a user ID. The token goes into a URL path, so it can only hold the // characters of a bot token. @@ -138,64 +192,6 @@ func (c TelegramConfig) validate() error { return nil } -// 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 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 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 - owner int64 - offsetPath string - logger *slog.Logger - - mutex sync.Mutex - queue []Event - ready chan struct{} -} - -// Notify queues event for the chat. It never blocks, so a slow network does not -// stop the agent. The WhatsApp batches and the runs of the schedules do not go -// to the chat: the reply of the agent tells the user what needs the user. -func (b *TelegramBot) Notify(event Event) { - switch event.Kind { - case "busy", "idle", "reply", "error", "notice", "subagent": - case "work": - if event.Source == whatsappSource || event.Source == scheduleSource { - return - } - case "user": - if event.Source == telegramSource { - return - } - default: - return - } - - b.mutex.Lock() - - b.queue = append(b.queue, event) - b.mutex.Unlock() - - select { - case b.ready <- struct{}{}: - default: - } -} - // Run polls for the messages of the owner and sends the queued events to the // chat until ctx ends. It returns an error only when a message or the offset or // image cannot be written, because then the bot would lose or repeat messages. @@ -268,7 +264,7 @@ func (b *TelegramBot) send(ctx context.Context) { func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) error { offset, err := telegramLoadOffset(b.offsetPath) if err != nil { - return err + return fmt.Errorf("load Telegram offset: %w", err) } outage := &TelegramOutage{ @@ -317,7 +313,7 @@ poll: continue poll } - return err + return fmt.Errorf("handle Telegram update: %w", err) } imageOutage.End() @@ -327,7 +323,7 @@ poll: offset = update.UpdateID + 1 err = telegramSaveOffset(b.offsetPath, offset) if err != nil { - return err + return fmt.Errorf("save Telegram offset: %w", err) } } } @@ -353,7 +349,8 @@ func (b *TelegramBot) next() (Event, bool) { // typing shows the typing action once. A failure is not logged, because the // next message or poll logs the state of the network. func (b *TelegramBot) typing(ctx context.Context) { - _ = b.api.SendTyping(ctx, b.owner) + // The next message or poll reports a network failure. + b.api.SendTyping(ctx, b.owner) // nolint: errcheck } // post sends event to the chat. A reply goes as rich messages of its Markdown, @@ -399,8 +396,9 @@ func (b *TelegramBot) pause(ctx context.Context, outage *TelegramOutage, err err // /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 { @@ -412,6 +410,17 @@ func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, whatsapp *WhatsAp 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 { @@ -467,7 +476,7 @@ func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, whatsapp *WhatsAp default: _, err := l.Submit(telegramSource, message.Text) if err != nil { - return err + return fmt.Errorf("submit Telegram text: %w", err) } err = l.Flush(context.WithoutCancel(ctx)) @@ -558,15 +567,6 @@ func (b *TelegramBot) deliverRich(ctx context.Context, markdown string) { } } -// sendOutage returns a new outage of the messages to the chat. -func (b *TelegramBot) sendOutage() *TelegramOutage { - return &TelegramOutage{ - logger: b.logger, - loss: "send failed", - back: "send back", - } -} - // notice queues a short answer of the bot for the chat. func (b *TelegramBot) notice(text string) { b.Notify(Event{ @@ -575,25 +575,42 @@ func (b *TelegramBot) notice(text string) { }) } -// TelegramOutage is one run of failures of a kind of Bot API call. The first -// failure logs loss, its cause, and the wait of the next try. The retries log -// nothing. The next answer of Telegram logs back and the length of the outage. -// loss and back say what the user loses and gets again. -type TelegramOutage struct { - logger *slog.Logger - loss string - back string - start time.Time +// sendOutage returns a new outage of the messages to the chat. +func (b *TelegramBot) sendOutage() *TelegramOutage { + return &TelegramOutage{ + logger: b.logger, + loss: "send failed", + back: "send back", + } } -// Fail logs the start of an outage with err and delay, unless an outage runs. -func (o *TelegramOutage) Fail(err error, delay time.Duration) { - if !o.start.IsZero() { +// Notify queues event for the chat. It never blocks, so a slow network does not +// stop the agent. The WhatsApp batches and the runs of the schedules do not go +// to the chat: the reply of the agent tells the user what needs the user. +func (b *TelegramBot) Notify(event Event) { + switch event.Kind { + case "busy", "idle", "reply", "error", "notice", "subagent": + case "work": + if event.Source == whatsappSource || event.Source == scheduleSource || event.Source == notifySource { + return + } + case "user": + if event.Source == telegramSource { + return + } + default: return } - o.start = time.Now() - o.logger.Warn(o.loss, "err", err, "retry", delay) + b.mutex.Lock() + + b.queue = append(b.queue, event) + b.mutex.Unlock() + + select { + case b.ready <- struct{}{}: + default: + } } // End logs the end of the outage that runs, with its length. @@ -607,16 +624,14 @@ func (o *TelegramOutage) End() { o.start = time.Time{} } -// TelegramPart is the text of one message to the chat, at most -// telegramMessageUnits long, and its entities. -type TelegramPart struct { - text string - entities []TelegramEntity -} +// Fail logs the start of an outage with err and delay, unless an outage runs. +func (o *TelegramOutage) Fail(err error, delay time.Duration) { + if !o.start.IsZero() { + return + } -// TelegramOffsetFile is the content of the offset file. -type TelegramOffsetFile struct { - Offset int64 `json:"offset"` + o.start = time.Now() + o.logger.Warn(o.loss, "err", err, "retry", delay) } // telegramLoadOffset reads the update offset at path. A missing file gives 0,