X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/522d7fd117285e4e4ce3ea5ec9b5a1722521b681..b1be0ee8530df5db8973bfe355f2f6d3149383a0:/telegram.go diff --git a/telegram.go b/telegram.go index 2a9345a..f3b3bb2 100644 --- a/telegram.go +++ b/telegram.go @@ -7,7 +7,7 @@ import ( "errors" "fmt" "io" - "log" + "log/slog" "net/http" "os" "path/filepath" @@ -18,6 +18,14 @@ import ( ) 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" @@ -97,14 +105,12 @@ func loadTelegramConfig(dir string) (TelegramConfig, bool, error) { // newTelegramBot returns the bot of owner on api. The offset file is at // offsetPath, and logger gets the failures. No log line holds the text of a // message. -func newTelegramBot(api *TelegramAPI, owner int64, offsetPath string, logger *log.Logger) *TelegramBot { +func newTelegramBot(api *TelegramAPI, owner int64, offsetPath string, logger *slog.Logger) *TelegramBot { return &TelegramBot{ api: api, owner: owner, offsetPath: offsetPath, logger: logger, - minWait: time.Second, - maxWait: time.Minute, ready: make(chan struct{}, 1), } } @@ -133,30 +139,28 @@ func (c TelegramConfig) validate() error { } // TelegramBot connects the private chat of the owner with the agent. It polls -// the Bot API for new messages and sends each text message of the owner 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 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 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 owner int64 offsetPath string - logger *log.Logger - minWait time.Duration - maxWait time.Duration + logger *slog.Logger mutex sync.Mutex queue []Event @@ -193,8 +197,8 @@ func (b *TelegramBot) Notify(event Event) { } // 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 -// cannot be written, because then the bot would lose or repeat messages. +// 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. func (b *TelegramBot) Run(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) error { ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -258,9 +262,9 @@ func (b *TelegramBot) send(ctx context.Context) { } } -// poll gets the updates from the stored offset on and handles each one, until -// ctx ends. Each update is handled to its end also when ctx ends during it, so -// a stop does not repeat its message. +// poll gets updates from the stored offset and handles them until ctx ends. A +// failed image download retries from the same offset. A submitted message is +// flushed even if ctx ends, and only then can the offset advance. func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) error { offset, err := telegramLoadOffset(b.offsetPath) if err != nil { @@ -273,7 +277,16 @@ func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) back: "receive back", } - wait := b.minWait + imageOutage := &TelegramOutage{ + logger: b.logger, + loss: "image receive failed", + back: "image receive back", + } + + imageWait := telegramRetryDelay + wait := telegramRetryDelay + +poll: for ctx.Err() == nil { updates, err := b.api.GetUpdates(ctx, offset) if ctx.Err() != nil { @@ -289,17 +302,28 @@ func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) outage.End() - wait = b.minWait + wait = telegramRetryDelay + for _, update := range updates { if ctx.Err() != nil { return nil } - err = b.handle(context.WithoutCancel(ctx), l, whatsapp, update) + err = b.handle(ctx, l, whatsapp, update) if err != nil { + retry, asked := telegramRetryable(err) + if retry { + imageWait = b.pause(ctx, imageOutage, err, imageWait, asked) + continue poll + } + return err } + imageOutage.End() + + imageWait = telegramRetryDelay + offset = update.UpdateID + 1 err = telegramSaveOffset(b.offsetPath, offset) if err != nil { @@ -353,7 +377,8 @@ func (b *TelegramBot) post(ctx context.Context, event Event) { } // 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) @@ -366,20 +391,20 @@ func (b *TelegramBot) pause(ctx context.Context, outage *TelegramOutage, err err 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 // to the agent, and handle returns when it is durable in the log. The line // /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. Any other message of the owner gets the answer -// that only text is supported. Every other update is ignored, and its content -// is not logged. +// 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. 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 { - b.logger.Print("message ignored: not from the owner") + b.logger.Info("message ignored: not from the owner") return nil } @@ -387,6 +412,41 @@ func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, whatsapp *WhatsAp name, stop := stopSubagentName(message.Text) args, isWhatsApp := whatsappCommandArgs(message.Text) switch { + case len(message.Photo) > 0 || message.Document != nil: + text, err := telegramImageText(ctx, b.api, l.agent.Images(), message) + if err != nil { + retry, _ := telegramRetryable(err) + if retry { + return fmt.Errorf("receive image: %w", err) + } + + switch { + case errors.Is(err, errTelegramOtherMedia): + b.notice(telegramTextOnly) + case errors.Is(err, errImageSize), errors.Is(err, errImageFormat), errors.Is(err, errTelegramPhotoSize): + b.notice(err.Error()) + default: + var answer *TelegramError + + if !errors.As(err, &answer) { + return fmt.Errorf("receive Telegram image: %w", err) + } + + b.notice("Cannot download image.") + } + + return nil + } + + _, err = l.Submit(telegramSource, text) + if err != nil { + return fmt.Errorf("submit Telegram image: %w", err) + } + + err = l.Flush(context.WithoutCancel(ctx)) + if err != nil { + return fmt.Errorf("flush Telegram image: %w", err) + } case message.Text == "": b.notice(telegramTextOnly) case strings.TrimSpace(message.Text) == stopCommand: @@ -410,7 +470,7 @@ func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, whatsapp *WhatsAp return err } - err = l.Flush(ctx) + err = l.Flush(context.WithoutCancel(ctx)) if err != nil { return fmt.Errorf("flush the message from Telegram: %w", err) } @@ -425,7 +485,7 @@ func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, whatsapp *WhatsAp // 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) @@ -440,7 +500,7 @@ func (b *TelegramBot) deliver(ctx context.Context, part TelegramPart) { var answer *TelegramError if len(part.entities) > 0 && errors.As(err, &answer) && answer.Status == http.StatusBadRequest { - b.logger.Printf("reply sent as plain text: %v", err) + b.logger.Warn("reply sent as plain text", "err", err) part.entities = nil @@ -448,7 +508,7 @@ func (b *TelegramBot) deliver(ctx context.Context, part TelegramPart) { } if !retry { - b.logger.Printf("reply lost: %v", err) + b.logger.Error("reply lost", "err", err) return } @@ -466,7 +526,7 @@ func (b *TelegramBot) deliver(ctx context.Context, part TelegramPart) { // 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) @@ -479,7 +539,7 @@ func (b *TelegramBot) deliverRich(ctx context.Context, markdown string) { } if telegramRejected(err) { - b.logger.Printf("reply sent with simple formatting: %v", err) + b.logger.Warn("reply sent with simple formatting", "err", err) for _, part := range telegramReplyParts(telegramUnescapeTags(markdown)) { b.deliver(ctx, part) @@ -489,7 +549,7 @@ func (b *TelegramBot) deliverRich(ctx context.Context, markdown string) { } if !retry { - b.logger.Printf("reply lost: %v", err) + b.logger.Error("reply lost", "err", err) return } @@ -520,7 +580,7 @@ func (b *TelegramBot) notice(text string) { // 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 *log.Logger + logger *slog.Logger loss string back string start time.Time @@ -533,7 +593,7 @@ func (o *TelegramOutage) Fail(err error, delay time.Duration) { } o.start = time.Now() - o.logger.Printf("%s: %v (retry in %s)", o.loss, err, durationText(delay)) + o.logger.Warn(o.loss, "err", err, "retry", delay) } // End logs the end of the outage that runs, with its length. @@ -542,7 +602,7 @@ func (o *TelegramOutage) End() { return } - o.logger.Printf("%s after %s", o.back, durationText(time.Since(o.start))) + o.logger.Info(o.back, "after", time.Since(o.start)) o.start = time.Time{} } @@ -721,10 +781,10 @@ func telegramRichCut(markdown string, limit int) (int, string, string) { end += start + 1 text := strings.TrimRight(markdown[start:end], "\r\n") - _, opens := telegramCodeFence(text) + _, opens := markdownCodeFence(text) switch { case fence != "": - if telegramClosesFence(text, fence) { + if markdownClosesFence(text, fence) { fence = "" } else if end > len(opening)+1 && end+len(fence)+1 <= limit { code, inFence, inOpening = end, fence, opening