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 b3a65ea94193a6431693313ca6f5833d836ccefb..a1eba1c1665f11e35cb800a71a25978dd30f6fa6 100644 (file)
--- 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,
                owner:   owner,
                clock:   clock,
                logger:  logger,

-               check:   scheduleCheck,

                changed: make(chan struct{}, 1),
        }
 
                changed: make(chan struct{}, 1),
        }
 


@@ -118,7 +117,6 @@ type Schedules struct {
        owner   string
        clock   func() time.Time
        logger  *slog.Logger
        owner   string
        clock   func() time.Time
        logger  *slog.Logger

-       check   time.Duration

        changed chan struct{}
 
        mutex sync.Mutex
        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
 }
 
 // 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)
 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 {
 
                wait, found := s.Wait(s.clock())
                if found {

-                       timer = time.NewTimer(min(wait, s.check))
+                       timer = time.NewTimer(min(wait, scheduleCheck))

                        timeout = timer.C
                }
 
                        timeout = timer.C
                }
 

diff --git a/schedule_test.go b/schedule_test.go
index 6610c26b141a83a3f99bbc9f4a38946206c73f19..5743a22bb3fdae5fd18289a348a58b0ead9fb4cd 100644 (file)
--- a/schedule_test.go
+++ b/schedule_test.go
@@ -7,6 +7,7 @@ import (
        "strings"
        "sync"
        "testing"
        "strings"
        "sync"
        "testing"

+       "testing/synctest"

        "time"
 
        "github.com/stretchr/testify/require"
        "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))
        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())
 
        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, 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)
        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))
        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]")
 }
 
        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{
 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
 }
 
 // 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()
 
 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)
        test := newTestTurnLoop(t, server)
        ctx, cancel := context.WithCancel(t.Context())
        done := make(chan error)

diff --git a/telegram.go b/telegram.go
index a595b68fd0d1756efc0b520e4c15d17781386c6a..f3b3bb22ee30db537859b1a6682294e18becd010 100644 (file)
--- a/telegram.go
+++ b/telegram.go
@@ -18,6 +18,14 @@ import (
 )
 
 const (
 )
 
 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"
 
        // 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,
                owner:      owner,
                offsetPath: offsetPath,
                logger:     logger,

-               minWait:    time.Second,
-               maxWait:    time.Minute,

                ready:      make(chan struct{}, 1),
        }
 }
                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.
 //
 // 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
 type TelegramBot struct {
        api        *TelegramAPI
        owner      int64
        offsetPath string
        logger     *slog.Logger

-       minWait    time.Duration
-       maxWait    time.Duration

 
        mutex sync.Mutex
        queue []Event
 
        mutex sync.Mutex
        queue []Event


@@ -279,8 +283,8 @@ func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop, whatsapp *WhatsApp)
                back:   "image receive back",
        }
 
                back:   "image receive back",
        }
 

