X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/6d96a69d33eb634f61f809001aca19ae57113180..3b7a3676dc3614ee3099e8798b5d0c07cb116b93:/telegram.go diff --git a/telegram.go b/telegram.go index 13196d4..2a9345a 100644 --- a/telegram.go +++ b/telegram.go @@ -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") @@ -145,7 +147,9 @@ 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 @@ -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": + 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 } @@ -311,8 +333,9 @@ func (b *TelegramBot) typing(ctx context.Context) { } // post sends event to the chat. A reply goes as rich messages of its Markdown, -// split with telegramSplitRich. Every other event goes as plain text, split -// with telegramSplitMessage. +// with each < and & outside code escaped by telegramEscapeTags and then split +// with telegramSplitRich, so that the length limit counts the escaped text. +// Every other event goes as plain text, split with telegramSplitMessage. func (b *TelegramBot) post(ctx context.Context, event Event) { if event.Kind != "reply" { for _, text := range telegramSplitMessage(telegramText(event), telegramMessageUnits) { @@ -324,16 +347,16 @@ func (b *TelegramBot) post(ctx context.Context, event Event) { return } - for _, markdown := range telegramSplitRich(event.Text, telegramRichBytes) { + for _, markdown := range telegramSplitRich(telegramEscapeTags(event.Text), telegramRichBytes) { b.deliverRich(ctx, markdown) } } -// 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() @@ -348,16 +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. Any other message of the owner gets the answer +// /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, update TelegramUpdate) error { +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.Print("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) @@ -365,6 +393,17 @@ func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, update TelegramUp if !l.Interrupt() { b.notice("No turn runs.") } + case stop: + found, err := l.StopSubagent(name) + if err != nil { + return fmt.Errorf("stop subagent %s: %w", name, err) + } + + 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 { @@ -385,60 +424,86 @@ 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.Printf("reply sent as plain text: %v", err) + part.entities = nil continue } - retry, asked := telegramRetryable(err) if !retry { - b.logger.Printf("message dropped: %v", err) + b.logger.Printf("reply lost: %v", err) + return } - wait = b.pause(ctx, err, wait, asked) + wait = b.pause(ctx, outage, err, wait, asked) } } -// deliverRich sends markdown to the chat as a rich message. When the Bot API -// refuses it with HTTP 400 or 404, the message was not sent, so deliverRich -// sends the same Markdown once with message entities, and the reply is not -// lost and does not repeat. It tries again after each failure that a later try -// can fix, until ctx ends. It drops the message after any other failure. +// deliverRich sends markdown, escaped with telegramEscapeTags, to the chat as a +// rich message. When the Bot API refuses it with HTTP 400 or 404, the message +// was not sent, so deliverRich sends the text of the part once with message +// entities, and the reply is not lost and does not repeat. Message entities +// read no HTML, so the fallback gets the Markdown of telegramUnescapeTags, as +// 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) - for _, part := range telegramReplyParts(markdown) { + b.logger.Printf("reply sent with simple formatting: %v", err) + + for _, part := range telegramReplyParts(telegramUnescapeTags(markdown)) { b.deliver(ctx, part) } return } - retry, asked := telegramRetryable(err) if !retry { - b.logger.Printf("message dropped: %v", err) + b.logger.Printf("reply lost: %v", 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", } } @@ -450,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 *log.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.Printf("%s: %v (retry in %s)", o.loss, err, durationText(delay)) +} + +// End logs the end of the outage that runs, with its length. +func (o *TelegramOutage) End() { + if o.start.IsZero() { + return + } + + o.logger.Printf("%s after %s", o.back, durationText(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 { @@ -475,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) @@ -506,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) { @@ -527,11 +626,17 @@ func telegramSplitMessage(text string, limit int) []string { } // telegramText returns the text of event for the chat. A user message of -// another client has its source as a mark. +// another client has its source as a mark. A subagent shows only its start and +// how its run ended, because the reply of the agent carries the content of the +// report. func telegramText(event Event) string { switch event.Kind { case "user": return "[" + event.Source + "] " + event.Text + case "subagent": + return "[" + event.Text + "] started" + case "work": + return "[" + event.Source + "] " + event.Status case "error": return "Error: " + event.Text default: @@ -545,9 +650,11 @@ func telegramText(event Event) string { // after the last line break outside a code block, then after the last line // break in a code block, which it closes at the end of the part and opens again // in the next part with the same fence line. A line longer than limit is cut at -// the limit. It never cuts a character. +// the limit, or before an & in the four bytes before the limit, so that it does +// 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 == "" { @@ -603,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 { @@ -648,6 +757,12 @@ func telegramRichCut(markdown string, limit int) (int, string, string) { cut-- } + start := max(cut-len("&")+1, 0) + entity := strings.LastIndexByte(markdown[start:cut], '&') + if entity >= 0 && start+entity > 0 { + cut = start + entity + } + return cut, "", "" } @@ -656,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)