From 0da1ef837bbd190cc5787214187d1b29d62b9410 Mon Sep 17 00:00:00 2001
From: "Anderson A. Giacomolli" <agiacomolli@gmail.com>
Date: Sun, 8 Nov 2026 12:03:17 -0300
Subject: [PATCH] Wait for the WhatsApp catch-up before a turn

After each connection, the WhatsApp servers send the messages that came
while the daemon was away, and then the event OfflineSyncCompleted.
Until that event, or one minute at the most, the daemon keeps the closed
batches in the state file and logs none. At the end it closes all
batches, logs them in order, and wakes the turn loop once with the
batches that need the user, so the whole catch-up gives one turn and not
one turn for each chat. The batches that the state file holds at the
start wait for the first catch-up in the same way.

TurnLoop.Work now only logs and notifies, and the new TurnLoop.Wake adds
several logged messages to the waiting messages at once. A report of a
subagent uses both, and the WhatsApp batches of one delivery join one
turn through one Wake.

Issue: 6f69ca1
---
 README.rst       |   9 +++
 turn.go          |  30 ++++----
 whatsapp.go      | 176 ++++++++++++++++++++++++++++++++++-------------
 whatsapp_test.go |  92 ++++++++++++++++++++++++-
 4 files changed, 245 insertions(+), 62 deletions(-)

diff --git a/README.rst b/README.rst
index b17cd64..b3787d2 100644
--- a/README.rst
+++ b/README.rst
@@ -237,6 +237,15 @@ private chat holds the phone number of the contact, such as
 ``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.
 
diff --git a/turn.go b/turn.go
index d6a1c41..e304714 100644
--- a/turn.go
+++ b/turn.go
@@ -122,18 +122,19 @@ func (l *TurnLoop) Submit(source, text string) (Message, error) {
 // 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
@@ -147,20 +148,25 @@ func (l *TurnLoop) Work(source, text, status string, wake bool) (Message, error)
 		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
diff --git a/whatsapp.go b/whatsapp.go
index 20143a9..8306734 100644
--- a/whatsapp.go
+++ b/whatsapp.go
@@ -81,6 +81,10 @@ const (
 
 	// 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
@@ -102,16 +106,17 @@ type WhatsAppClient interface {
 // 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
@@ -140,6 +145,7 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge
 
 	if w.device.ID != nil {
 		w.linked = true
+		w.catchingUp = true
 		w.own = whatsappOwnIDs(w.device.ID.User, w.device.LID.User)
 	}
 
@@ -161,11 +167,18 @@ func openWhatsApp(ctx context.Context, dir string, clock func() time.Time, logge
 //
 // 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
@@ -175,16 +188,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 {
-	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
@@ -202,6 +216,10 @@ type WhatsApp struct {
 	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
@@ -230,7 +248,7 @@ func (w *WhatsApp) Run(ctx context.Context, l *TurnLoop, notify func(Event)) err
 	}()
 
 	for {
-		err := w.deliver(ctx, l)
+		err := w.deliver(ctx, l, false)
 		if err != nil {
 			return err
 		}
@@ -406,6 +424,10 @@ func (w *WhatsApp) close() {
 		w.timer.Stop()
 	}
 
+	if w.catchupTimer != nil {
+		w.catchupTimer.Stop()
+	}
+
 	client := w.client
 	w.client = nil
 	w.mutex.Unlock()
@@ -422,7 +444,7 @@ func (w *WhatsApp) loggedOut(ctx context.Context, l *TurnLoop) error {
 		return nil
 	}
 
-	err := w.deliverAll(ctx, l)
+	err := w.deliver(ctx, l, true)
 	if err != nil {
 		return err
 	}
@@ -458,7 +480,7 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str
 		return nil
 	}
 
-	err := w.deliverAll(ctx, l)
+	err := w.deliver(ctx, l, true)
 	if err != nil {
 		return err
 	}
@@ -493,21 +515,39 @@ func (w *WhatsApp) unlink(ctx context.Context, l *TurnLoop, answer func(text str
 }
 
 // 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)
 		}
@@ -528,19 +568,15 @@ func (w *WhatsApp) deliver(ctx context.Context, l *TurnLoop) error {
 		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
@@ -594,6 +630,7 @@ func (w *WhatsApp) removeSession() error {
 	w.linked = false
 	w.own = nil
 	w.connected = false
+	w.catchingUp = false
 	if w.timer != nil {
 		w.timer.Stop()
 	}
@@ -759,10 +796,11 @@ func (w *WhatsApp) linkedBy(client WhatsAppClient) {
 
 // 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 {
@@ -783,6 +821,16 @@ func (w *WhatsApp) handle(event any) bool {
 	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()
 		}
@@ -802,6 +850,8 @@ func (w *WhatsApp) handle(event any) bool {
 		}
 
 		w.connected = false
+	case *events.OfflineSyncCompleted:
+		w.endCatchup()
 	case *events.LoggedOut:
 		select {
 		case w.logouts <- struct{}{}:
@@ -946,6 +996,36 @@ func (w *WhatsApp) text(ctx context.Context, device *store.Device, own []string,
 	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) {
diff --git a/whatsapp_test.go b/whatsapp_test.go
index 38a7b8d..6785bff 100644
--- a/whatsapp_test.go
+++ b/whatsapp_test.go
@@ -485,7 +485,7 @@ func TestWhatsAppPrivateBatchStartsATurnWithAnEmptyReply(t *testing.T) {
 	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)))
@@ -563,7 +563,7 @@ func TestWhatsAppGroupTurns(t *testing.T) {
 			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"
@@ -670,3 +670,91 @@ func whatsappStartTestLinked(t *testing.T) (*WhatsApp, *WhatsAppFake) {
 
 	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{}))
+}
-- 
2.53.0