-       imageWait := b.minWait
-       wait := b.minWait
+       imageWait := telegramRetryDelay
+       wait := telegramRetryDelay

 
 poll:
        for ctx.Err() == nil {
 
 poll:
        for ctx.Err() == nil {


@@ -298,7 +302,7 @@ poll:
 
                outage.End()
 
 
                outage.End()
 

-               wait = b.minWait
+               wait = telegramRetryDelay

 
                for _, update := range updates {
                        if ctx.Err() != nil {
 
                for _, update := range updates {
                        if ctx.Err() != nil {


@@ -318,7 +322,7 @@ poll:
 
                        imageOutage.End()
 
 
                        imageOutage.End()
 

-                       imageWait = b.minWait
+                       imageWait = telegramRetryDelay

 
                        offset = update.UpdateID + 1
                        err = telegramSaveOffset(b.offsetPath, offset)
 
                        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
 }
 
 // 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)
 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():
        }
 
        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
 }
 
 // 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()
 // 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)
        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()
 // 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)
        for {
                err := b.api.SendRichMessage(ctx, b.owner, markdown)
                retry, asked := telegramRetryable(err)

diff --git a/telegram_api.go b/telegram_api.go
index 32ad9a93b7616ad1dc6371de439386a35c948ac1..a04b2fa56e88af4aca8480d0dc29bbad964846af 100644 (file)
--- 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{
 // 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,
 }
 
 // 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 {
 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,
 }
 
 // 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"`
        }
 
                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)
        }
        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")
        }
 
                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)
        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 {
 
        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 {
        }
 
        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 {
 
        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 {
        }
 
        if len(data) > imageMaxBytes {


@@ -177,14 +173,14 @@ func (b *TelegramAPI) SendMessage(ctx context.Context, chat int64, text string,
                params["entities"] = entities
        }
 
                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 {
 }
 
 // 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,
                "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 {
 // 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)
                "chat_id": chat,
                "action":  "typing",
        }, nil)

diff --git a/telegram_api_test.go b/telegram_api_test.go
index 98aa232a9539894007921f177c3622c81a88462d..6fa4cba18179bd048ba21c45961cae910e22b9f4 100644 (file)
--- a/telegram_api_test.go
+++ b/telegram_api_test.go
@@ -6,6 +6,7 @@ import (
        "net/http"
        "net/http/httptest"
        "testing"
        "net/http"
        "net/http/httptest"
        "testing"

+       "testing/synctest"

        "time"
 
        "github.com/stretchr/testify/require"
        "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) {
 
        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 6a89054a8b0e10702b8b2d0ba05de3768fda7ffa..f6ccd8425d41e36372fc51fb8e8a5afc7f2b2216 100644 (file)
--- 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
 }
 
        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 {
 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
 }
 
        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{
 // 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 837a64e7cc5c40e9eaebc4892aee95d4e158803b..7f7b7c008ef979089e543b52fa05281f2b19dc84 100644 (file)
--- 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)
                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)
        server := newModelFake(t)
        socket := daemonStartTestWithBot(t, server, bot)
        terminal := terminalStartTestClient(t, socket)

diff --git a/telegram_test.go b/telegram_test.go
index 60e9e86b8147888ca0d76ce84410f9adb316455c..84f41c80c50894cbcd8da6e0ad2f2be3b2f56315 100644 (file)
--- a/telegram_test.go
+++ b/telegram_test.go
@@ -1,6 +1,7 @@
 package main
 
 import (
 package main
 
 import (

+       "context"

        "net/http"
        "os"
        "path/filepath"
        "net/http"
        "os"
        "path/filepath"


@@ -8,6 +9,7 @@ import (
        "strconv"
        "strings"
        "testing"
        "strconv"
        "strings"
        "testing"

+       "testing/synctest"

        "time"
 
        "github.com/stretchr/testify/require"
        "time"
 
        "github.com/stretchr/testify/require"


@@ -265,7 +267,7 @@ func TestBotTriesAgainAfterFailures(t *testing.T) {
        })
 
        telegramWaitForSent(t, telegram, "ok")
        })
 
        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")
 
        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) {
 }
 
 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) {
 }
 
 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,
                        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",
                },
                {
                        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 {
        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) {
        }, 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) {
 }
 
 func TestSplitMessage(t *testing.T) {


@@ -906,15 +926,12 @@ func TestLoadBotConfig(t *testing.T) {
 }
 
 // telegramStartTestBot returns a bot of telegramTestOwner on server, with its
 }
 
 // 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))
 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
 }
 
        return bot, logs
 }

