owner: owner,
clock: clock,
logger: logger,
- check: scheduleCheck,
changed: make(chan struct{}, 1),
}
owner string
clock func() time.Time
logger *slog.Logger
- check time.Duration
changed chan struct{}
mutex sync.Mutex
}
// 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)
wait, found := s.Wait(s.clock())
if found {
- timer = time.NewTimer(min(wait, s.check))
+ timer = time.NewTimer(min(wait, scheduleCheck))
timeout = timer.C
}
"strings"
"sync"
"testing"
+ "testing/synctest"
"time"
"github.com/stretchr/testify/require"
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())
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)
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{
}
// 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)
)
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"
owner: owner,
offsetPath: offsetPath,
logger: logger,
- minWait: time.Second,
- maxWait: time.Minute,
ready: make(chan struct{}, 1),
}
}
// 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
back: "image receive back",
}
- imageWait := b.minWait
- wait := b.minWait
+ imageWait := telegramRetryDelay
+ wait := telegramRetryDelay
poll:
for ctx.Err() == nil {
outage.End()
- wait = b.minWait
+ wait = telegramRetryDelay
for _, update := range updates {
if ctx.Err() != nil {
imageOutage.End()
- imageWait = b.minWait
+ imageWait = telegramRetryDelay
offset = update.UpdateID + 1
err = telegramSaveOffset(b.offsetPath, offset)
}
// 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)
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
// 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)
// 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)
// 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{},
}
}
}
// 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,
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)
}
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)
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))
}
}()
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 {
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,
// 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)
"net/http"
"net/http/httptest"
"testing"
+ "testing/synctest"
"time"
"github.com/stretchr/testify/require"
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))
+ })
})
}
}
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 {
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{
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)
package main
import (
+ "context"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
"testing"
+ "testing/synctest"
"time"
"github.com/stretchr/testify/require"
})
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")
}
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) {
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",
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) {
}
// 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
}
--- /dev/null
+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
+}
// 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."
)
// 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
}
// 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
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)
})
}
// 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
"slices"
"sync"
"testing"
+ "testing/synctest"
"time"
"github.com/stretchr/testify/require"
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.
}
// 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()
whatsappTestSession(t, dir, true)
whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock())
- whatsapp.state.quiet = 50 * time.Millisecond
return whatsapp, fake
}
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,
+ }
+}
)
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
return false
}
- ctx, cancel := context.WithTimeout(ctx, whatsappLookupTime)
+ ctx, cancel := context.WithTimeout(ctx, whatsappGroupsLookupTime)
defer cancel()
groups, err := link.client.GetJoinedGroups(ctx)
"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.
}
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) {
)
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"
// 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),
}
}
//
// 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
// 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
}
// 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()
i.catchupTimer.Stop()
}
- i.catchupTimer = time.AfterFunc(i.catchupLimit, func() {
+ i.catchupTimer = time.AfterFunc(whatsappCatchupLimit, func() {
i.caughtUpBy(catchups)
})
}
// 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)
}
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)
import (
"net/http"
"testing"
+ "testing/synctest"
"time"
"github.com/stretchr/testify/require"
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 {
}
}
+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 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 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())
},
}, 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")
)
// 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)
// 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
// 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
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
// 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
}
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)
}
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
package main
import (
- "context"
"errors"
"os"
"path/filepath"
}
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) {
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,
- }
-}
projects / onechat / commitdiff