From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 18:01:07 +0000 (-0300) Subject: Keep time limits with their operations X-Git-Url: https://git.giacomolli.uk/onechat/commitdiff_plain/48d7bce28b18ecef4c52c98821675605464cecb9 Keep time limits with their operations Operation files own their duration constants without struct fields that only supply shorter test waits. Fake time checks the full send, lookup, batch, catch-up, notice, and schedule waits. Retry delays and batch policy values stay the same. --- diff --git a/schedule.go b/schedule.go index b3a65ea..a1eba1c 100644 --- a/schedule.go +++ b/schedule.go @@ -58,7 +58,6 @@ func openSchedules(path, owner string, clock func() time.Time, logger *slog.Logg owner: owner, clock: clock, logger: logger, - check: scheduleCheck, changed: make(chan struct{}, 1), } @@ -118,7 +117,6 @@ type Schedules struct { owner string clock func() time.Time logger *slog.Logger - check time.Duration changed chan struct{} mutex sync.Mutex @@ -126,9 +124,9 @@ type Schedules struct { } // Run starts the runs of the due schedules in l until ctx ends. It waits until -// the first due time, at most check at a time, and also wakes at a set or a -// remove. So a schedule that came due while the daemon was stopped runs at the -// start. It returns the error of a write of the log or of the file. +// the first due time, at most scheduleCheck at a time, and also wakes at a set +// or a remove. So a schedule that came due while the daemon was stopped runs at +// the start. It returns the error of a write of the log or of the file. func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error { for { err := s.deliver(ctx, l) @@ -143,7 +141,7 @@ func (s *Schedules) Run(ctx context.Context, l *TurnLoop) error { wait, found := s.Wait(s.clock()) if found { - timer = time.NewTimer(min(wait, s.check)) + timer = time.NewTimer(min(wait, scheduleCheck)) timeout = timer.C } diff --git a/schedule_test.go b/schedule_test.go index 6610c26..5743a22 100644 --- a/schedule_test.go +++ b/schedule_test.go @@ -7,6 +7,7 @@ import ( "strings" "sync" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/require" @@ -251,6 +252,9 @@ func TestSchedulesRunStartsATurnWhenDue(t *testing.T) { require.NoError(t, err) clock.Set(clock.Now().Add(5 * time.Minute)) + + schedules.changed <- struct{}{} + test.events.WaitFor(t, "idle", 1) require.Equal(t, 1, server.Calls()) @@ -295,7 +299,6 @@ func TestDaemonRunsADueSchedule(t *testing.T) { schedules, err := openSchedules(filepath.Join(t.TempDir(), scheduleFileName), "OneChat", clock.Now, discardLog()) require.NoError(t, err) - schedules.check = time.Millisecond whatsapp, _, _ := whatsappStartTest(t, t.TempDir(), newTestClock()) socket, _ := daemonStartTestWithSchedules(t, newModelFake(t), nil, whatsapp, schedules, clock.Now) client := terminalStartTestClient(t, socket) @@ -305,9 +308,43 @@ func TestDaemonRunsADueSchedule(t *testing.T) { require.NoError(t, err) clock.Set(clock.Now().Add(5 * time.Minute)) + + schedules.changed <- struct{}{} + client.WaitFor(t, "[Schedule 1] Sunday 2026-11-08 20:05 -03:00, once\nRemind the user to drink water.\n"+scheduleEmptyReply+"\n[working]\na line\n[done]") } +func TestSchedulesCheckTheClockAfterItsWaitLimit(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + whatsapp, _, _ := whatsappStartTest(t, t.TempDir(), newTestClock()) + loop, events := newTestIdleTurnLoop(t, whatsapp) + clock := &ScheduleTestClock{now: time.Now()} + schedules, err := openSchedules(filepath.Join(t.TempDir(), scheduleFileName), "OneChat", clock.Now, discardLog()) + require.NoError(t, err) + + _, err = schedules.Set("Check the clock.", new(clock.Now().Add(5*time.Minute).Format(scheduleAtLayout)), nil) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + done := make(chan error, 1) + go func() { + done <- schedules.Run(ctx, loop) + }() + + synctest.Wait() + require.Zero(t, loop.memory.Len()) + clock.Set(clock.Now().Add(5 * time.Minute)) + time.Sleep(scheduleCheck) + synctest.Wait() + require.Equal(t, 1, loop.memory.Len()) + require.Contains(t, events.Last("work"), "Check the clock.") + cancel() + require.NoError(t, <-done) + }) +} + func TestToolboxRunsTheScheduleTools(t *testing.T) { schedules, _ := scheduleTestOpen(t) box := &Toolbox{ @@ -396,11 +433,10 @@ func scheduleTestOpen(t *testing.T) (*Schedules, string) { } // scheduleTestRun runs schedules in a new turn loop that calls server, until -// the test ends. Run looks at the times every millisecond. +// the test ends. func scheduleTestRun(t *testing.T, schedules *Schedules, server *ModelFake) *TurnTestLoop { t.Helper() - schedules.check = time.Millisecond test := newTestTurnLoop(t, server) ctx, cancel := context.WithCancel(t.Context()) done := make(chan error) diff --git a/telegram.go b/telegram.go index a595b68..f3b3bb2 100644 --- a/telegram.go +++ b/telegram.go @@ -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" @@ -103,8 +111,6 @@ func newTelegramBot(api *TelegramAPI, owner int64, offsetPath string, logger *sl owner: owner, offsetPath: offsetPath, logger: logger, - minWait: time.Second, - maxWait: time.Minute, ready: make(chan struct{}, 1), } } @@ -146,17 +152,15 @@ func (c TelegramConfig) validate() error { // 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 *slog.Logger - minWait time.Duration - maxWait time.Duration mutex sync.Mutex queue []Event @@ -279,8 +283,8 @@ func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp) back: "image receive back", } - imageWait := b.minWait - wait := b.minWait + imageWait := telegramRetryDelay + wait := telegramRetryDelay poll: for ctx.Err() == nil { @@ -298,7 +302,7 @@ poll: outage.End() - wait = b.minWait + wait = telegramRetryDelay for _, update := range updates { if ctx.Err() != nil { @@ -318,7 +322,7 @@ poll: imageOutage.End() - imageWait = b.minWait + imageWait = telegramRetryDelay offset = update.UpdateID + 1 err = telegramSaveOffset(b.offsetPath, offset) @@ -373,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) @@ -386,7 +391,7 @@ 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 @@ -480,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) @@ -521,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) diff --git a/telegram_api.go b/telegram_api.go index 32ad9a9..a04b2fa 100644 --- a/telegram_api.go +++ b/telegram_api.go @@ -41,10 +41,9 @@ const ( // newTelegramAPI returns the client of the Bot API at base with token. func newTelegramAPI(base, token string) *TelegramAPI { return &TelegramAPI{ - base: base, - token: token, - http: &http.Client{}, - sendTimeout: telegramSendTimeout, + base: base, + token: token, + http: &http.Client{}, } } @@ -78,14 +77,11 @@ func telegramRejected(err error) bool { } // TelegramAPI calls the Telegram Bot API. The bot token is a part of each URL, -// so no error of TelegramAPI holds a URL, a net/http error, or the token. Its -// sendTimeout field limits each sendMessage, sendRichMessage, and -// sendChatAction call. +// so no error of TelegramAPI holds a URL, a net/http error, or the token. type TelegramAPI struct { - base string - token string - http *http.Client - sendTimeout time.Duration + base string + token string + http *http.Client } // GetUpdates waits for the updates from offset on, up to telegramPollSeconds, @@ -111,7 +107,7 @@ func (b *TelegramAPI) GetImage(ctx context.Context, id string) (data []byte, err Size int64 `json:"file_size"` } - err = b.call(ctx, b.sendTimeout, "getFile", map[string]any{"file_id": id}, &file) + err = b.call(ctx, telegramSendTimeout, "getFile", map[string]any{"file_id": id}, &file) if err != nil { return nil, fmt.Errorf("get image file: %w", err) } @@ -124,7 +120,7 @@ func (b *TelegramAPI) GetImage(ctx context.Context, id string) (data []byte, err return nil, errors.New("invalid image file path") } - ctx, cancel := context.WithTimeout(ctx, b.sendTimeout) + ctx, cancel := context.WithTimeout(ctx, telegramSendTimeout) defer cancel() request, err := http.NewRequestWithContext(ctx, http.MethodGet, b.base+"/file/bot"+b.token+"/"+file.Path, nil) @@ -134,13 +130,13 @@ func (b *TelegramAPI) GetImage(ctx context.Context, id string) (data []byte, err response, err := b.http.Do(request) if err != nil { - return nil, b.networkError("image download failed", err, b.sendTimeout) + return nil, b.networkError("image download failed", err, telegramSendTimeout) } defer func() { closeErr := response.Body.Close() if closeErr != nil { - err = errors.Join(err, b.networkError("close image download", closeErr, b.sendTimeout)) + err = errors.Join(err, b.networkError("close image download", closeErr, telegramSendTimeout)) } }() @@ -153,7 +149,7 @@ func (b *TelegramAPI) GetImage(ctx context.Context, id string) (data []byte, err data, err = io.ReadAll(io.LimitReader(response.Body, imageMaxBytes+1)) if err != nil { - return nil, b.networkError("image cut off", err, b.sendTimeout) + return nil, b.networkError("image cut off", err, telegramSendTimeout) } if len(data) > imageMaxBytes { @@ -177,14 +173,14 @@ func (b *TelegramAPI) SendMessage(ctx context.Context, chat int64, text string, params["entities"] = entities } - return b.call(ctx, b.sendTimeout, "sendMessage", params, nil) + return b.call(ctx, telegramSendTimeout, "sendMessage", params, nil) } // SendRichMessage sends markdown to chat as a rich message, which Telegram // renders itself, tables included. A call that takes longer than // telegramSendTimeout fails with a TelegramNetworkError. func (b *TelegramAPI) SendRichMessage(ctx context.Context, chat int64, markdown string) error { - return b.call(ctx, b.sendTimeout, "sendRichMessage", map[string]any{ + return b.call(ctx, telegramSendTimeout, "sendRichMessage", map[string]any{ "chat_id": chat, "rich_message": map[string]any{ "markdown": markdown, @@ -195,7 +191,7 @@ func (b *TelegramAPI) SendRichMessage(ctx context.Context, chat int64, markdown // SendTyping shows the typing action in chat for about five seconds. A call // that takes longer than telegramSendTimeout fails with a TelegramNetworkError. func (b *TelegramAPI) SendTyping(ctx context.Context, chat int64) error { - return b.call(ctx, b.sendTimeout, "sendChatAction", map[string]any{ + return b.call(ctx, telegramSendTimeout, "sendChatAction", map[string]any{ "chat_id": chat, "action": "typing", }, nil) diff --git a/telegram_api_test.go b/telegram_api_test.go index 98aa232..6fa4cba 100644 --- a/telegram_api_test.go +++ b/telegram_api_test.go @@ -6,6 +6,7 @@ import ( "net/http" "net/http/httptest" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/require" @@ -103,31 +104,34 @@ func TestSendCallsTimeOutOnAStalledConnection(t *testing.T) { for _, tc := range testCases { t.Run(tc.method, func(t *testing.T) { - server := newTelegramFake(t) - server.stall = func(call TelegramFakeCall) bool { - return call.method == tc.method - } - - api := newTelegramAPI(server.URL, telegramTestToken) - api.sendTimeout = 50 * time.Millisecond - start := time.Now() - err := tc.send(t.Context(), api) - require.EqualError(t, err, "no answer in 50ms") - require.Less(t, time.Since(start), 5*time.Second) - require.NotContains(t, err.Error(), telegramTestToken) - require.NotContains(t, err.Error(), server.URL) + synctest.Test(t, func(t *testing.T) { + server := newTelegramLocalFake(t) + server.stall = func(call TelegramFakeCall) bool { + return call.method == tc.method + } + + api := newTelegramAPI(server.URL, telegramTestToken) + api.http.Transport = server + start := time.Now() + err := tc.send(t.Context(), api) + require.EqualError(t, err, "no answer in 30s") + require.Equal(t, telegramSendTimeout, time.Since(start)) + require.NotContains(t, err.Error(), telegramTestToken) + require.NotContains(t, err.Error(), server.URL) + + retry, _ := telegramRetryable(err) + require.True(t, retry) - retry, _ := telegramRetryable(err) - require.True(t, retry) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() - api.sendTimeout = time.Hour - ctx, cancel := context.WithCancel(t.Context()) - time.AfterFunc(50*time.Millisecond, cancel) + time.AfterFunc(time.Second, cancel) - start = time.Now() - err = tc.send(ctx, api) - require.Error(t, err) - require.Less(t, time.Since(start), 5*time.Second) + start = time.Now() + err = tc.send(ctx, api) + require.Error(t, err) + require.Equal(t, time.Second, time.Since(start)) + }) }) } } diff --git a/telegram_fake_helper_test.go b/telegram_fake_helper_test.go index 6a89054..f6ccd84 100644 --- a/telegram_fake_helper_test.go +++ b/telegram_fake_helper_test.go @@ -98,6 +98,19 @@ func (f *TelegramFake) Image(id string, data []byte) { f.images[id] = data } +// RoundTrip handles a request in memory, so synctest can control its timers. +func (f *TelegramFake) RoundTrip(request *http.Request) (*http.Response, error) { + response := httptest.NewRecorder() + f.serve(response, request) + + ctx := request.Context() + if ctx.Err() != nil { + return nil, ctx.Err() + } + + return response.Result(), nil +} + func (f *TelegramFake) serve(w http.ResponseWriter, r *http.Request) { id, file := strings.CutPrefix(r.URL.Path, "/file/bot"+telegramTestToken+"/") if file { @@ -256,6 +269,23 @@ func newTelegramFake(t *testing.T) *TelegramFake { return f } +// newTelegramLocalFake returns the same fake API without network goroutines. +// Its RoundTrip handles requests directly for tests with fake time. +func newTelegramLocalFake(t *testing.T) *TelegramFake { + t.Helper() + + f := &TelegramFake{ + Server: &httptest.Server{URL: "https://example.com"}, + added: make(chan struct{}, 1), + stopped: make(chan struct{}), + } + t.Cleanup(func() { + close(f.stopped) + }) + + return f +} + // telegramOwnerMessage returns a text message of the owner in the private chat. func telegramOwnerMessage(text string) *TelegramMessage { return &TelegramMessage{ diff --git a/telegram_image_test.go b/telegram_image_test.go index 837a64e..7f7b7c0 100644 --- a/telegram_image_test.go +++ b/telegram_image_test.go @@ -194,8 +194,6 @@ func TestTelegramRetriesImageDownloadBeforeAdvancingOffset(t *testing.T) { return 0, "" } bot, logs := telegramStartTestBot(t, telegram) - bot.minWait = time.Millisecond - bot.maxWait = 5 * time.Millisecond server := newModelFake(t) socket := daemonStartTestWithBot(t, server, bot) terminal := terminalStartTestClient(t, socket) diff --git a/telegram_test.go b/telegram_test.go index 60e9e86..84f41c8 100644 --- a/telegram_test.go +++ b/telegram_test.go @@ -1,6 +1,7 @@ package main import ( + "context" "net/http" "os" "path/filepath" @@ -8,6 +9,7 @@ import ( "strconv" "strings" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/require" @@ -265,7 +267,7 @@ func TestBotTriesAgainAfterFailures(t *testing.T) { }) telegramWaitForSent(t, telegram, "ok") - require.Contains(t, logs.String(), "<4>receive failed retry=1ms: 502") + require.Contains(t, logs.String(), "<4>receive failed retry=1s: 502") require.Contains(t, logs.String(), "<6>receive back after=") require.Contains(t, logs.String(), "<4>send failed retry=1s: 429") @@ -279,30 +281,32 @@ func TestBotTriesAgainAfterFailures(t *testing.T) { } func TestDeliverTriesAgainAfterATimeout(t *testing.T) { - telegram := newTelegramFake(t) - stalls := 1 - telegram.stall = func(call TelegramFakeCall) bool { - if call.method != "sendMessage" || stalls == 0 { - return false - } + synctest.Test(t, func(t *testing.T) { + telegram := newTelegramLocalFake(t) + stalls := 1 + telegram.stall = func(call TelegramFakeCall) bool { + if call.method != "sendMessage" || stalls == 0 { + return false + } - stalls-- + stalls-- - return true - } + return true + } - bot, logs := telegramStartTestBot(t, telegram) - bot.api.sendTimeout = 50 * time.Millisecond - bot.deliver(t.Context(), TelegramPart{ - text: "hello", - }) + bot, logs := telegramStartTestBot(t, telegram) + bot.api.http.Transport = telegram + bot.deliver(t.Context(), TelegramPart{ + text: "hello", + }) - require.Equal(t, []string{"hello"}, telegram.Sent()) - require.Len(t, telegram.Calls(), 2) - require.Contains(t, logs.String(), "<4>send failed retry=1ms: no answer in 50ms\n") - require.Contains(t, logs.String(), "<6>send back after=") - require.Equal(t, 2, strings.Count(logs.String(), "\n")) - require.NotContains(t, logs.String(), telegramTestToken) + require.Equal(t, []string{"hello"}, telegram.Sent()) + require.Len(t, telegram.Calls(), 2) + require.Contains(t, logs.String(), "<4>send failed retry=1s: no answer in 30s\n") + require.Contains(t, logs.String(), "<6>send back after=1s\n") + require.Equal(t, 2, strings.Count(logs.String(), "\n")) + require.NotContains(t, logs.String(), telegramTestToken) + }) } func TestRepliesGoAsRichMessages(t *testing.T) { @@ -408,7 +412,7 @@ func TestRichMessageFailures(t *testing.T) { methods: []string{"sendRichMessage", "sendRichMessage"}, sent: []string{"a **secret** word"}, lines: 2, - logText: "<4>send failed retry=1ms: 500", + logText: "<4>send failed retry=1s: 500", }, { name: "other refusal drops the reply", @@ -639,45 +643,61 @@ func TestBotLogsNoTokenWhenTheNetworkFails(t *testing.T) { daemonStartTestWithBot(t, newModelFake(t), bot) require.Eventually(t, func() bool { - return strings.Contains(logs.String(), "<4>receive failed retry=1ms: no answer: ") + return strings.Contains(logs.String(), "<4>receive failed retry=1s: no answer: ") }, 10*time.Second, time.Millisecond) require.NotContains(t, logs.String(), telegramTestToken) } func TestBotLogsAnOutageOnlyAtItsStartAndEnd(t *testing.T) { - server := newModelFake(t) - server.reply = func(w http.ResponseWriter, _ ModelFakeCall) { - modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("ok")}, 100, 0, 5)) - } + synctest.Test(t, func(t *testing.T) { + telegram := newTelegramLocalFake(t) + failures := 5 + telegram.fail = func(call TelegramFakeCall) (int, string) { + if call.method != "getUpdates" || failures == 0 { + return 0, "" + } - telegram := newTelegramFake(t) - failures := 5 - telegram.fail = func(call TelegramFakeCall) (int, string) { - if call.method != "getUpdates" || failures == 0 { - return 0, "" - } + failures-- - failures-- + return http.StatusBadGateway, "Bad Gateway" + } - return http.StatusBadGateway, "Bad Gateway" - } + bot, logs := telegramStartTestBot(t, telegram) + bot.api.http.Transport = telegram + whatsapp, _, _ := whatsappStartTest(t, t.TempDir(), newTestClock()) + loop, _ := newTestIdleTurnLoop(t, whatsapp) - bot, logs := telegramStartTestBot(t, telegram) - daemonStartTestWithBot(t, server, bot) + telegram.Add(TelegramUpdate{ + UpdateID: 1, + Message: telegramOwnerMessage("a secret word"), + }) - telegram.Add(TelegramUpdate{ - UpdateID: 1, - Message: telegramOwnerMessage("a secret word"), + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + done := make(chan error, 1) + go func() { + done <- bot.poll(ctx, loop, whatsapp) + }() + + time.Sleep(31 * time.Second) + synctest.Wait() + require.Equal(t, 1, loop.memory.Len()) + + message, err := loop.memory.Message(0) + require.NoError(t, err) + require.Equal(t, "a secret word", message.Text) + + lines := strings.Split(strings.TrimSuffix(logs.String(), "\n"), "\n") + require.Equal(t, []string{ + "<4>receive failed retry=1s: 502 Bad Gateway", + "<6>receive back after=31s", + }, lines) + require.NotContains(t, logs.String(), "secret word") + require.NotContains(t, logs.String(), strconv.FormatInt(telegramTestOwner, 10)) + cancel() + require.NoError(t, <-done) }) - - telegramWaitForSent(t, telegram, "ok") - - lines := strings.Split(strings.TrimSuffix(logs.String(), "\n"), "\n") - require.Len(t, lines, 2) - require.Equal(t, "<4>receive failed retry=1ms: 502 Bad Gateway", lines[0]) - require.Regexp(t, `^<6>receive back after=[0-9]+m?s$`, lines[1]) - require.NotContains(t, logs.String(), "secret word") - require.NotContains(t, logs.String(), strconv.FormatInt(telegramTestOwner, 10)) } func TestSplitMessage(t *testing.T) { @@ -906,15 +926,12 @@ func TestLoadBotConfig(t *testing.T) { } // telegramStartTestBot returns a bot of telegramTestOwner on server, with its -// offset file in a temporary directory and short waits, and the buffer of its -// log. +// offset file in a temporary directory and the buffer of its log. func telegramStartTestBot(t *testing.T, server *TelegramFake) (*TelegramBot, *SyncBuffer) { t.Helper() logs := &SyncBuffer{} bot := newTelegramBot(newTelegramAPI(server.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), daemonTestLogger(logs)) - bot.minWait = time.Millisecond - bot.maxWait = 10 * time.Millisecond return bot, logs } diff --git a/turn_loop_helper_test.go b/turn_loop_helper_test.go new file mode 100644 index 0000000..06a6f93 --- /dev/null +++ b/turn_loop_helper_test.go @@ -0,0 +1,31 @@ +package main + +import ( + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// newTestIdleTurnLoop opens a loop with no model calls or turn goroutine. +// Callers can test queued messages and events with synctest fake time. +func newTestIdleTurnLoop(t *testing.T, whatsapp *WhatsApp) (*TurnLoop, *TurnEventRecorder) { + t.Helper() + + agent, err := openAgent(t.TempDir()) + require.NoError(t, err) + + memory, err := openMemory(t.Context(), agent.dir, time.Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, memory.Close()) + }) + + events := &TurnEventRecorder{} + client := newModelClient("https://example.com", ModelAPIKey(modelTestKey), agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl")) + loop, err := newTurnLoop(agent, memory, newTestSubagents(t), whatsapp, newTestSchedules(t, time.Now), client, time.Now, discardLog(), events.Add) + require.NoError(t, err) + + return loop, events +} diff --git a/whatsapp.go b/whatsapp.go index be1acc6..9365173 100644 --- a/whatsapp.go +++ b/whatsapp.go @@ -77,10 +77,6 @@ const ( // wait for Run. whatsappRequests = 8 - // whatsappLookupTime is the longest time of a read of the names in the - // session database or of the groups from WhatsApp. - whatsappLookupTime = 10 * time.Second - // whatsappNotLinked is the answer of a WhatsApp tool without a link. whatsappNotLinked = "WhatsApp is not linked. Tell the user to run /whatsapp pair PHONE in the terminal or in Telegram." ) @@ -107,20 +103,19 @@ type WhatsAppClient interface { // the client of a device with its event handler. func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logger *slog.Logger, newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient) (*WhatsApp, error) { w := &WhatsApp{ - ctx: ctx, - dir: filepath.Join(dir, whatsappDirName), - clock: clock, - logger: logger, - newClient: newClient, - noticeDelay: whatsappNoticeDelay, - requests: make(chan WhatsAppRequest, whatsappRequests), - logouts: make(chan struct{}, 1), - groups: newWhatsAppGroups(logger), + ctx: ctx, + dir: filepath.Join(dir, whatsappDirName), + clock: clock, + logger: logger, + newClient: newClient, + requests: make(chan WhatsAppRequest, whatsappRequests), + logouts: make(chan struct{}, 1), + groups: newWhatsAppGroups(logger), } var err error - w.state, err = openWhatsAppState(filepath.Join(w.dir, whatsappStateFileName), whatsappBatchQuiet, whatsappBatchWindow) + w.state, err = openWhatsAppState(filepath.Join(w.dir, whatsappStateFileName)) if err != nil { return nil, err } @@ -185,18 +180,17 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge // log, and those of private chats hold phone numbers, because the agent needs // them to send. type WhatsApp struct { - ctx context.Context - dir string - clock func() time.Time - logger *slog.Logger - newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient - noticeDelay time.Duration - requests chan WhatsAppRequest - logouts chan struct{} - state *WhatsAppState - groups *WhatsAppGroups - inbox *WhatsAppInbox - routes *Routes + ctx context.Context + dir string + clock func() time.Time + logger *slog.Logger + newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient + requests chan WhatsAppRequest + logouts chan struct{} + state *WhatsAppState + groups *WhatsAppGroups + inbox *WhatsAppInbox + routes *Routes sessionMutex sync.Mutex mutex sync.Mutex @@ -575,7 +569,7 @@ func (w *WhatsApp) startOutage() bool { w.disconnectedAt = w.clock() w.disconnections++ disconnections := w.disconnections - w.timer = time.AfterFunc(w.noticeDelay, func() { + w.timer = time.AfterFunc(whatsappNoticeDelay, func() { w.disconnectedLong(disconnections) }) } @@ -896,10 +890,11 @@ func (w *WhatsApp) linkedBy(client WhatsAppClient) { // handle takes one event of the library and tells if it was handled. A message // goes to the inbox. The names of the groups follow the events of the groups. A // connection starts a catch-up of the inbox, which ends at OfflineSyncCompleted -// or after its catchupLimit. A disconnection keeps the time of its start while -// Run or the library tries again. After noticeDelay of a disconnection, all -// clients get one notice, and they get one more when the connection comes back. -// A disconnection that the library does not try again gives a notice at once. +// or after whatsappCatchupLimit. A disconnection keeps the time of its start +// while Run or the library tries again. After whatsappNoticeDelay of a +// disconnection, all clients get one notice, and they get one more when the +// connection comes back. A disconnection that the library does not try again +// gives a notice at once. // // The log of the daemon gets one line for each change of the connection of the // linked device: connected, a failed first connection, a lost connection, and diff --git a/whatsapp_fake_helper_test.go b/whatsapp_fake_helper_test.go index e936f45..9f80b38 100644 --- a/whatsapp_fake_helper_test.go +++ b/whatsapp_fake_helper_test.go @@ -9,6 +9,7 @@ import ( "slices" "sync" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/require" @@ -282,6 +283,34 @@ type WhatsAppFakeSend struct { id string } +// WhatsAppRetryTest runs WhatsApp without a model or network client call. All +// timers and channels belong to its synctest bubble, so backoff uses fake time. +type WhatsAppRetryTest struct { + whatsapp *WhatsApp + fake *WhatsAppFake + logs *SyncBuffer + loop *TurnLoop + events *TurnEventRecorder + cancel context.CancelFunc + done chan error +} + +func (r *WhatsAppRetryTest) start(t *testing.T) { + t.Helper() + + ctx, cancel := context.WithCancel(t.Context()) + r.cancel = cancel + r.done = make(chan error, 1) + go func() { + r.done <- r.whatsapp.Run(ctx, r.loop, r.events.Add) + }() + + t.Cleanup(func() { + cancel() + synctest.Wait() + }) +} + // whatsappStartTest opens the WhatsApp of the data directory dir with a fake // client and the test clock, and closes it at the end of the test. The daemon // log goes to the returned buffer. @@ -385,7 +414,7 @@ func whatsappTestLogTexts(t *testing.T, memory *Memory) []string { } // whatsappStartTestLinked opens a WhatsApp with a linked device in a temporary -// directory, with a quiet time of 50 ms for its batches. +// directory. func whatsappStartTestLinked(t *testing.T) (*WhatsApp, *WhatsAppFake) { t.Helper() @@ -393,7 +422,6 @@ func whatsappStartTestLinked(t *testing.T) (*WhatsApp, *WhatsAppFake) { whatsappTestSession(t, dir, true) whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock()) - whatsapp.state.quiet = 50 * time.Millisecond return whatsapp, fake } @@ -421,3 +449,22 @@ func whatsappTestContacts(t *testing.T, whatsapp *WhatsApp) { require.NoError(t, err) require.NoError(t, contacts.PutContactName(t.Context(), whatsappTestOwn, "Me", "Me Souza")) } + +func whatsappRetryTest(t *testing.T) *WhatsAppRetryTest { + t.Helper() + + dir := t.TempDir() + whatsappTestSession(t, dir, true) + + whatsapp, fake, logs := whatsappStartTest(t, dir, newTestClock()) + whatsapp.clock = time.Now + loop, events := newTestIdleTurnLoop(t, whatsapp) + + return &WhatsAppRetryTest{ + whatsapp: whatsapp, + fake: fake, + logs: logs, + loop: loop, + events: events, + } +} diff --git a/whatsapp_groups.go b/whatsapp_groups.go index ec5c892..4d6c7e5 100644 --- a/whatsapp_groups.go +++ b/whatsapp_groups.go @@ -9,6 +9,10 @@ import ( ) const ( + // whatsappGroupsLookupTime is the longest time of a read of group + // names. + whatsappGroupsLookupTime = 10 * time.Second + // whatsappGroupsRefresh is the shortest time between two reads of the // groups from WhatsApp for a group with no name. whatsappGroupsRefresh = time.Minute @@ -77,7 +81,7 @@ func (g *WhatsAppGroups) read(ctx context.Context, link WhatsAppLink) bool { return false } - ctx, cancel := context.WithTimeout(ctx, whatsappLookupTime) + ctx, cancel := context.WithTimeout(ctx, whatsappGroupsLookupTime) defer cancel() groups, err := link.client.GetJoinedGroups(ctx) diff --git a/whatsapp_image.go b/whatsapp_image.go index 4277b0a..d4b88d8 100644 --- a/whatsapp_image.go +++ b/whatsapp_image.go @@ -4,15 +4,11 @@ import ( "context" "errors" "fmt" - "time" "go.mau.fi/whatsmeow" "go.mau.fi/whatsmeow/proto/waE2E" ) -// whatsappImageTime is the longest time of an image download. -const whatsappImageTime = 2 * time.Minute - // whatsappImage downloads only supported images and stores them before the // event can be acknowledged. Both the declared size and the downloaded bytes // must fit the image store. Decoded bytes determine the stored format. diff --git a/whatsapp_image_test.go b/whatsapp_image_test.go index 919d656..91af70a 100644 --- a/whatsapp_image_test.go +++ b/whatsapp_image_test.go @@ -80,28 +80,28 @@ func TestWhatsAppImagesAreDurableBeforeAcknowledgement(t *testing.T) { } func TestWhatsAppImageDownloadCanOutlastNameLookup(t *testing.T) { - whatsapp, fake := whatsappStartTestLinked(t) - whatsapp.client = fake - whatsapp.inbox.lookupTime = 10 * time.Millisecond - whatsapp.inbox.imageTime = time.Second - fake.downloadWait = 50 * time.Millisecond - data := imageTestPNG(t, 3, 2) - fake.downloadData = data - - require.True(t, whatsapp.handle(whatsappImageTestEvent(whatsappTestAna, "Slow photo."))) - - restored := whatsappTestState(t, whatsapp.state.path) - require.NoError(t, restored.CloseAll()) - require.Len(t, restored.Closed(), 1) - - text := restored.Closed()[0].Messages[0].Text - names := imageFiles(text) - require.Len(t, names, 1) - require.Equal(t, imageText(names, "Slow photo."), text) - - saved, err := whatsapp.inbox.images.Read(names[0]) - require.NoError(t, err) - require.Equal(t, data, saved) + synctest.Test(t, func(t *testing.T) { + whatsapp, fake := whatsappStartTestLinked(t) + whatsapp.client = fake + fake.downloadWait = whatsappLookupTime + time.Second + data := imageTestPNG(t, 3, 2) + fake.downloadData = data + + require.True(t, whatsapp.handle(whatsappImageTestEvent(whatsappTestAna, "Slow photo."))) + + restored := whatsappTestState(t, whatsapp.state.path) + require.NoError(t, restored.CloseAll()) + require.Len(t, restored.Closed(), 1) + + text := restored.Closed()[0].Messages[0].Text + names := imageFiles(text) + require.Len(t, names, 1) + require.Equal(t, imageText(names, "Slow photo."), text) + + saved, err := whatsapp.inbox.images.Read(names[0]) + require.NoError(t, err) + require.Equal(t, data, saved) + }) } func TestWhatsAppRunCancellationEndsAnImageDownload(t *testing.T) { diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go index 06f1c55..392593f 100644 --- a/whatsapp_inbox.go +++ b/whatsapp_inbox.go @@ -16,6 +16,13 @@ import ( ) const ( + // whatsappLookupTime is the longest time of a read of names in the + // session. + whatsappLookupTime = 10 * time.Second + + // whatsappImageTime is the longest time of an image download. + whatsappImageTime = 2 * time.Minute + // whatsappSource is the source of the work events of the batches. whatsappSource = "whatsapp" @@ -29,16 +36,13 @@ const ( // holds shared images, and logger gets image failures. func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *WhatsAppGroups, images *ImageStore, logger *slog.Logger) *WhatsAppInbox { return &WhatsAppInbox{ - state: state, - clock: clock, - groups: groups, - images: images, - logger: logger, - lookupTime: whatsappLookupTime, - imageTime: whatsappImageTime, - catchupLimit: whatsappCatchupLimit, - wake: make(chan struct{}, 1), - failures: make(chan error, 1), + state: state, + clock: clock, + groups: groups, + images: images, + logger: logger, + wake: make(chan struct{}, 1), + failures: make(chan error, 1), } } @@ -48,16 +52,16 @@ func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *What // // After each connection, the servers send the messages that came while the // daemon was away, and then OfflineSyncCompleted. Until then, or until -// catchupLimit passes, the closed batches wait in the state file. At the end of -// the catch-up, deliver closes all batches and logs them. It wakes the root -// once for its batches, so their catch-up gives one turn, not one for each -// chat. The batches of the state file at the start wait for the first catch-up -// in the same way. The event handler acknowledges a message only after the -// state file holds it, so the servers send again a message that a crash lost. A -// batch leaves the state file only after its work message is durable in the log -// or agent queue. A crash between the two can repeat an unrouted batch once. A -// routed batch enters the durable agent queue, which prevents a second log -// entry and wakes only the child. A missing child uses the root. +// whatsappCatchupLimit passes, the closed batches wait in the state file. At +// the end of the catch-up, deliver closes all batches and logs them. It wakes +// the root once for its batches, so their catch-up gives one turn, not one for +// each chat. The batches of the state file at the start wait for the first +// catch-up in the same way. The event handler acknowledges a message only after +// the state file holds it, so the servers send again a message that a crash +// lost. A batch leaves the state file only after its work message is durable in +// the log or agent queue. A crash between the two can repeat an unrouted batch +// once. A routed batch enters the durable agent queue, which prevents a second +// log entry and wakes only the child. A missing child uses the root. // // The state file is the one of WhatsApp, which keeps the send records in it // too. groups gives the names of the groups. The mutex protects the state of @@ -65,16 +69,13 @@ func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *What // goroutines. wake tells Run that a batch can be due or that a catch-up ended, // and failures gives Run the error of a write of the state file. type WhatsAppInbox struct { - state *WhatsAppState - clock func() time.Time - groups *WhatsAppGroups - images *ImageStore - logger *slog.Logger - lookupTime time.Duration - imageTime time.Duration - catchupLimit time.Duration - wake chan struct{} - failures chan error + state *WhatsAppState + clock func() time.Time + groups *WhatsAppGroups + images *ImageStore + logger *slog.Logger + wake chan struct{} + failures chan error mutex sync.Mutex catchingUp bool @@ -206,7 +207,7 @@ func (i *WhatsAppInbox) waitForCatchup() { } // startCatchup starts the catch-up of a new connection, which ends at -// endCatchup or after catchupLimit. +// endCatchup or after whatsappCatchupLimit. func (i *WhatsAppInbox) startCatchup() { i.mutex.Lock() defer i.mutex.Unlock() @@ -218,7 +219,7 @@ func (i *WhatsAppInbox) startCatchup() { i.catchupTimer.Stop() } - i.catchupTimer = time.AfterFunc(i.catchupLimit, func() { + i.catchupTimer = time.AfterFunc(whatsappCatchupLimit, func() { i.caughtUpBy(catchups) }) } @@ -229,7 +230,7 @@ func (i *WhatsAppInbox) startCatchup() { // daemon. An image that cannot be downloaded or stored gets a text marker. A // message that the batches do not take is handled with nothing to do. func (i *WhatsAppInbox) receive(ctx context.Context, link WhatsAppLink, event *events.Message) bool { - lookupCtx, cancel := context.WithTimeout(ctx, i.lookupTime) + lookupCtx, cancel := context.WithTimeout(ctx, whatsappLookupTime) defer cancel() chat, message, found := whatsappMessage(lookupCtx, link.device, link.own, event) @@ -238,7 +239,7 @@ func (i *WhatsAppInbox) receive(ctx context.Context, link WhatsAppLink, event *e } if event.Message.GetImageMessage() != nil { - ctx, cancel := context.WithTimeout(ctx, i.imageTime) + ctx, cancel := context.WithTimeout(ctx, whatsappImageTime) defer cancel() message.Text = i.imageText(ctx, link, event, message.Text) diff --git a/whatsapp_inbox_test.go b/whatsapp_inbox_test.go index cea65fe..a0a0f9d 100644 --- a/whatsapp_inbox_test.go +++ b/whatsapp_inbox_test.go @@ -3,6 +3,7 @@ package main import ( "net/http" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/require" @@ -115,6 +116,9 @@ func TestWhatsAppGroupTurns(t *testing.T) { event.Info.PushName = "Bruno Lima" tc.message(event) require.True(t, fake.Emit(event)) + require.NoError(t, whatsapp.state.CloseAll()) + + whatsapp.inbox.wake <- struct{}{} text := "[WhatsApp] Family, group 120363000000000001@g.us, Thursday 2026-01-01\n" + tc.line if tc.turn { @@ -133,6 +137,27 @@ func TestWhatsAppGroupTurns(t *testing.T) { } } +func TestWhatsAppRunClosesABatchAfterQuietTime(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + test := whatsappRetryTest(t) + test.start(t) + synctest.Wait() + require.True(t, test.fake.Emit(&events.Connected{})) + require.True(t, test.fake.Emit(&events.OfflineSyncCompleted{})) + synctest.Wait() + require.True(t, test.fake.Emit(whatsappTestMessage(whatsappTestAna, whatsappTestOwn, "m1", "On my way.", time.Now()))) + synctest.Wait() + time.Sleep(whatsappBatchQuiet - time.Nanosecond) + synctest.Wait() + require.Zero(t, test.loop.memory.Len()) + + time.Sleep(time.Nanosecond) + synctest.Wait() + require.Equal(t, 1, test.loop.memory.Len()) + require.Contains(t, test.events.Last("work"), "You: On my way.") + }) +} + func TestWhatsAppSkipsExcludedMessages(t *testing.T) { at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) testCases := []struct { @@ -235,17 +260,20 @@ func TestWhatsAppCatchupGivesOneTurn(t *testing.T) { } func TestWhatsAppCatchupEndsAfterItsLimit(t *testing.T) { - whatsapp, fake := whatsappStartTestLinked(t) - whatsapp.inbox.catchupLimit = 100 * time.Millisecond - socket, memory := daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp) - client := terminalStartTestClient(t, socket) - client.WaitFor(t, "[connected to OneChat]") - <-fake.connects + synctest.Test(t, func(t *testing.T) { + test := whatsappRetryTest(t) + test.whatsapp.client = test.fake + require.True(t, test.whatsapp.handle(&events.Connected{})) + require.True(t, test.whatsapp.handle(whatsappTestMessage(whatsappTestAna, whatsappTestOwn, "m1", "On my way.", time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)))) + require.NoError(t, test.whatsapp.inbox.deliver(t.Context(), test.loop, test.whatsapp.link(), false)) + require.Zero(t, test.loop.memory.Len()) - require.True(t, fake.Emit(&events.Connected{})) - require.True(t, fake.Emit(whatsappTestMessage(whatsappTestAna, whatsappTestOwn, "m1", "On my way.", time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)))) - client.WaitFor(t, "12:00 You: On my way.") - require.Len(t, whatsappTestLogTexts(t, memory), 1) + time.Sleep(whatsappCatchupLimit) + synctest.Wait() + require.NoError(t, test.whatsapp.inbox.deliver(t.Context(), test.loop, test.whatsapp.link(), false)) + require.Equal(t, 1, test.loop.memory.Len()) + require.Contains(t, test.events.Last("work"), "12:00 You: On my way.") + }) } func TestWhatsAppStateOfTheStateFileWaitForTheCatchup(t *testing.T) { diff --git a/whatsapp_route_test.go b/whatsapp_route_test.go index 9c855c9..c6dbf44 100644 --- a/whatsapp_route_test.go +++ b/whatsapp_route_test.go @@ -390,7 +390,7 @@ func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) { func whatsappRouteTestInbox(t *testing.T, dir string) *WhatsAppInbox { t.Helper() - state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName), whatsappBatchQuiet, whatsappBatchWindow) + state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName)) require.NoError(t, err) return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()), &ImageStore{dir: filepath.Join(dir, imageDirName)}, discardLog()) diff --git a/whatsapp_send_test.go b/whatsapp_send_test.go index 469912b..480e7db 100644 --- a/whatsapp_send_test.go +++ b/whatsapp_send_test.go @@ -39,7 +39,7 @@ func TestWhatsAppSendSendsOnceForEachID(t *testing.T) { }, }, fake.Sent()) - again, err := openWhatsAppState(filepath.Join(whatsapp.dir, whatsappStateFileName), whatsappBatchQuiet, whatsappBatchWindow) + again, err := openWhatsAppState(filepath.Join(whatsapp.dir, whatsappStateFileName)) require.NoError(t, err) record, found := again.Send("keys") diff --git a/whatsapp_state.go b/whatsapp_state.go index c09423d..41bd2f0 100644 --- a/whatsapp_state.go +++ b/whatsapp_state.go @@ -32,13 +32,11 @@ const ( ) // openWhatsAppState reads the batches and the send records of the state file at -// path. A missing file holds none. A batch closes after quiet with no message, -// or after window at the latest. -func openWhatsAppState(path string, quiet, window time.Duration) (*WhatsAppState, error) { +// path. A missing file holds none. A batch closes after whatsappBatchQuiet with +// no message, or after whatsappBatchWindow at the latest. +func openWhatsAppState(path string) (*WhatsAppState, error) { s := &WhatsAppState{ - path: path, - quiet: quiet, - window: window, + path: path, } data, err := os.ReadFile(path) @@ -61,11 +59,12 @@ func openWhatsAppState(path string, quiet, window time.Duration) (*WhatsAppState // WhatsAppState holds the state file: the send records, and the batches of the // WhatsApp messages that are not in the log yet, open and closed, in the order // of their first message. The messages of one chat form one batch while the -// times of the messages are at most quiet apart, at most window after the -// first, and the batch has fewer than whatsappBatchMessages messages. The times -// are the WhatsApp times, so a catch-up after a stop gives the same batches as -// live messages. A batch also closes when quiet passes after the arrival of its -// last message, or window after the arrival of its first message, in real time. +// times of the messages are at most whatsappBatchQuiet apart, at most +// whatsappBatchWindow after the first, and the batch has fewer than +// whatsappBatchMessages messages. The times are the WhatsApp times, so a +// catch-up after a stop gives the same batches as live messages. A batch also +// closes when whatsappBatchQuiet passes after the arrival of its last message, +// or whatsappBatchWindow after the arrival of its first message, in real time. // // A send record keeps the result of a send by its ID, so that a repeated call // of the send tool does not send again. The state file keeps the last @@ -76,9 +75,7 @@ func openWhatsAppState(path string, quiet, window time.Duration) (*WhatsAppState // library adds messages while Run closes and removes batches, and the turns // send. type WhatsAppState struct { - path string - quiet time.Duration - window time.Duration + path string mutex sync.Mutex file WhatsAppStateFile @@ -102,7 +99,7 @@ func (s *WhatsAppState) Add(chat WhatsAppChat, message WhatsAppMessage, arrival return false, nil } - if message.Time.Sub(batch.Messages[len(batch.Messages)-1].Time) > s.quiet || message.Time.Sub(batch.Messages[0].Time) > s.window { + if message.Time.Sub(batch.Messages[len(batch.Messages)-1].Time) > whatsappBatchQuiet || message.Time.Sub(batch.Messages[0].Time) > whatsappBatchWindow { batch.Closed = true closed = true index = -1 @@ -261,8 +258,8 @@ func (s *WhatsAppState) open(chat string) int { // deadline returns when batch closes in real time. func (s *WhatsAppState) deadline(batch WhatsAppBatch) time.Time { - quiet := batch.Last.Add(s.quiet) - window := batch.First.Add(s.window) + quiet := batch.Last.Add(whatsappBatchQuiet) + window := batch.First.Add(whatsappBatchWindow) if window.Before(quiet) { return window } diff --git a/whatsapp_state_test.go b/whatsapp_state_test.go index 2631b4d..1cfedee 100644 --- a/whatsapp_state_test.go +++ b/whatsapp_state_test.go @@ -134,7 +134,7 @@ func TestWhatsAppStateRefusesABrokenFile(t *testing.T) { path := filepath.Join(t.TempDir(), whatsappStateFileName) require.NoError(t, os.WriteFile(path, []byte("{"), 0o600)) - _, err := openWhatsAppState(path, whatsappBatchQuiet, whatsappBatchWindow) + _, err := openWhatsAppState(path) require.ErrorContains(t, err, path) } @@ -159,7 +159,7 @@ func TestWhatsAppStateKeepsTheLastSendRecords(t *testing.T) { func whatsappTestState(t *testing.T, path string) *WhatsAppState { t.Helper() - batches, err := openWhatsAppState(path, whatsappBatchQuiet, whatsappBatchWindow) + batches, err := openWhatsAppState(path) require.NoError(t, err) return batches diff --git a/whatsapp_test.go b/whatsapp_test.go index 96ec2df..f098c90 100644 --- a/whatsapp_test.go +++ b/whatsapp_test.go @@ -1,7 +1,6 @@ package main import ( - "context" "errors" "os" "path/filepath" @@ -606,34 +605,33 @@ func TestWhatsAppLogoutFromThePhone(t *testing.T) { } func TestWhatsAppLongDisconnectionGivesOneNotice(t *testing.T) { - dir := t.TempDir() - whatsappTestSession(t, dir, true) - - clock := newTestClock() - whatsapp, fake, _ := whatsappStartTest(t, dir, clock) - whatsapp.noticeDelay = 300 * time.Millisecond - socket, _ := daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp) - client := terminalStartTestClient(t, socket) - client.WaitFor(t, "[connected to OneChat]") - <-fake.connects - - require.True(t, fake.Emit(&events.Connected{})) - require.True(t, fake.Emit(&events.Disconnected{})) - require.True(t, fake.Emit(&events.Connected{})) - time.Sleep(400 * time.Millisecond) - require.Equal(t, "[connected to OneChat]\n", client.Output()) + synctest.Test(t, func(t *testing.T) { + test := whatsappRetryTest(t) + test.start(t) + synctest.Wait() - clock.now = clock.now.Add(time.Hour) + require.True(t, test.fake.Emit(&events.Connected{})) + require.True(t, test.fake.Emit(&events.Disconnected{})) + require.True(t, test.fake.Emit(&events.Connected{})) + time.Sleep(whatsappNoticeDelay + time.Second) + synctest.Wait() + require.Zero(t, test.events.Count("notice")) - require.True(t, fake.Emit(&events.Disconnected{})) - require.True(t, fake.Emit(&events.Disconnected{})) - client.WaitFor(t, "WhatsApp is disconnected since 2026-01-01 13:00. OneChat tries again.\n") - require.True(t, fake.Emit(&events.Disconnected{})) - time.Sleep(400 * time.Millisecond) + require.True(t, test.fake.Emit(&events.Disconnected{})) + require.True(t, test.fake.Emit(&events.Disconnected{})) + time.Sleep(whatsappNoticeDelay) + synctest.Wait() + require.Equal(t, 1, test.events.Count("notice")) + require.Contains(t, test.events.Last("notice"), "OneChat tries again.") + require.True(t, test.fake.Emit(&events.Disconnected{})) + time.Sleep(whatsappNoticeDelay) + synctest.Wait() + require.Equal(t, 1, test.events.Count("notice")) - require.True(t, fake.Emit(&events.Connected{})) - client.WaitFor(t, "WhatsApp is connected again.\n") - require.Equal(t, 1, strings.Count(client.Output(), "WhatsApp is disconnected since")) + require.True(t, test.fake.Emit(&events.Connected{})) + require.Equal(t, 2, test.events.Count("notice")) + require.Equal(t, "WhatsApp is connected again.", test.events.Last("notice")) + }) } func TestWhatsAppDisconnectionWithNoRetryGivesANoticeAtOnce(t *testing.T) { @@ -755,62 +753,3 @@ func TestSubagentHasNoWhatsAppTools(t *testing.T) { require.Equal(t, `error: unknown tool "whatsapp_find"`, box.run(t.Context(), "whatsapp_find", `{"name":"ana"}`)) require.Equal(t, `error: unknown tool "whatsapp_send"`, box.run(t.Context(), "whatsapp_send", `{"id":"a","chat":"15550002222@s.whatsapp.net","text":"hi"}`)) } - -// WhatsAppRetryTest runs WhatsApp without a model or network client call. All -// timers and channels belong to its synctest bubble, so backoff uses fake time. -type WhatsAppRetryTest struct { - whatsapp *WhatsApp - fake *WhatsAppFake - logs *SyncBuffer - loop *TurnLoop - events *TurnEventRecorder - cancel context.CancelFunc - done chan error -} - -func (r *WhatsAppRetryTest) start(t *testing.T) { - t.Helper() - - ctx, cancel := context.WithCancel(t.Context()) - r.cancel = cancel - r.done = make(chan error, 1) - go func() { - r.done <- r.whatsapp.Run(ctx, r.loop, r.events.Add) - }() - - t.Cleanup(func() { - cancel() - synctest.Wait() - }) -} - -func whatsappRetryTest(t *testing.T) *WhatsAppRetryTest { - t.Helper() - - dir := t.TempDir() - whatsappTestSession(t, dir, true) - - whatsapp, fake, logs := whatsappStartTest(t, dir, newTestClock()) - whatsapp.clock = time.Now - agent, err := openAgent(t.TempDir()) - require.NoError(t, err) - - memory, err := openMemory(t.Context(), agent.dir, time.Now, newFakeSummarizer(), discardLog()) - require.NoError(t, err) - t.Cleanup(func() { - require.NoError(t, memory.Close()) - }) - - events := &TurnEventRecorder{} - client := newModelClient("https://example.com", ModelAPIKey(modelTestKey), agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl")) - loop, err := newTurnLoop(agent, memory, newTestSubagents(t), whatsapp, newTestSchedules(t, time.Now), client, time.Now, discardLog(), events.Add) - require.NoError(t, err) - - return &WhatsAppRetryTest{ - whatsapp: whatsapp, - fake: fake, - logs: logs, - loop: loop, - events: events, - } -}