diff --git a/turn_loop_helper_test.go b/turn_loop_helper_test.go
new file mode 100644 (file)
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 be1acc65b8e763b131f107513b3efd9c2e71af0f..93651736c5285c885f3656a814a8056f2108977f 100644 (file)
--- a/whatsapp.go
+++ b/whatsapp.go
@@ -77,10 +77,6 @@ const (
        // wait for Run.
        whatsappRequests = 8
 
        // 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."
 )
        // 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{
 // 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
 
        }
 
        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
        }
        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 {
 // 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
 
        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.disconnectedAt = w.clock()
                w.disconnections++
                disconnections := w.disconnections

-               w.timer = time.AfterFunc(w.noticeDelay, func() {
+               w.timer = time.AfterFunc(whatsappNoticeDelay, func() {

                        w.disconnectedLong(disconnections)
                })
        }
                        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
 // 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
 //
 // 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 e936f45c4af856545e13849b0adcf6d78ad57b97..9f80b382dfb096d91fe4cb18638e7faef45684f8 100644 (file)
--- a/whatsapp_fake_helper_test.go
+++ b/whatsapp_fake_helper_test.go
@@ -9,6 +9,7 @@ import (
        "slices"
        "sync"
        "testing"
        "slices"
        "sync"
        "testing"

+       "testing/synctest"

        "time"
 
        "github.com/stretchr/testify/require"
        "time"
 
        "github.com/stretchr/testify/require"


@@ -282,6 +283,34 @@ type WhatsAppFakeSend struct {
        id   string
 }
 
        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.
 // 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
 }
 
 // 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()
 
 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())
        whatsappTestSession(t, dir, true)
 
        whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock())

-       whatsapp.state.quiet = 50 * time.Millisecond

 
        return whatsapp, fake
 }
 
        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"))
 }
        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 ec5c892309e4aa282412730f5be35f95125acdb0..4d6c7e588be6501bb4c7e06745ba2fb6a427e892 100644 (file)
--- a/whatsapp_groups.go
+++ b/whatsapp_groups.go
@@ -9,6 +9,10 @@ import (
 )
 
 const (
 )
 
 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
        // 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
        }
 
                return false
        }
 

-       ctx, cancel := context.WithTimeout(ctx, whatsappLookupTime)
+       ctx, cancel := context.WithTimeout(ctx, whatsappGroupsLookupTime)

        defer cancel()
 
        groups, err := link.client.GetJoinedGroups(ctx)
        defer cancel()
 
        groups, err := link.client.GetJoinedGroups(ctx)

diff --git a/whatsapp_image.go b/whatsapp_image.go
index 4277b0aec60e1e79a888a56faa0d4a35465103cc..d4b88d8303a6c2a2aed12fb59d780fbb4825acbc 100644 (file)
--- a/whatsapp_image.go
+++ b/whatsapp_image.go
@@ -4,15 +4,11 @@ import (
        "context"
        "errors"
        "fmt"
        "context"
        "errors"
        "fmt"

-       "time"

 
        "go.mau.fi/whatsmeow"
        "go.mau.fi/whatsmeow/proto/waE2E"
 )
 
 
        "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.
 // 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 919d6560c4470a1e3af407c987db4f904935141a..91af70a921e64cbd63e26e88111527d14d29df10 100644 (file)
--- a/whatsapp_image_test.go
+++ b/whatsapp_image_test.go
@@ -80,28 +80,28 @@ func TestWhatsAppImagesAreDurableBeforeAcknowledgement(t *testing.T) {
 }
 
 func TestWhatsAppImageDownloadCanOutlastNameLookup(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) {
 }
 
 func TestWhatsAppRunCancellationEndsAnImageDownload(t *testing.T) {

diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go
index 06f1c558d03b930bd52f597c4bd61e49806ca9e5..392593fda5109d1f4ba9f921b9baa6eb1d169562 100644 (file)
--- a/whatsapp_inbox.go
+++ b/whatsapp_inbox.go
@@ -16,6 +16,13 @@ import (
 )
 
 const (
 )
 
 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"
 
        // 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{
 // 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
 //
 // 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
 //
 // 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 {
 // 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
 
        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
 }
 
 // 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()
 func (i *WhatsAppInbox) startCatchup() {
        i.mutex.Lock()
        defer i.mutex.Unlock()


@@ -218,7 +219,7 @@ func (i *WhatsAppInbox) startCatchup() {
                i.catchupTimer.Stop()
        }
 
                i.catchupTimer.Stop()
        }
 

