X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/eeb821ff17ad79c87ca9b64cd9be26f8612f4a03..3bf0f670487e571a0525844a6c00de495b8a465a:/telegram.go?ds=inline diff --git a/telegram.go b/telegram.go index c560104..f2c687a 100644 --- a/telegram.go +++ b/telegram.go @@ -7,7 +7,7 @@ import ( "errors" "fmt" "io" - "log" + "log/slog" "net/http" "os" "path/filepath" @@ -73,8 +73,10 @@ func loadTelegramConfig(dir string) (TelegramConfig, bool, error) { } var config TelegramConfig + decoder := json.NewDecoder(bytes.NewReader(data)) decoder.DisallowUnknownFields() + err = decoder.Decode(&config) if err == nil && !errors.Is(decoder.Decode(&struct{}{}), io.EOF) { err = errors.New("data after the JSON object") @@ -95,7 +97,7 @@ 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, @@ -145,12 +147,14 @@ func (c TelegramConfig) validate() error { // 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. +// 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 + logger *slog.Logger minWait time.Duration maxWait time.Duration @@ -160,10 +164,15 @@ type TelegramBot struct { } // Notify queues event for the chat. It never blocks, so a slow network does not -// stop the agent. +// 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", "work": + case "busy", "idle", "reply", "error", "notice", "subagent": + case "work": + if event.Source == whatsappSource || event.Source == scheduleSource { + return + } case "user": if event.Source == telegramSource { return @@ -173,6 +182,7 @@ func (b *TelegramBot) Notify(event Event) { } b.mutex.Lock() + b.queue = append(b.queue, event) b.mutex.Unlock() @@ -185,19 +195,22 @@ 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. -func (b *TelegramBot) Run(ctx context.Context, l *TurnLoop) error { +func (b *TelegramBot) Run(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) error { ctx, cancel := context.WithCancel(ctx) defer cancel() var sender sync.WaitGroup + sender.Add(1) + go func() { defer sender.Done() b.send(ctx) }() - err := b.poll(ctx, l) + err := b.poll(ctx, l, whatsapp) + cancel() sender.Wait() @@ -230,6 +243,7 @@ func (b *TelegramBot) send(ctx context.Context) { switch event.Kind { case "busy": busy = true + b.typing(ctx) ticker.Reset(telegramTypingInterval) case "idle": @@ -247,12 +261,18 @@ 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. -func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop) error { +func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) error { offset, err := telegramLoadOffset(b.offsetPath) if err != nil { return err } + outage := &TelegramOutage{ + logger: b.logger, + loss: "receive failed", + back: "receive back", + } + wait := b.minWait for ctx.Err() == nil { updates, err := b.api.GetUpdates(ctx, offset) @@ -262,18 +282,20 @@ func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop) error { if err != nil { _, asked := telegramRetryable(err) - wait = b.pause(ctx, err, wait, asked) + wait = b.pause(ctx, outage, err, wait, asked) continue } + outage.End() + wait = b.minWait for _, update := range updates { if ctx.Err() != nil { return nil } - err = b.handle(context.WithoutCancel(ctx), l, update) + err = b.handle(context.WithoutCancel(ctx), l, whatsapp, update) if err != nil { return err } @@ -330,11 +352,11 @@ func (b *TelegramBot) post(ctx context.Context, event Event) { } } -// pause logs err 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. -func (b *TelegramBot) pause(ctx context.Context, err error, wait, asked time.Duration) time.Duration { +// 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. +func (b *TelegramBot) pause(ctx context.Context, outage *TelegramOutage, err error, wait, asked time.Duration) time.Duration { delay := max(wait, asked) - b.logger.Printf("%v (next try in %s)", err, delay) + outage.Fail(err, delay) timer := time.NewTimer(delay) defer timer.Stop() @@ -349,17 +371,21 @@ func (b *TelegramBot) pause(ctx context.Context, err error, wait, asked time.Dur // 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, and the line /stop NAME stops the subagent NAME. -// 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. -func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, update TelegramUpdate) error { +// /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. +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("update ignored: not a message of the owner in the private chat") + b.logger.Info("message ignored: not from the owner") + return nil } name, stop := stopSubagentName(message.Text) + args, isWhatsApp := whatsappCommandArgs(message.Text) switch { case message.Text == "": b.notice(telegramTextOnly) @@ -376,6 +402,8 @@ func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, update TelegramUp if !found { b.notice("No subagent " + name + " runs.") } + case isWhatsApp: + whatsapp.Command(args, b.notice) default: _, err := l.Submit(telegramSource, message.Text) if err != nil { @@ -396,28 +424,36 @@ func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, update TelegramUp // reply is not lost to a bad entity. It tries again after each failure that a // 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 for { err := b.api.SendMessage(ctx, b.owner, part.text, part.entities) + retry, asked := telegramRetryable(err) + if !retry { + outage.End() + } + if err == nil || ctx.Err() != nil { return } var answer *TelegramError + if len(part.entities) > 0 && errors.As(err, &answer) && answer.Status == http.StatusBadRequest { - b.logger.Printf("entities refused, the message goes again as plain text: %v", err) + b.logger.Warn("reply sent as plain text", "err", err) + part.entities = nil continue } - retry, asked := telegramRetryable(err) if !retry { - b.logger.Printf("message dropped: %v", err) + b.logger.Error("reply lost", "err", err) + return } - wait = b.pause(ctx, err, wait, asked) + wait = b.pause(ctx, outage, err, wait, asked) } } @@ -429,15 +465,22 @@ func (b *TelegramBot) deliver(ctx context.Context, part TelegramPart) { // the model wrote it. It tries again after each failure that a later try can // 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 for { err := b.api.SendRichMessage(ctx, b.owner, markdown) + retry, asked := telegramRetryable(err) + if !retry { + outage.End() + } + if err == nil || ctx.Err() != nil { return } if telegramRejected(err) { - b.logger.Printf("rich message refused, the reply goes with entities: %v", err) + b.logger.Warn("reply sent with simple formatting", "err", err) + for _, part := range telegramReplyParts(telegramUnescapeTags(markdown)) { b.deliver(ctx, part) } @@ -445,13 +488,22 @@ func (b *TelegramBot) deliverRich(ctx context.Context, markdown string) { return } - retry, asked := telegramRetryable(err) if !retry { - b.logger.Printf("message dropped: %v", err) + b.logger.Error("reply lost", "err", err) + return } - wait = b.pause(ctx, err, wait, asked) + wait = b.pause(ctx, outage, err, wait, asked) + } +} + +// 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", } } @@ -463,6 +515,38 @@ 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 +} + +// 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 + } + + o.start = time.Now() + o.logger.Warn(o.loss, "err", err, "retry", delay) +} + +// End logs the end of the outage that runs, with its length. +func (o *TelegramOutage) End() { + if o.start.IsZero() { + return + } + + o.logger.Info(o.back, "after", time.Since(o.start)) + + o.start = time.Time{} +} + // TelegramPart is the text of one message to the chat, at most // telegramMessageUnits long, and its entities. type TelegramPart struct { @@ -488,6 +572,7 @@ func telegramLoadOffset(path string) (int64, error) { } var file TelegramOffsetFile + err = json.Unmarshal(data, &file) if err != nil { return 0, fmt.Errorf("%s: %w", path, err) @@ -519,6 +604,7 @@ func telegramSaveOffset(path string, offset int64) error { // never cuts a character. func telegramSplitMessage(text string, limit int) []string { var parts []string + for { end := telegramPrefixEnd(text, limit) if end == len(text) { @@ -568,6 +654,7 @@ func telegramText(event Event) string { // not cut an entity of telegramEscapeTags. It never cuts a character. func telegramSplitRich(markdown string, limit int) []string { var parts []string + for len(markdown) > limit { cut, fence, opening := telegramRichCut(markdown, limit) if fence == "" { @@ -623,7 +710,9 @@ func telegramPrefixEnd(text string, limit int) int { // so each cut makes the rest shorter. func telegramRichCut(markdown string, limit int) (int, string, string) { blank, line, code := 0, 0, 0 + var fence, opening, inFence, inOpening string + for start := 0; start < limit; { end := strings.IndexByte(markdown[start:], '\n') if end < 0 || start+end+1 > limit { @@ -632,10 +721,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 @@ -682,10 +771,13 @@ func telegramRichCut(markdown string, limit int) (int, string, string) { // An entity that crosses a cut becomes one entity in each part. func telegramSplitEntities(text string, entities []TelegramEntity, limit int) []TelegramPart { var parts []TelegramPart + start := 0 for _, part := range telegramSplitMessage(text, limit) { end := start + telegramUTF16Units(part) + var kept []TelegramEntity + for _, entity := range entities { from := max(entity.Offset, start) to := min(entity.Offset+entity.Length, end)