From 4b063ca25798c78cc03d1d07f67b02c94c95248d Mon Sep 17 00:00:00 2001 From: "Anderson A. Giacomolli" Date: Sat, 7 Nov 2026 10:26:55 -0300 Subject: [PATCH] Talk to the agent through a Telegram bot When telegram.json exists in the data directory, the daemon runs a Telegram bot next to the socket clients. The file holds the bot token and the user ID of the owner, and the daemon refuses it unless its mode is 0600, because the token is a secret. The bot uses getUpdates long polling only, so it needs no webhook and no open port. The bot reads only the text messages of the owner in the private chat and ignores every other update without a log of its content. Each text goes to the agent as a user message from the source telegram, and /stop interrupts the turn. A message that is not text gets the answer that only text is supported. The chat gets the replies as plain text in parts of at most 4096 UTF-16 units, the errors as short messages, and the messages typed in the terminal with a mark. It shows the typing action while a turn runs. Tool calls stay out of the chat, so the phone stays readable and the bot stays far from the rate limit. agent/telegram-offset.json holds the next update ID. It moves past an update only after the message is flushed to the log, so a restart loses no message. A crash between the flush and the offset write repeats that message once, which the user accepts. A failed Bot API call waits and tries again, as long as HTTP 429 asks or from one second up to a minute, and never stops the agent. A failed write of a message or of the offset stops the daemon, because the bot could then lose or repeat messages. Issue: a143bd8 --- README.rst | 48 ++++- botapi_test.go | 20 +- daemon.go | 59 ++++-- daemon_test.go | 9 +- telegram.go | 482 +++++++++++++++++++++++++++++++++++++++++++++++ telegram_test.go | 359 +++++++++++++++++++++++++++++++++++ 6 files changed, 952 insertions(+), 25 deletions(-) create mode 100644 telegram.go create mode 100644 telegram_test.go diff --git a/README.rst b/README.rst index c0c6c5d..4bda34e 100644 --- a/README.rst +++ b/README.rst @@ -3,8 +3,8 @@ ######### OneChat is one AI agent in a single chat that never ends. A daemon owns -the agent and its files, and a terminal client talks to it. It is in -early development. Linux is supported. +the agent and its files, and a terminal client and a Telegram bot talk +to it. It is in early development. Linux is supported. **************** Data directory @@ -80,9 +80,10 @@ listens on the Unix socket ``onechat.sock`` there. In the client, each line is a message to the agent, the line ``/stop`` interrupts the running turn, and Ctrl-D ends the client. The client shows the replies, the tool calls, and the first lines of each tool result as they happen. -The daemon is the only process that uses the token file while it runs, -so ``onechat login`` and ``onechat probe`` refuse to start until it -stops. +It also shows each message that the user sends from another client, with +the client as a mark, such as ``[telegram] text``. The daemon is the +only process that uses the token file while it runs, so ``onechat +login`` and ``onechat probe`` refuse to start until it stops. The agent runs commands with bash as the user in its working directory, with no sandbox. It has the rights of the user, also on the token file. @@ -106,6 +107,43 @@ prompt cache key comes from it. how their files are organized, and how they want work done. The prompt ends with them, and a change applies at the next call. +********** + Telegram +********** + +The daemon also connects the agent to a Telegram bot when +``telegram.json`` exists in the data directory. It polls the Bot API for +new messages, so it needs no open port. + +#. Make a bot with BotFather and keep its token. + +#. Find your numeric Telegram user ID. + +#. Write the token and the ID to ``telegram.json``, and give the file + mode 0600, because the token is a secret. The daemon refuses the file + with any other mode. + +#. Start the daemon, open a private chat with the bot, and write. + +.. code:: json + + {"token": "123456:ABC-token-from-BotFather", "owner": 123456789} + +The bot reads only the private chat of the owner and ignores all other +updates. Each text message goes to the agent as a user message, and the +line ``/stop`` interrupts the running turn. The bot sends the replies as +plain text, in parts of at most 4096 characters, and each error as a +short message. It shows the typing action while a turn runs. It does not +send the tool calls. It sends each message that the user types in the +terminal client, with the mark ``[terminal]``. + +``agent/telegram-offset.json`` holds the ID of the next update. The bot +moves it past an update only after the message is durable in the log, so +a restart of the daemon does not lose a message. A crash just after the +write of a message can make the message repeat once. A network failure +or a rate limit makes the bot wait and try again, and does not stop the +daemon. + ******* Probe ******* diff --git a/botapi_test.go b/botapi_test.go index 565f27d..b571c6a 100644 --- a/botapi_test.go +++ b/botapi_test.go @@ -141,7 +141,9 @@ func TestBotErrorsDoNotHoldTheToken(t *testing.T) { // fakeBot is a fake Bot API server for testBotToken. getUpdates returns the // added updates from the offset on, or waits a short time when there are none. -// The other methods succeed, unless fail gives an error answer for a call. +// The other methods succeed, unless fail gives an error answer for a call. An +// answer with HTTP 429 asks for a wait of one second. Fail runs with the mutex +// held. type fakeBot struct { *httptest.Server @@ -226,18 +228,22 @@ func (f *fakeBot) serve(w http.ResponseWriter, r *http.Request) { } f.mutex.Lock() - fail := f.fail - f.mutex.Unlock() - status, description := 0, "" - if fail != nil { - status, description = fail(call) + if f.fail != nil { + status, description = f.fail(call) } + f.mutex.Unlock() + if status != 0 { f.record(call) w.WriteHeader(status) - _, _ = fmt.Fprintf(w, `{"ok":false,"error_code":%d,"description":%q,"parameters":{"retry_after":1}}`, status, description) + wait := 0 + if status == http.StatusTooManyRequests { + wait = 1 + } + + _, _ = fmt.Fprintf(w, `{"ok":false,"error_code":%d,"description":%q,"parameters":{"retry_after":%d}}`, status, description, wait) return } diff --git a/daemon.go b/daemon.go index 68073d0..b37e0cf 100644 --- a/daemon.go +++ b/daemon.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "log" "net" "os" "path/filepath" @@ -51,9 +52,9 @@ type clientRequest struct { } // daemon runs "onechat daemon": it owns the root agent and its files and serves -// the clients on the Unix socket in the data directory until ctx ends. It holds -// the lock of the token file, so it is the only process that refreshes the -// tokens. +// the clients on the Unix socket in the data directory until ctx ends. When +// telegram.json exists, it also runs the Telegram bot. It holds the lock of the +// token file, so it is the only process that refreshes the tokens. func daemon(ctx context.Context, out io.Writer) (err error) { dir, err := dataDir() if err != nil { @@ -73,6 +74,11 @@ func daemon(ctx context.Context, out io.Writer) (err error) { return err } + botSettings, hasBot, err := loadBotConfig(dir) + if err != nil { + return err + } + tokens, err := lockTokens(dir) if err != nil { return err @@ -112,14 +118,24 @@ func daemon(ctx context.Context, out io.Writer) (err error) { return errors.Join(err, listener.Close()) } - return serveAgent(ctx, listener, func(notify func(Event)) (*loop, error) { + var telegram *telegramBot + if hasBot { + telegram = newTelegramBot(newBotAPI(telegramAPI, botSettings.Token), botSettings.Owner, filepath.Join(agent.dir, offsetFileName), log.New(out, "telegram: ", log.LstdFlags)) + _, err = fmt.Fprintln(out, "onechat: the Telegram bot waits for messages of its owner") + if err != nil { + return errors.Join(err, listener.Close()) + } + } + + return serveAgent(ctx, listener, telegram, func(notify func(Event)) (*loop, error) { return newLoop(agent, memory, client, time.Now, notify) }) } // serveAgent runs the loop that start makes and serves the clients of -// listener until ctx ends or the loop fails. It closes listener. -func serveAgent(ctx context.Context, listener net.Listener, start func(notify func(Event)) (*loop, error)) error { +// listener until ctx ends or the loop fails. It closes listener. When bot is +// not nil, it runs the bot too, and a failure of the bot stops the agent. +func serveAgent(ctx context.Context, listener net.Listener, bot *telegramBot, start func(notify func(Event)) (*loop, error)) error { ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -127,7 +143,15 @@ func serveAgent(ctx context.Context, listener net.Listener, start func(notify fu clients: make(map[*hubClient]bool), } - l, err := start(hub.Broadcast) + notify := hub.Broadcast + if bot != nil { + notify = func(event Event) { + hub.Broadcast(event) + bot.Notify(event) + } + } + + l, err := start(notify) if err != nil { return errors.Join(err, listener.Close()) } @@ -140,17 +164,28 @@ func serveAgent(ctx context.Context, listener net.Listener, start func(notify fu hub.Accept(ctx, listener, l) }() + var botErr error + if bot != nil { + workers.Add(1) + go func() { + defer workers.Done() + + botErr = bot.Run(ctx, l) + cancel() + }() + } + err = l.Run(ctx) + if errors.Is(err, context.Canceled) { + err = nil + } + cancel() err = errors.Join(err, listener.Close()) hub.CloseAll() workers.Wait() - if errors.Is(err, context.Canceled) { - return nil - } - - return err + return errors.Join(err, botErr) } // hub holds the connected clients and sends each event to all of them. diff --git a/daemon_test.go b/daemon_test.go index 00680e6..47fb9a5 100644 --- a/daemon_test.go +++ b/daemon_test.go @@ -242,6 +242,13 @@ func (b *syncBuffer) String() string { func startTestDaemon(t *testing.T, server *fakeResponses) string { t.Helper() + return startTestDaemonWithBot(t, server, nil) +} + +// startTestDaemonWithBot is startTestDaemon with bot, which can be nil. +func startTestDaemonWithBot(t *testing.T, server *fakeResponses, bot *telegramBot) string { + t.Helper() + dir := t.TempDir() agent, err := openAgent(filepath.Join(dir, agentDirName)) require.NoError(t, err) @@ -259,7 +266,7 @@ func startTestDaemon(t *testing.T, server *fakeResponses) string { ctx, cancel := context.WithCancel(t.Context()) done := make(chan error) go func() { - done <- serveAgent(ctx, listener, func(notify func(Event)) (*loop, error) { + done <- serveAgent(ctx, listener, bot, func(notify func(Event)) (*loop, error) { return newLoop(agent, memory, client, clock.Now, notify) }) }() diff --git a/telegram.go b/telegram.go new file mode 100644 index 0000000..dabf2e8 --- /dev/null +++ b/telegram.go @@ -0,0 +1,482 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "log" + "os" + "path/filepath" + "strings" + "sync" + "time" +) + +const ( + // telegramAPI is the Telegram Bot API service. + telegramAPI = "https://api.telegram.org" + + // botConfigFileName is the bot config in the data directory. The daemon + // runs the bot only when the file exists. + botConfigFileName = "telegram.json" + + // offsetFileName is the update offset of the bot in the agent directory. + offsetFileName = "telegram-offset.json" + + // telegramSource is the source of the user messages from Telegram. + telegramSource = "telegram" + + // messageUnits is the most UTF-16 code units of one Telegram message. + messageUnits = 4096 + + // typingInterval is how often the bot shows the typing action while a + // turn runs. Telegram shows the action for about five seconds. + typingInterval = 4 * time.Second + + // textOnly is the answer to a message that is not text. + textOnly = "Only text messages are supported." +) + +// botConfig 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 botConfig 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 each text message of the owner to the +// agent. It ignores every other update. Its sender sends the replies, the +// errors, and the user messages of the other clients to the chat as plain text, +// and shows the typing action while a turn runs. 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. +type telegramBot struct { + api *botAPI + owner int64 + offsetPath string + logger *log.Logger + minWait time.Duration + maxWait time.Duration + + mutex sync.Mutex + queue []Event + ready chan struct{} +} + +// offsetFile is the content of the offset file. +type offsetFile struct { + Offset int64 `json:"offset"` +} + +// loadBotConfig reads telegram.json in dir. It returns false when the file +// does not exist. An error never holds the token. +func loadBotConfig(dir string) (botConfig, bool, error) { + path := filepath.Join(dir, botConfigFileName) + info, err := os.Stat(path) + if errors.Is(err, os.ErrNotExist) { + return botConfig{}, false, nil + } + + if err != nil { + return botConfig{}, false, fmt.Errorf("read bot config: %w", err) + } + + if info.Mode().Perm()&0o077 != 0 { + return botConfig{}, false, fmt.Errorf("%s holds the bot token, so its mode must be 0600", botConfigFileName) + } + + data, err := os.ReadFile(path) + if err != nil { + return botConfig{}, false, fmt.Errorf("read bot config: %w", err) + } + + var config botConfig + 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") + } + + if err != nil { + return botConfig{}, false, fmt.Errorf("%s is not valid JSON with the fields token and owner", botConfigFileName) + } + + err = config.validate() + if err != nil { + return botConfig{}, false, fmt.Errorf("%s: %w", botConfigFileName, err) + } + + return config, true, nil +} + +// 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. +func (c botConfig) validate() error { + id, secret, found := strings.Cut(c.Token, ":") + if !found || id == "" || secret == "" || strings.Trim(id, "0123456789") != "" || strings.Trim(secret, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ_-") != "" { + return errors.New("token does not have the form of a bot token from BotFather") + } + + if c.Owner <= 0 { + return errors.New("owner must be the numeric Telegram user ID of the owner") + } + + return nil +} + +// 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 *botAPI, owner int64, offsetPath string, logger *log.Logger) *telegramBot { + return &telegramBot{ + api: api, + owner: owner, + offsetPath: offsetPath, + logger: logger, + minWait: time.Second, + maxWait: time.Minute, + ready: make(chan struct{}, 1), + } +} + +// Notify queues event for the chat. It never blocks, so a slow network does not +// stop the agent. +func (b *telegramBot) Notify(event Event) { + switch event.Kind { + case "busy", "idle", "reply", "error": + 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 +// cannot be written, because then the bot would lose or repeat messages. +func (b *telegramBot) Run(ctx context.Context, l *loop) 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) + cancel() + sender.Wait() + + return err +} + +// send sends each queued event to the chat in order until ctx ends. While a +// turn runs, it shows the typing action every typingInterval and after each +// message, because a message ends the action. +func (b *telegramBot) send(ctx context.Context) { + ticker := time.NewTicker(typingInterval) + defer ticker.Stop() + + busy := false + for ctx.Err() == nil { + event, found := b.next() + if !found { + select { + case <-b.ready: + case <-ticker.C: + if busy { + b.typing(ctx) + } + case <-ctx.Done(): + } + + continue + } + + switch event.Kind { + case "busy": + busy = true + b.typing(ctx) + ticker.Reset(typingInterval) + case "idle": + busy = false + default: + for _, part := range splitMessage(botText(event), messageUnits) { + b.deliver(ctx, part) + } + + if busy { + b.typing(ctx) + } + } + } +} + +// next removes the first queued event and returns it. +func (b *telegramBot) next() (Event, bool) { + b.mutex.Lock() + defer b.mutex.Unlock() + + if len(b.queue) == 0 { + return Event{}, false + } + + event := b.queue[0] + b.queue = b.queue[1:] + + return event, true +} + +// 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) +} + +// deliver sends text to the chat. It tries again after each failure that a +// later try can fix, until ctx ends. It drops the text after any other failure. +func (b *telegramBot) deliver(ctx context.Context, text string) { + wait := b.minWait + for { + err := b.api.SendMessage(ctx, b.owner, text) + if err == nil || ctx.Err() != nil { + return + } + + retry, asked := retryable(err) + if !retry { + b.logger.Printf("message dropped: %v", err) + return + } + + wait = b.pause(ctx, err, wait, asked) + } +} + +// 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 { + delay := max(wait, asked) + b.logger.Printf("%v (next try in %s)", err, delay) + + timer := time.NewTimer(delay) + defer timer.Stop() + + select { + case <-timer.C: + case <-ctx.Done(): + } + + return min(2*wait, b.maxWait) +} + +// 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 *loop) error { + offset, err := loadOffset(b.offsetPath) + if err != nil { + return err + } + + wait := b.minWait + for ctx.Err() == nil { + updates, err := b.api.GetUpdates(ctx, offset) + if ctx.Err() != nil { + return nil + } + + if err != nil { + _, asked := retryable(err) + wait = b.pause(ctx, err, wait, asked) + + continue + } + + wait = b.minWait + for _, update := range updates { + if ctx.Err() != nil { + return nil + } + + err = b.handle(context.WithoutCancel(ctx), l, update) + if err != nil { + return err + } + + offset = update.UpdateID + 1 + err = saveOffset(b.offsetPath, offset) + if err != nil { + return err + } + } + } + + return nil +} + +// 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 +// that only text is supported. Every other update is ignored, and its content +// is not logged. +func (b *telegramBot) handle(ctx context.Context, l *loop, update botUpdate) 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") + return nil + } + + switch { + case message.Text == "": + b.notice(textOnly) + case strings.TrimSpace(message.Text) == stopCommand: + if !l.Interrupt() { + b.notice("No turn runs.") + } + default: + _, err := l.Submit(telegramSource, message.Text) + if err != nil { + return err + } + + err = l.Flush(ctx) + if err != nil { + return fmt.Errorf("flush the message from Telegram: %w", err) + } + } + + return nil +} + +// notice queues a short answer of the bot for the chat. +func (b *telegramBot) notice(text string) { + b.Notify(Event{ + Kind: "reply", + Text: text, + }) +} + +// loadOffset reads the update offset at path. A missing file gives 0, which +// asks for every update that Telegram keeps. +func loadOffset(path string) (int64, error) { + data, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return 0, nil + } + + if err != nil { + return 0, fmt.Errorf("read update offset: %w", err) + } + + var file offsetFile + err = json.Unmarshal(data, &file) + if err != nil { + return 0, fmt.Errorf("%s: %w", path, err) + } + + return file.Offset, nil +} + +// saveOffset replaces the update offset at path. +func saveOffset(path string, offset int64) error { + data, err := json.Marshal(offsetFile{ + Offset: offset, + }) + if err != nil { + return fmt.Errorf("encode update offset: %w", err) + } + + err = writeFileAtomic(path, append(data, '\n')) + if err != nil { + return fmt.Errorf("write update offset: %w", err) + } + + return nil +} + +// botText returns the text of event for the chat. A user message of another +// client has its source as a mark. +func botText(event Event) string { + switch event.Kind { + case "user": + return "[" + event.Source + "] " + event.Text + case "error": + return "Error: " + event.Text + default: + return event.Text + } +} + +// splitMessage cuts text into parts of at most limit UTF-16 code units, the +// unit of the Telegram message limit. It cuts after the last line break or +// space of a part when one is in its second half, and else at the limit. It +// never cuts a character. +func splitMessage(text string, limit int) []string { + var parts []string + for { + end := prefixEnd(text, limit) + if end == len(text) { + return append(parts, text) + } + + cut := strings.LastIndexByte(text[:end], '\n') + 1 + if cut <= end/2 { + cut = strings.LastIndexByte(text[:end], ' ') + 1 + } + + if cut <= end/2 { + cut = end + } + + parts = append(parts, text[:cut]) + text = text[cut:] + } +} + +// prefixEnd returns the byte length of the longest prefix of text with at most +// limit UTF-16 code units. +func prefixEnd(text string, limit int) int { + units := 0 + for index, char := range text { + size := 1 + if char > 0xFFFF { + size = 2 + } + + if units+size > limit { + return index + } + + units += size + } + + return len(text) +} diff --git a/telegram_test.go b/telegram_test.go new file mode 100644 index 0000000..d18e794 --- /dev/null +++ b/telegram_test.go @@ -0,0 +1,359 @@ +package main + +import ( + "log" + "net/http" + "os" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestOwnerTalksToTheAgentThroughTelegram(t *testing.T) { + server := newFakeResponses(t) + server.reply = func(w http.ResponseWriter, call fakeCall) { + writeEvents(w, completedEvent([]any{messageItem("Hello from the agent.")}, 100, 0, 5)) + } + + telegram := newFakeBot(t) + bot, _ := startTestBot(t, telegram) + socket := startTestDaemonWithBot(t, server, bot) + terminal := startTestClient(t, socket) + terminal.WaitFor(t, "[connected to OneChat]") + + telegram.Add(botUpdate{ + UpdateID: 1, + Message: ownerMessage("hi from the phone"), + }) + + terminal.WaitFor(t, "[telegram] hi from the phone\n[working]\nHello from the agent.\n[done]") + waitForSent(t, telegram, "Hello from the agent.") + require.True(t, slices.ContainsFunc(telegram.Calls(), func(call fakeBotCall) bool { + return call.method == "sendChatAction" && call.params["action"] == "typing" + })) + + terminal.Type("hi from the desk") + require.Equal(t, []string{"Hello from the agent.", "[terminal] hi from the desk", "Hello from the agent."}, waitForSent(t, telegram, "Hello from the agent.", "[terminal] hi from the desk", "Hello from the agent.")) +} + +func TestBotIgnoresEveryoneButTheOwner(t *testing.T) { + telegram := newFakeBot(t) + bot, logs := startTestBot(t, telegram) + socket := startTestDaemonWithBot(t, newFakeResponses(t), bot) + terminal := startTestClient(t, socket) + terminal.WaitFor(t, "[connected to OneChat]") + + group := ownerMessage("secret in a group") + group.Chat = botChat{ + ID: -100, + Type: "group", + } + + stranger := ownerMessage("secret of a stranger") + stranger.Chat.ID = 77 + stranger.From.ID = 77 + + unsigned := ownerMessage("secret with no sender") + unsigned.From = nil + + for id, message := range []*botMessage{group, stranger, unsigned, nil} { + telegram.Add(botUpdate{ + UpdateID: int64(10 + id), + Message: message, + }) + } + + require.Eventually(t, func() bool { + offset, err := loadOffset(bot.offsetPath) + return err == nil && offset == 14 + }, 10*time.Second, time.Millisecond) + + require.Empty(t, telegram.Sent()) + require.NotContains(t, terminal.Output(), "secret") + require.NotContains(t, logs.String(), "secret") + require.Contains(t, logs.String(), "update ignored") +} + +func TestBotAnswersThatOnlyTextIsSupported(t *testing.T) { + telegram := newFakeBot(t) + bot, _ := startTestBot(t, telegram) + startTestDaemonWithBot(t, newFakeResponses(t), bot) + + telegram.Add(botUpdate{ + UpdateID: 1, + Message: ownerMessage(""), + }) + + waitForSent(t, telegram, textOnly) +} + +func TestStopFromTelegramInterruptsTheTurn(t *testing.T) { + server := newFakeResponses(t) + arrived := make(chan struct{}) + release := make(chan struct{}) + t.Cleanup(func() { + close(release) + }) + + server.reply = func(w http.ResponseWriter, call fakeCall) { + close(arrived) + <-release + } + + telegram := newFakeBot(t) + bot, _ := startTestBot(t, telegram) + startTestDaemonWithBot(t, server, bot) + + telegram.Add(botUpdate{ + UpdateID: 1, + Message: ownerMessage("a long task"), + }) + + <-arrived + telegram.Add(botUpdate{ + UpdateID: 2, + Message: ownerMessage(stopCommand), + }) + + waitForSent(t, telegram, "Error: interrupted") + + telegram.Add(botUpdate{ + UpdateID: 3, + Message: ownerMessage(stopCommand), + }) + + waitForSent(t, telegram, "Error: interrupted", "No turn runs.") +} + +func TestBotStartsAtTheStoredOffset(t *testing.T) { + server := newFakeResponses(t) + server.reply = func(w http.ResponseWriter, call fakeCall) { + writeEvents(w, completedEvent([]any{messageItem("ok")}, 100, 0, 5)) + } + + telegram := newFakeBot(t) + bot, _ := startTestBot(t, telegram) + require.NoError(t, saveOffset(bot.offsetPath, 10)) + + telegram.Add(botUpdate{ + UpdateID: 9, + Message: ownerMessage("handled before the restart"), + }) + + telegram.Add(botUpdate{ + UpdateID: 10, + Message: ownerMessage("new after the restart"), + }) + + socket := startTestDaemonWithBot(t, server, bot) + terminal := startTestClient(t, socket) + terminal.WaitFor(t, "[connected to OneChat]") + waitForSent(t, telegram, "ok") + + require.Equal(t, float64(10), telegram.Calls()[0].params["offset"]) + require.NotContains(t, terminal.Output(), "handled before the restart") + + require.Eventually(t, func() bool { + offset, err := loadOffset(bot.offsetPath) + return err == nil && offset == 11 + }, 10*time.Second, time.Millisecond) +} + +func TestBotTriesAgainAfterFailures(t *testing.T) { + server := newFakeResponses(t) + server.reply = func(w http.ResponseWriter, call fakeCall) { + writeEvents(w, completedEvent([]any{messageItem("ok")}, 100, 0, 5)) + } + + telegram := newFakeBot(t) + failures := map[string]int{ + "getUpdates": 2, + "sendMessage": 1, + } + + telegram.fail = func(call fakeBotCall) (int, string) { + if failures[call.method] == 0 { + return 0, "" + } + + failures[call.method]-- + if call.method == "sendMessage" { + return http.StatusTooManyRequests, "Too Many Requests: retry after 1" + } + + return http.StatusBadGateway, "Bad Gateway" + } + + bot, logs := startTestBot(t, telegram) + startTestDaemonWithBot(t, server, bot) + + telegram.Add(botUpdate{ + UpdateID: 1, + Message: ownerMessage("hello"), + }) + + waitForSent(t, telegram, "ok") + require.Contains(t, logs.String(), "answered 502") + require.Contains(t, logs.String(), "answered 429") + require.NotContains(t, logs.String(), testBotToken) +} + +func TestBotLogsNoTokenWhenTheNetworkFails(t *testing.T) { + telegram := newFakeBot(t) + telegram.Close() + + bot, logs := startTestBot(t, telegram) + startTestDaemonWithBot(t, newFakeResponses(t), bot) + + require.Eventually(t, func() bool { + return strings.Count(logs.String(), "no valid answer") >= 2 + }, 10*time.Second, time.Millisecond) + require.NotContains(t, logs.String(), testBotToken) +} + +func TestSplitMessage(t *testing.T) { + testCases := []struct { + name string + text string + limit int + parts []string + }{ + { + name: "short text stays whole", + text: "one two", + limit: 10, + parts: []string{"one two"}, + }, + { + name: "cut after the last line break", + text: "first line\nsecond line", + limit: 16, + parts: []string{"first line\n", "second line"}, + }, + { + name: "cut after the last space without a line break", + text: "aaaa bbbb cccc", + limit: 12, + parts: []string{"aaaa bbbb ", "cccc"}, + }, + { + name: "hard cut of one long word", + text: "abcdefghij", + limit: 4, + parts: []string{"abcd", "efgh", "ij"}, + }, + { + name: "a character outside the BMP counts two units and is not cut", + text: "ab\U0001F600cd", + limit: 3, + parts: []string{"ab", "\U0001F600c", "d"}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + require.Equal(t, tc.parts, splitMessage(tc.text, tc.limit)) + }) + } + + long := strings.Repeat("word ", 2000) + parts := splitMessage(long, messageUnits) + require.Len(t, parts, 3) + require.Equal(t, long, strings.Join(parts, "")) +} + +func TestLoadBotConfig(t *testing.T) { + testCases := []struct { + name string + content string + mode os.FileMode + errorText string + }{ + { + name: "owner file is read", + content: `{"token": "` + testBotToken + `", "owner": 4242}`, + mode: 0o600, + }, + { + name: "file that others can read is refused", + content: `{"token": "` + testBotToken + `", "owner": 4242}`, + mode: 0o640, + errorText: "mode must be 0600", + }, + { + name: "token with a path character is refused", + content: `{"token": "123:abc/../x", "owner": 4242}`, + mode: 0o600, + errorText: "form of a bot token", + }, + { + name: "missing owner is refused", + content: `{"token": "` + testBotToken + `"}`, + mode: 0o600, + errorText: "numeric Telegram user ID", + }, + { + name: "unknown field is refused", + content: `{"token": "` + testBotToken + `", "owner": 4242, "chat": 1}`, + mode: 0o600, + errorText: "not valid JSON", + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, botConfigFileName) + require.NoError(t, os.WriteFile(path, []byte(tc.content), tc.mode)) + require.NoError(t, os.Chmod(path, tc.mode)) + + config, found, err := loadBotConfig(dir) + if tc.errorText != "" { + require.ErrorContains(t, err, tc.errorText) + require.NotContains(t, err.Error(), testBotToken) + + return + } + + require.NoError(t, err) + require.True(t, found) + require.Equal(t, botConfig{ + Token: testBotToken, + Owner: testOwner, + }, config) + }) + } + + _, found, err := loadBotConfig(t.TempDir()) + require.NoError(t, err) + require.False(t, found) +} + +// startTestBot returns a bot of testOwner on server, with its offset file in a +// temporary directory and short waits, and the buffer of its log. +func startTestBot(t *testing.T, server *fakeBot) (*telegramBot, *syncBuffer) { + t.Helper() + + logs := &syncBuffer{} + bot := newTelegramBot(newBotAPI(server.URL, testBotToken), testOwner, filepath.Join(t.TempDir(), offsetFileName), log.New(logs, "", 0)) + bot.minWait = time.Millisecond + bot.maxWait = 10 * time.Millisecond + + return bot, logs +} + +// waitForSent waits until the bot sent the texts in order, and returns them. +func waitForSent(t *testing.T, server *fakeBot, texts ...string) []string { + t.Helper() + + require.Eventually(t, func() bool { + sent := server.Sent() + return len(sent) >= len(texts) && slices.Equal(sent[len(sent)-len(texts):], texts) + }, 10*time.Second, time.Millisecond, "sent so far: %q", server.Sent()) + + return server.Sent() +} -- 2.53.0