-       i.catchupTimer = time.AfterFunc(i.catchupLimit, func() {
+       i.catchupTimer = time.AfterFunc(whatsappCatchupLimit, func() {

                i.caughtUpBy(catchups)
        })
 }
                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 {
 // 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)
        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 {
        }
 
        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)
                defer cancel()
 
                message.Text = i.imageText(ctx, link, event, message.Text)

diff --git a/whatsapp_inbox_test.go b/whatsapp_inbox_test.go
index cea65fec15ca501cf6b62b97bd96de14efc38a08..a0a0f9dfc4e61cd53139907edb994acf3f124799 100644 (file)
--- a/whatsapp_inbox_test.go
+++ b/whatsapp_inbox_test.go
@@ -3,6 +3,7 @@ package main
 import (
        "net/http"
        "testing"
 import (
        "net/http"
        "testing"

+       "testing/synctest"

        "time"
 
        "github.com/stretchr/testify/require"
        "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))
                        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 {
 
                        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 {
 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) {
 }
 
 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) {
 }
 
 func TestWhatsAppStateOfTheStateFileWaitForTheCatchup(t *testing.T) {

diff --git a/whatsapp_route_test.go b/whatsapp_route_test.go
index 9c855c9935b4b61d3a4e9da64acad8d6e2481221..c6dbf447fee99a9cf7fc1d259c4689b091b212d2 100644 (file)
--- 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()
 
 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())
        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 469912b0e0806adcb4c80712de0a82498c1181cb..480e7dba35cbc6c3f09e1400aa9f1d75c21207bf 100644 (file)
--- a/whatsapp_send_test.go
+++ b/whatsapp_send_test.go
@@ -39,7 +39,7 @@ func TestWhatsAppSendSendsOnceForEachID(t *testing.T) {
                },
        }, fake.Sent())
 
                },
        }, 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")
        require.NoError(t, err)
 
        record, found := again.Send("keys")

diff --git a/whatsapp_state.go b/whatsapp_state.go
index c09423d83d6e73b97ede0feb152f8f701006a391..41bd2f0b50b81e7bc90a812fc6d7947f07fb58ae 100644 (file)
--- 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
 )
 
 // 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{
        s := &WhatsAppState{

-               path:   path,
-               quiet:  quiet,
-               window: window,
+               path: path,

        }
 
        data, err := os.ReadFile(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
 // 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
 //
 // 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 {
 // 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
 
        mutex sync.Mutex
        file  WhatsAppStateFile


@@ -102,7 +99,7 @@ func (s *WhatsAppState) Add(chat WhatsAppChat, message WhatsAppMessage, arrival
                        return false, nil
                }
 
                        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
                        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 {
 
 // 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
        }
        if window.Before(quiet) {
                return window
        }

diff --git a/whatsapp_state_test.go b/whatsapp_state_test.go
index 2631b4d81caa90d2f7434932bfa21d7db001c8a9..1cfedeef567521bb4f1f18226cec6ed21cbe390f 100644 (file)
--- 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))
 
        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)
 }
 
        require.ErrorContains(t, err, path)
 }
 


@@ -159,7 +159,7 @@ func TestWhatsAppStateKeepsTheLastSendRecords(t *testing.T) {
 func whatsappTestState(t *testing.T, path string) *WhatsAppState {
        t.Helper()
 
 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
        require.NoError(t, err)
 
        return batches

diff --git a/whatsapp_test.go b/whatsapp_test.go
index 96ec2df0031cb169310fb99324046ce8e9698bf4..f098c908012013298260094afa083c716ea4217b 100644 (file)
--- a/whatsapp_test.go
+++ b/whatsapp_test.go
@@ -1,7 +1,6 @@
 package main
 
 import (
 package main
 
 import (

-       "context"

        "errors"
        "os"
        "path/filepath"
        "errors"
        "os"
        "path/filepath"


@@ -606,34 +605,33 @@ func TestWhatsAppLogoutFromThePhone(t *testing.T) {
 }
 
 func TestWhatsAppLongDisconnectionGivesOneNotice(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) {
 }
 
 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"}`))
 }
        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,
-       }
-}