``15550002222@s.whatsapp.net``, and the ID of a group can hold the
number of the person who made it.
+While the daemon is stopped or disconnected, the WhatsApp servers keep
+the messages for the linked device and send them at the next connection.
+They keep them only for a limited time, so a long stop can lose
+messages. These messages form batches by their original times, as live
+messages do. The daemon waits for the end of this catch-up, or one
+minute at the most, and then gives all its batches to the agent in one
+turn, not one turn for each chat. The batches of ``whatsapp/state.json``
+at the start wait for the first catch-up in the same way.
+
Each batch of a group goes to the chat, also with no turn, and the
memory tree summarizes it with model calls, which use the quota.
// Submit does. A failed write stops the loop, because the log refuses all
// writes after it.
func (l *TurnLoop) Report(name, text, status string) {
- _, err := l.Work(name, text, status, status != subagentStopped)
- if err != nil {
+ message, err := l.Work(name, text, status)
+ switch {
+ case err != nil:
l.fail(fmt.Errorf("%w: %w", errChainLogWrite, err))
+ case status != subagentStopped:
+ l.Wake(message)
}
}
// Work logs text as a work message of source, such as the name of a subagent,
-// and tells the clients, with status. With wake, the message joins the waiting
-// messages and wakes the loop, as a user message of Submit does: it joins the
-// first request of a new turn, or the next request of a running turn after the
-// tool results.
-func (l *TurnLoop) Work(source, text, status string, wake bool) (Message, error) {
+// and tells the clients, with status. The message starts no turn until it goes
+// to Wake.
+func (l *TurnLoop) Work(source, text, status string) (Message, error) {
message, err := l.memory.Append("work", text)
if err != nil {
return Message{}, err
Status: status,
})
- if !wake {
- return message, nil
+ return message, nil
+}
+
+// Wake adds the logged messages to the waiting messages and wakes the loop
+// once, as a user message of Submit does. They join the first request of a new
+// turn together, or the next request of a running turn after the tool results.
+func (l *TurnLoop) Wake(messages ...Message) {
+ if len(messages) == 0 {
+ return
}
l.mutex.Lock()
- l.pending = append(l.pending, message)
+ l.pending = append(l.pending, messages...)
l.mutex.Unlock()
select {
case l.wake <- struct{}{}:
default:
}
-
- return message, nil
}
// StopSubagent ends the run of the subagent name and tells if it ran. Its
// whatsappUnknownGroup is the name of a group whose name is not known.
whatsappUnknownGroup = "A group with no known name"
+
+ // whatsappCatchupLimit is the longest wait for the end of the catch-up
+ // after a connection.
+ whatsappCatchupLimit = time.Minute
)
// WhatsAppClient is the part of the WhatsApp client that the daemon uses. The
// the client of a device with its event handler.
func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logger *log.Logger, newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient) (*WhatsApp, error) {
w := &WhatsApp{
- dir: filepath.Join(dir, whatsappDirName),
- clock: clock,
- logger: logger,
- newClient: newClient,
- noticeDelay: whatsappNoticeDelay,
- requests: make(chan WhatsAppRequest, whatsappRequests),
- logouts: make(chan struct{}, 1),
- wake: make(chan struct{}, 1),
- failures: make(chan error, 1),
- groupNames: make(map[string]string),
+ dir: filepath.Join(dir, whatsappDirName),
+ clock: clock,
+ logger: logger,
+ newClient: newClient,
+ noticeDelay: whatsappNoticeDelay,
+ catchupLimit: whatsappCatchupLimit,
+ requests: make(chan WhatsAppRequest, whatsappRequests),
+ logouts: make(chan struct{}, 1),
+ wake: make(chan struct{}, 1),
+ failures: make(chan error, 1),
+ groupNames: make(map[string]string),
}
var err error
if w.device.ID != nil {
w.linked = true
+ w.catchingUp = true
w.own = whatsappOwnIDs(w.device.ID.User, w.device.LID.User)
}
//
// The text messages go into batches, one chat at a time, as WhatsAppBatches
// says. Run logs each closed batch as a work message that starts with
-// [WhatsApp], and a batch that needs the user wakes the turn loop. 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. A crash between the two
-// makes the batch repeat once.
+// [WhatsApp], and a batch that needs the user wakes the turn loop.
+//
+// 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, Run closes all batches, logs them, and wakes the turn loop
+// once, so the whole catch-up gives one turn and not one turn 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. A
+// crash between the two makes the batch repeat once.
//
// The library logs nothing, because its log lines hold the ID of the own
// account, which is the phone number of the user. The own number and the own
// log, and those of private chats hold phone numbers, because the agent needs
// them to send.
type WhatsApp struct {
- dir string
- clock func() time.Time
- logger *log.Logger
- newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient
- noticeDelay time.Duration
- requests chan WhatsAppRequest
- logouts chan struct{}
- wake chan struct{}
- failures chan error
- batches *WhatsAppBatches
+ dir string
+ clock func() time.Time
+ logger *log.Logger
+ newClient func(device *store.Device, handle func(event any) bool) WhatsAppClient
+ noticeDelay time.Duration
+ catchupLimit time.Duration
+ requests chan WhatsAppRequest
+ logouts chan struct{}
+ wake chan struct{}
+ failures chan error
+ batches *WhatsAppBatches
mutex sync.Mutex
container *sqlstore.Container
closed bool
groupNames map[string]string
groupsAt time.Time
+ catchingUp bool
+ caughtUp bool
+ catchups int
+ catchupTimer *time.Timer
}
// Run connects a linked device, logs the closed batches in l, runs the pair and
}()
for {
- err := w.deliver(ctx, l)
+ err := w.deliver(ctx, l, false)
if err != nil {
return err
}
w.timer.Stop()
}
+ if w.catchupTimer != nil {
+ w.catchupTimer.Stop()
+ }
+
client := w.client
w.client = nil
w.mutex.Unlock()
return nil
}
- err := w.deliverAll(ctx, l)
+ err := w.deliver(ctx, l, true)
if err != nil {
return err
}
return nil
}
- err := w.deliverAll(ctx, l)
+ err := w.deliver(ctx, l, true)
if err != nil {
return err
}
}
// deliver logs the closed batches in l in order, each as a work message of
-// whatsappSource that wakes the loop when the batch starts a turn. A batch
-// leaves the state file after its message is durable in the log. A group with
-// no name gets its name from WhatsApp first.
-func (w *WhatsApp) deliver(ctx context.Context, l *TurnLoop) error {
- _, err := w.batches.CloseDue(time.Now())
+// whatsappSource, and wakes the loop once with the batches that start a turn. A
+// batch leaves the state file after its message is durable in the log. A group
+// with no name gets its name from WhatsApp first. During a catch-up it logs
+// nothing, and at its end it closes all batches first. With all, it closes and
+// logs all batches also during a catch-up.
+func (w *WhatsApp) deliver(ctx context.Context, l *TurnLoop, all bool) error {
+ w.mutex.Lock()
+ catchingUp := w.catchingUp
+ caughtUp := w.caughtUp
+ w.caughtUp = false
+ w.mutex.Unlock()
+
+ var err error
+ switch {
+ case all || caughtUp:
+ err = w.batches.CloseAll()
+ case catchingUp:
+ return nil
+ default:
+ _, err = w.batches.CloseDue(time.Now())
+ }
+
if err != nil {
return err
}
+ var turns []Message
for _, batch := range w.batches.Closed() {
if batch.Chat.Group && batch.Chat.Name == "" {
batch.Chat.Name = w.groupName(ctx, batch.Chat.ID)
}
- _, err = l.Work(whatsappSource, batch.Text(w.clock().Location()), "", batch.Turn())
+ message, err := l.Work(whatsappSource, batch.Text(w.clock().Location()), "")
if err != nil {
return fmt.Errorf("log a WhatsApp batch: %w", err)
}
if err != nil {
return err
}
- }
-
- return nil
-}
-// deliverAll closes every open batch and logs all of them.
-func (w *WhatsApp) deliverAll(ctx context.Context, l *TurnLoop) error {
- err := w.batches.CloseAll()
- if err != nil {
- return err
+ if batch.Turn() {
+ turns = append(turns, message)
+ }
}
- return w.deliver(ctx, l)
+ l.Wake(turns...)
+
+ return nil
}
// groupName returns the name of the group id, or whatsappUnknownGroup. A group
w.linked = false
w.own = nil
w.connected = false
+ w.catchingUp = false
if w.timer != nil {
w.timer.Stop()
}
// handle takes one event of the library and tells if it was handled. A message
// goes to receive. The names of the groups follow the events of the groups. A
-// disconnection keeps the time of its start while 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.
+// connection starts a catch-up, which ends at OfflineSyncCompleted or after
+// catchupLimit. A disconnection keeps the time of its start while 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.
func (w *WhatsApp) handle(event any) bool {
message, isMessage := event.(*events.Message)
if isMessage {
case *events.Connected:
w.connected = true
w.connectedAt = w.clock()
+ w.catchingUp = true
+ w.catchups++
+ catchups := w.catchups
+ if w.catchupTimer != nil {
+ w.catchupTimer.Stop()
+ }
+
+ w.catchupTimer = time.AfterFunc(w.catchupLimit, func() {
+ w.caughtUpBy(catchups)
+ })
if w.timer != nil {
w.timer.Stop()
}
}
w.connected = false
+ case *events.OfflineSyncCompleted:
+ w.endCatchup()
case *events.LoggedOut:
select {
case w.logouts <- struct{}{}:
return chat, message, true
}
+// caughtUpBy ends the catch-up of the connection with the number catchups, when
+// no later connection started a new one.
+func (w *WhatsApp) caughtUpBy(catchups int) {
+ w.mutex.Lock()
+ defer w.mutex.Unlock()
+
+ if w.catchups == catchups {
+ w.endCatchup()
+ }
+}
+
+// endCatchup ends a running catch-up and wakes Run, which logs all batches. The
+// caller holds the mutex.
+func (w *WhatsApp) endCatchup() {
+ if !w.catchingUp {
+ return
+ }
+
+ w.catchingUp = false
+ w.caughtUp = true
+ if w.catchupTimer != nil {
+ w.catchupTimer.Stop()
+ }
+
+ select {
+ case w.wake <- struct{}{}:
+ default:
+ }
+}
+
// disconnectedLong gives the notice of a long disconnection, unless the
// disconnection with the number disconnections ended.
func (w *WhatsApp) disconnectedLong(disconnections int) {
socket, memory := daemonStartTestWithWhatsApp(t, server, bot, whatsapp)
client := terminalStartTestClient(t, socket)
client.WaitFor(t, "[connected to OneChat]")
- <-fake.connects
+ whatsappTestConnect(t, fake)
at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
require.True(t, fake.Emit(whatsappTestMessage(whatsappTestAna, whatsappTestAna, "m1", "Are you at home?", at)))
socket, memory := daemonStartTestWithWhatsApp(t, server, nil, whatsapp)
client := terminalStartTestClient(t, socket)
client.WaitFor(t, "[connected to OneChat]")
- <-fake.connects
+ whatsappTestConnect(t, fake)
event := whatsappTestMessage(whatsappTestFamily, whatsappTestBruno, "m1", "Who brings the cake?", time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC))
event.Info.PushName = "Bruno Lima"
return whatsapp, fake
}
+
+func TestWhatsAppCatchupGivesOneTurn(t *testing.T) {
+ whatsapp, fake := whatsappStartTestLinked(t)
+ server := newModelFake(t)
+ socket, memory := daemonStartTestWithWhatsApp(t, server, nil, whatsapp)
+ client := terminalStartTestClient(t, socket)
+ client.WaitFor(t, "[connected to OneChat]")
+ <-fake.connects
+
+ require.True(t, fake.Emit(&events.Connected{}))
+ at := time.Date(2026, 1, 1, 8, 0, 0, 0, time.UTC)
+ for index, event := range []*events.Message{
+ whatsappTestMessage(whatsappTestAna, whatsappTestAna, "m1", "Good morning", at),
+ whatsappTestMessage(whatsappTestBruno, whatsappTestBruno, "m2", "Call me", at.Add(time.Hour)),
+ whatsappTestMessage(whatsappTestAna, whatsappTestAna, "m3", "Are you there?", at.Add(2*time.Hour)),
+ } {
+ event.Info.PushName = []string{"Ana", "Bruno", "Ana"}[index]
+ require.True(t, fake.Emit(event))
+ }
+
+ time.Sleep(100 * time.Millisecond)
+ require.Zero(t, memory.Len())
+
+ require.True(t, fake.Emit(&events.OfflineSyncCompleted{}))
+ client.WaitFor(t, "[done]")
+ require.Equal(t, []string{
+ "[WhatsApp] Ana, private chat 15550002222@s.whatsapp.net, Thursday 2026-01-01\n08:00 Ana: Good morning\n" + whatsappBatchEmptyReply,
+ "[WhatsApp] Bruno, private chat 15550003333@s.whatsapp.net, Thursday 2026-01-01\n09:00 Bruno: Call me\n" + whatsappBatchEmptyReply,
+ "[WhatsApp] Ana, private chat 15550002222@s.whatsapp.net, Thursday 2026-01-01\n10:00 Ana: Are you there?\n" + whatsappBatchEmptyReply,
+ }, whatsappTestLogTexts(t, memory)[:3])
+
+ time.Sleep(100 * time.Millisecond)
+ require.Equal(t, 1, server.Calls())
+ for _, text := range []string{"Good morning", "Call me", "Are you there?"} {
+ require.Contains(t, string(server.Call(0).raw), text)
+ }
+}
+
+func TestWhatsAppCatchupEndsAfterItsLimit(t *testing.T) {
+ whatsapp, fake := whatsappStartTestLinked(t)
+ whatsapp.catchupLimit = 100 * time.Millisecond
+ socket, memory := 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(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)
+}
+
+func TestWhatsAppBatchesOfTheStateFileWaitForTheCatchup(t *testing.T) {
+ dir := t.TempDir()
+ whatsappTestSession(t, dir, true)
+ before, _, _ := whatsappStartTest(t, dir, newTestClock())
+ _, err := before.batches.Add(whatsappTestChat, WhatsAppMessage{
+ ID: "m1",
+ Sender: "Ana Souza",
+ Time: time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC),
+ Text: "Did you see my message?",
+ }, time.Now().Add(-time.Hour))
+ require.NoError(t, err)
+
+ whatsapp, fake, _ := whatsappStartTest(t, dir, newTestClock())
+ socket, memory := daemonStartTestWithWhatsApp(t, newModelFake(t), nil, whatsapp)
+ client := terminalStartTestClient(t, socket)
+ client.WaitFor(t, "[connected to OneChat]")
+ <-fake.connects
+
+ time.Sleep(100 * time.Millisecond)
+ require.Zero(t, memory.Len())
+
+ require.True(t, fake.Emit(&events.Connected{}))
+ require.True(t, fake.Emit(&events.OfflineSyncCompleted{}))
+ client.WaitFor(t, "12:00 Ana Souza: Did you see my message?")
+ client.WaitFor(t, "[done]")
+}
+
+// whatsappTestConnect waits for the connection of the daemon and gives it the
+// events of a connection with an empty catch-up.
+func whatsappTestConnect(t *testing.T, fake *WhatsAppFake) {
+ t.Helper()
+
+ <-fake.connects
+ require.True(t, fake.Emit(&events.Connected{}))
+ require.True(t, fake.Emit(&events.OfflineSyncCompleted{}))
+}
projects / onechat / commitdiff