Deliver routed WhatsApp batches to children

A routed batch enters the durable agent message queue as work and wakes
only its child. Saved log IDs prevent a second entry on restart, and a
receipt remains until inbox removal even if the child finishes first.
Receipts use batch numbers rather than a watermark because chats can
close out of number order.

Unrouted batches keep the root path. A missing child sends new inputs
and its queued external batches to the root. Routed events keep their
WhatsApp source without a subagent completion status, and daemon logs
name the destination without message text or chat IDs.

diff --git a/README.rst b/README.rst
index 36db09c371132db8ddb1455b901533b5935521dc..7912d0daa3c618ac8bc0f412ed8eeeacdcd8b624 100644 (file)
--- a/README.rst
+++ b/README.rst
@@ -497,9 +497,10 @@ 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
 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.
+minute at the most, and then gives all its unrouted batches to the root
+in one turn, not one turn for each chat. Routed batches wake their
+child. The batches of ``whatsapp/state.json`` at the start wait for the
+first catch-up in the same way.

 
 The agent finds a chat with the tool ``whatsapp_find``: it gives the
 contacts and the groups whose name holds a text, with no regard to case
 
 The agent finds a chat with the tool ``whatsapp_find``: it gives the
 contacts and the groups whose name holds a text, with no regard to case


@@ -524,7 +525,11 @@ The root manages source routes with ``route(source, agent)``,
 source is ``whatsapp:CHAT-ID``, with the ID from ``whatsapp_find`` or a
 ``[WhatsApp]`` message, and the agent must be its child. The root keeps
 the map in ``agent/routes.json`` with mode 0600. Children and subagents
 source is ``whatsapp:CHAT-ID``, with the ID from ``whatsapp_find`` or a
 ``[WhatsApp]`` message, and the agent must be its child. The root keeps
 the map in ``agent/routes.json`` with mode 0600. Children and subagents

-have no route tools. There is no command to change routes.
+have no route tools. There is no command to change routes. Each routed
+batch enters only its child's log and wakes it, including a group batch
+with no mention. The durable agent queue prevents a second log entry
+after a restart. An unrouted source or a missing child uses the root. A
+queued batch whose child is removed while stopped goes to the root.

 
 When the session holds a linked device, the daemon connects at its
 start. If the first connection fails, it tries again after 2 seconds,
 
 When the session holds a linked device, the daemon connects at its
 start. If the first connection fails, it tries again after 2 seconds,

diff --git a/daemon_message.go b/daemon_message.go
index 40a99a91f3bf76d8b5c5de314a18cb0d00cfeaef..8ee26e83b4a5d1cbcf5692cefe8d8331494a85cd 100644 (file)
--- a/daemon_message.go
+++ b/daemon_message.go
@@ -15,6 +15,7 @@ import (
 // openDaemonMessages connects the root and its children. It restores queued
 // messages before the loops start. A queue entry stays until its recipient
 // finishes the turn that takes it, so a stop can interrupt delivery or a turn.
 // openDaemonMessages connects the root and its children. It restores queued
 // messages before the loops start. A queue entry stays until its recipient
 // finishes the turn that takes it, so a stop can interrupt delivery or a turn.

+// A routed batch with a removed child goes to the root.

 func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
        routes, err := openRoutes(filepath.Join(loops[0].agent.dir, "routes.json"), loops[0].logger)
        if err != nil {
 func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
        routes, err := openRoutes(filepath.Join(loops[0].agent.dir, "routes.json"), loops[0].logger)
        if err != nil {


@@ -66,15 +67,32 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
                return nil, errors.New("invalid agent message sequence")
        }
 
                return nil, errors.New("invalid agent message sequence")
        }
 

+       for i, number := range m.state.WhatsAppBatches {
+               if number <= 0 || slices.Contains(m.state.WhatsAppBatches[:i], number) {
+                       return nil, errors.New("invalid WhatsApp batch receipt")
+               }
+       }
+

        previous := 0
        previous := 0

-       for _, message := range m.state.Pending {
+       for i := range m.state.Pending {
+               message := &m.state.Pending[i]

                source := m.loops[message.Source]
                target := m.loops[message.Target]
                source := m.loops[message.Source]
                target := m.loops[message.Target]

+               if message.WhatsAppBatch > 0 && source == loops[0] && (target == nil || (target != source && target.agent.config.Parent != message.Source)) {
+                       message.Target = message.Source
+                       message.LogID = nil
+                       target = source
+               }
+

                if source == nil || target == nil || message.ID <= previous || message.ID > m.state.Next || (message.LogID != nil && *message.LogID < 0) {
                        return nil, errors.New("invalid queued agent message")
                }
 
                if source == nil || target == nil || message.ID <= previous || message.ID > m.state.Next || (message.LogID != nil && *message.LogID < 0) {
                        return nil, errors.New("invalid queued agent message")
                }
 

-               if (message.Kind != "user" || target.agent.config.Parent != message.Source) && (message.Kind != "work" || source.agent.config.Parent != message.Target) {
+               if message.WhatsAppBatch != 0 {
+                       if message.WhatsAppBatch < 0 || message.Kind != "work" || source != loops[0] || (target != source && target.agent.config.Parent != message.Source) {
+                               return nil, errors.New("invalid queued WhatsApp batch")
+                       }
+               } else if (message.Kind != "user" || target.agent.config.Parent != message.Source) && (message.Kind != "work" || source.agent.config.Parent != message.Target) {

                        return nil, errors.New("queued agents are not parent and child")
                }
 
                        return nil, errors.New("queued agents are not parent and child")
                }
 


@@ -91,10 +109,11 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) {
        return m, nil
 }
 
        return m, nil
 }
 

-// DaemonMessages owns the durable queue between the root and its children. The
-// mutex serializes queue writes and delivery. Backups take it before the locks
-// of the loops, memory, and subagents. A stored log ID makes delivery
-// idempotent, including a stop between the log append and the wake.
+// DaemonMessages owns the durable queue between the root and its children and
+// the routed inputs. The mutex serializes queue writes and delivery. Backups
+// take it before the locks of the loops, memory, and subagents. A stored log ID
+// makes delivery idempotent, including a stop between the log append and the
+// wake.

 type DaemonMessages struct {
        mutex   sync.Mutex
        path    string
 type DaemonMessages struct {
        mutex   sync.Mutex
        path    string


@@ -118,7 +137,10 @@ func (m *DaemonMessages) deliver(message *DaemonMessage) error {
        }
 
        status := ""
        }
 
        status := ""

-       if message.Kind == "work" {
+       source := message.Source
+       if message.WhatsAppBatch > 0 {
+               source = whatsappSource
+       } else if message.Kind == "work" {

                status = subagentFinished
        }
 
                status = subagentFinished
        }
 


@@ -126,11 +148,16 @@ func (m *DaemonMessages) deliver(message *DaemonMessage) error {
                Kind:   message.Kind,
                ID:     logged.ID,
                Text:   message.Text,
                Kind:   message.Kind,
                ID:     logged.ID,
                Text:   message.Text,

-               Source: message.Source,
+               Source: source,

                Status: status,
        })
        loop.Wake(logged)
                Status: status,
        })
        loop.Wake(logged)

-       loop.logger.Info("message received", "from", message.Source, "id", message.ID)
+
+       if message.WhatsAppBatch > 0 {
+               loop.logger.Info("batch routed", "kind", whatsappSource, "id", message.WhatsAppBatch, "to", message.Target)
+       } else {
+               loop.logger.Info("message received", "from", message.Source, "id", message.ID)
+       }

 
        return nil
 }
 
        return nil
 }


@@ -257,6 +284,82 @@ func (m *DaemonMessages) Tell(parent, child, text string) error {
        return nil
 }
 
        return nil
 }
 

+// RouteWhatsApp queues a routed batch as work and wakes its child. It returns
+// false for an unrouted source or a missing child. A saved receipt survives
+// turn acknowledgement until the inbox removes the batch. A repeat after a stop
+// between queueing and inbox removal cannot append a second entry.
+func (m *DaemonMessages) RouteWhatsApp(batch WhatsAppBatch, text string) (bool, error) {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       if m.failure != nil {
+               return false, fmt.Errorf("agent message queue stopped: %w", m.failure)
+       }
+
+       if batch.Number <= 0 {
+               return false, errors.New("invalid WhatsApp batch number")
+       }
+
+       if slices.Contains(m.state.WhatsAppBatches, batch.Number) {
+               return true, nil
+       }
+
+       root := m.ordered[0].agent.config.Name
+       name := m.routes.Target(whatsappSource + ":" + batch.Chat.ID)
+       child := m.loops[name]
+       if child == nil || child.agent.config.Parent != root {
+               return false, nil
+       }
+
+       m.add(root, name, "work", text)
+
+       message := &m.state.Pending[len(m.state.Pending)-1]
+       message.WhatsAppBatch = batch.Number
+       m.state.WhatsAppBatches = append(m.state.WhatsAppBatches, batch.Number)
+       err := m.save()
+       if err == nil {
+               err = m.deliver(message)
+       }
+
+       if err != nil {
+               return false, fmt.Errorf("route WhatsApp batch: %w", m.failed(err))
+       }
+
+       return true, nil
+}
+
+// PruneWhatsApp drops receipts only for batches absent from the durable inbox.
+// Queued turns keep their own log IDs until they finish. Receipts left by a
+// stop after inbox removal are removed by the next inbox delivery.
+func (m *DaemonMessages) PruneWhatsApp(batches []WhatsAppBatch) error {
+       m.mutex.Lock()
+       defer m.mutex.Unlock()
+
+       if m.failure != nil {
+               return fmt.Errorf("agent message queue stopped: %w", m.failure)
+       }
+
+       held := make(map[int]bool, len(batches))
+       for _, batch := range batches {
+               held[batch.Number] = true
+       }
+
+       before := len(m.state.WhatsAppBatches)
+       m.state.WhatsAppBatches = slices.DeleteFunc(m.state.WhatsAppBatches, func(number int) bool {
+               return !held[number]
+       })
+       if len(m.state.WhatsAppBatches) == before {
+               return nil
+       }
+
+       err := m.save()
+       if err != nil {
+               return fmt.Errorf("prune WhatsApp receipts: %w", m.failed(err))
+       }
+
+       return nil
+}
+

 // add appends a queue entry under the mutex. Its ID is never reused.
 func (m *DaemonMessages) add(source, target, kind, text string) {
        m.state.Next++
 // add appends a queue entry under the mutex. Its ID is never reused.
 func (m *DaemonMessages) add(source, target, kind, text string) {
        m.state.Next++


@@ -339,19 +442,24 @@ func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string)
        return nil
 }
 
        return nil
 }
 

-// DaemonMessageFile holds the sequence and the unacknowledged deliveries.
+// DaemonMessageFile holds the sequence, unacknowledged deliveries, and receipts
+// for WhatsApp batches until the inbox removes them. A receipt stays when its
+// turn finishes before inbox removal.

 type DaemonMessageFile struct {
 type DaemonMessageFile struct {

-       Next    int             `json:"next"`
-       Pending []DaemonMessage `json:"pending"`
+       Next            int             `json:"next"`
+       Pending         []DaemonMessage `json:"pending"`
+       WhatsAppBatches []int           `json:"whatsapp_batches,omitempty"`

 }
 
 // DaemonMessage is one queued delivery. LogID is set before the log append. A
 }
 
 // DaemonMessage is one queued delivery. LogID is set before the log append. A

-// nil LogID means that delivery has not started.
+// nil LogID means that delivery has not started. WhatsAppBatch marks a routed
+// external batch, whose event source is WhatsApp rather than its queue sender.

 type DaemonMessage struct {
 type DaemonMessage struct {

-       ID     int    `json:"id"`
-       Source string `json:"source"`
-       Target string `json:"target"`
-       Kind   string `json:"kind"`
-       Text   string `json:"text"`
-       LogID  *int   `json:"log_id,omitempty"`
+       ID            int    `json:"id"`
+       Source        string `json:"source"`
+       Target        string `json:"target"`
+       Kind          string `json:"kind"`
+       Text          string `json:"text"`
+       LogID         *int   `json:"log_id,omitempty"`
+       WhatsAppBatch int    `json:"whatsapp_batch,omitempty"`

 }
 }

diff --git a/whatsapp_inbox.go b/whatsapp_inbox.go
index c7fbf51ff4ece235dd6b39de109818b6b4ce90e4..adff79e698f22d28bb1ce388298c90839d263cb7 100644 (file)
--- a/whatsapp_inbox.go
+++ b/whatsapp_inbox.go
@@ -42,13 +42,15 @@ func newWhatsAppInbox(state *WhatsAppState, clock func() time.Time, groups *What
 // After each connection, the servers send the messages that came while the
 // daemon was away, and then OfflineSyncCompleted. Until then, or until
 // catchupLimit passes, the closed batches wait in the state file. At the end of
 // 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, 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 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


@@ -87,6 +89,8 @@ func (i *WhatsAppInbox) due() <-chan time.Time {
 // with no name gets its name from WhatsApp first, with the client of link.
 // 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.
 // with no name gets its name from WhatsApp first, with the client of link.
 // 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.

+// Routed batches enter the durable child queue instead of the root log and wake
+// their child even when the batch does not need a root turn.

 func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppLink, all bool) error {
        i.mutex.Lock()
 
 func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppLink, all bool) error {
        i.mutex.Lock()
 


@@ -110,14 +114,39 @@ func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppL
                return err
        }
 
                return err
        }
 

+       batches := i.state.Closed()
+       if l.messages != nil {
+               err = l.messages.PruneWhatsApp(batches)
+               if err != nil {
+                       return fmt.Errorf("prune WhatsApp receipts: %w", err)
+               }
+       }
+

        var turns []Message
 
        var turns []Message
 

-       for _, batch := range i.state.Closed() {
+       for _, batch := range batches {

                if batch.Chat.Group && batch.Chat.Name == "" {
                        batch.Chat.Name = i.groups.name(ctx, link, batch.Chat.ID)
                }
 
                if batch.Chat.Group && batch.Chat.Name == "" {
                        batch.Chat.Name = i.groups.name(ctx, link, batch.Chat.ID)
                }
 

-               message, err := l.Work(whatsappSource, batch.Text(i.clock().Location()), "")
+               text := batch.Text(i.clock().Location())
+               if l.messages != nil {
+                       routed, err := l.messages.RouteWhatsApp(batch, text)
+                       if err != nil {
+                               return fmt.Errorf("route a WhatsApp batch: %w", err)
+                       }
+
+                       if routed {
+                               err = i.state.Remove(batch.Number)
+                               if err != nil {
+                                       return fmt.Errorf("remove routed WhatsApp batch: %w", err)
+                               }
+
+                               continue
+                       }
+               }
+
+               message, err := l.Work(whatsappSource, text, "")

                if err != nil {
                        return fmt.Errorf("log a WhatsApp batch: %w", err)
                }
                if err != nil {
                        return fmt.Errorf("log a WhatsApp batch: %w", err)
                }


@@ -146,6 +175,13 @@ func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppL
 
        l.Wake(turns...)
 
 
        l.Wake(turns...)
 

+       if l.messages != nil {
+               err = l.messages.PruneWhatsApp(i.state.Closed())
+               if err != nil {
+                       return fmt.Errorf("prune WhatsApp receipts: %w", err)
+               }
+       }
+

        return nil
 }
 
        return nil
 }
 

diff --git a/whatsapp_route_test.go b/whatsapp_route_test.go
new file mode 100644 (file)
index 0000000..92c9cf3
--- /dev/null
+++ b/whatsapp_route_test.go
@@ -0,0 +1,410 @@
+package main
+
+import (
+       "encoding/json"
+       "net/http"
+       "os"
+       "path/filepath"
+       "testing"
+       "time"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) {
+       testCases := []struct {
+               name string
+               chat WhatsAppChat
+       }{
+               {
+                       name: "private chat",
+                       chat: WhatsAppChat{
+                               ID:   whatsappTestAna.String(),
+                               Name: "Ana",
+                       },
+               },
+               {
+                       name: "group without a mention",
+                       chat: WhatsAppChat{
+                               ID:    whatsappTestFamily.String(),
+                               Name:  "Garden plans",
+                               Group: true,
+                       },
+               },
+       }
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       server := newModelFake(t)
+                       server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+                               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+                       }
+                       dir := t.TempDir()
+                       test := newDaemonMessageTest(t, dir, server)
+                       inbox := whatsappRouteTestInbox(t, dir)
+                       require.NoError(t, test.messages.routes.Set("whatsapp:"+tc.chat.ID, "Garden"))
+                       whatsappRouteTestAdd(t, inbox, tc.chat, "The gate needs a new latch.")
+                       test.Start(t, 0, 1, 2)
+                       require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
+                       test.events[1].WaitFor(t, "idle", 1)
+                       require.Equal(t, 1, server.Calls())
+                       require.Equal(t, "gpt-test-garden", server.Call(0).body["model"])
+                       require.Contains(t, string(server.Call(0).raw), "The gate needs a new latch.")
+                       require.Zero(t, test.agents[0].memory.Len())
+                       require.Empty(t, test.events[0].Kinds())
+                       require.Zero(t, test.agents[2].memory.Len())
+                       require.False(t, test.loops[0].waiting())
+                       require.Equal(t, whatsappSource, test.events[1].LastEvent("work").Source)
+                       require.Empty(t, test.events[1].LastEvent("work").Status)
+                       require.Equal(t, 1, test.agents[1].memory.Len())
+                       require.Empty(t, inbox.state.Closed())
+                       require.Empty(t, test.Queue(t).Pending)
+                       require.Contains(t, test.logs.String(), "batch routed name=Garden kind=whatsapp id=1 to=Garden")
+                       require.NotContains(t, test.logs.String(), tc.chat.ID)
+                       require.NotContains(t, test.logs.String(), "new latch")
+               })
+       }
+}
+
+func TestWhatsAppUnrouteRestoresRootDelivery(t *testing.T) {
+       dir := t.TempDir()
+       test := newDaemonMessageTest(t, dir, newModelFake(t))
+       inbox := whatsappRouteTestInbox(t, dir)
+       chat := WhatsAppChat{
+               ID:   whatsappTestAna.String(),
+               Name: "Ana",
+       }
+       source := "whatsapp:" + chat.ID
+       require.NoError(t, test.messages.routes.Set(source, "Garden"))
+       whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
+       require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
+       require.Zero(t, test.agents[0].memory.Len())
+       require.True(t, test.loops[1].waiting())
+       require.NoError(t, test.messages.routes.Set(source, ""))
+       whatsappRouteTestAdd(t, inbox, chat, "Please check the other plan.")
+       require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
+       require.Equal(t, 1, test.agents[0].memory.Len())
+       require.Equal(t, 1, test.agents[1].memory.Len())
+       require.True(t, test.loops[0].waiting())
+       require.Contains(t, test.events[0].Last("work"), "Please check the other plan.")
+       require.NotContains(t, test.events[0].Last("work"), "Keep the gate plan.")
+       require.Empty(t, inbox.state.Closed())
+}
+
+func TestMissingWhatsAppRouteChildFallsBackToRoot(t *testing.T) {
+       dir := t.TempDir()
+       test := newDaemonMessageTest(t, dir, newModelFake(t))
+       inbox := whatsappRouteTestInbox(t, dir)
+       chat := WhatsAppChat{
+               ID:   whatsappTestAna.String(),
+               Name: "Ana",
+       }
+       require.NoError(t, test.messages.routes.Set("whatsapp:"+chat.ID, "Missing"))
+       whatsappRouteTestAdd(t, inbox, chat, "Please check the plan.")
+       require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
+       require.Equal(t, 1, test.agents[0].memory.Len())
+       require.True(t, test.loops[0].waiting())
+       require.Zero(t, test.agents[1].memory.Len())
+       require.False(t, test.loops[1].waiting())
+       require.Empty(t, test.events[0].LastEvent("work").Status)
+       require.Equal(t, whatsappSource, test.events[0].LastEvent("work").Source)
+       require.Empty(t, inbox.state.Closed())
+}
+
+func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) {
+       testCases := []struct {
+               name   string
+               finish bool
+       }{
+               {
+                       name:   "queued turn resumes",
+                       finish: false,
+               },
+               {
+                       name:   "finished turn stays finished",
+                       finish: true,
+               },
+       }
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       dir := t.TempDir()
+                       server := newModelFake(t)
+                       server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+                               modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+                       }
+                       first := newDaemonMessageTest(t, dir, server)
+                       inbox := whatsappRouteTestInbox(t, dir)
+                       chat := WhatsAppChat{
+                               ID:   whatsappTestAna.String(),
+                               Name: "Ana",
+                       }
+                       source := "whatsapp:" + chat.ID
+                       require.NoError(t, first.messages.routes.Set(source, "Garden"))
+                       whatsappRouteTestAdd(t, inbox, chat, "Remember the gate plan.")
+                       require.NoError(t, inbox.state.CloseAll())
+
+                       batch := inbox.state.Closed()[0]
+                       routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
+                       require.NoError(t, err)
+                       require.True(t, routed)
+
+                       if tc.finish {
+                               first.Start(t, 1)
+                               first.events[1].WaitFor(t, "idle", 1)
+                               require.Empty(t, first.Queue(t).Pending)
+                       }
+
+                       first.Close(t)
+
+                       second := newDaemonMessageTest(t, dir, server)
+                       require.Equal(t, source+" -> Garden", second.messages.routes.List())
+                       require.Equal(t, 1, second.agents[1].memory.Len())
+                       require.Equal(t, !tc.finish, second.loops[1].waiting())
+                       require.NoError(t, second.messages.routes.Set(source, ""))
+
+                       restored := whatsappRouteTestInbox(t, dir)
+                       require.Len(t, restored.state.Closed(), 1)
+                       require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
+                       require.Empty(t, restored.state.Closed())
+                       require.Zero(t, second.agents[0].memory.Len())
+                       require.Equal(t, 1, second.agents[1].memory.Len())
+
+                       if !tc.finish {
+                               second.Start(t, 1)
+                               second.events[1].WaitFor(t, "idle", 1)
+                       }
+
+                       require.Equal(t, 1, server.Calls())
+                       require.Empty(t, second.Queue(t).Pending)
+                       second.Close(t)
+
+                       third := newDaemonMessageTest(t, dir, server)
+                       require.False(t, third.loops[1].waiting())
+                       require.Empty(t, third.events[1].Kinds())
+                       require.Equal(t, 1, third.agents[1].memory.Len())
+               })
+       }
+}
+
+func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) {
+       testCases := []struct {
+               name string
+               id   *int
+       }{
+               {
+                       name: "before log ID is stored",
+                       id:   nil,
+               },
+               {
+                       name: "after log ID is stored",
+                       id:   new(0),
+               },
+       }
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       dir := t.TempDir()
+                       server := newModelFake(t)
+                       first := newDaemonMessageTest(t, dir, server)
+                       path := first.messages.path
+                       first.Close(t)
+
+                       file := DaemonMessageFile{
+                               Next:            1,
+                               WhatsAppBatches: []int{1},
+                               Pending: []DaemonMessage{{
+                                       ID:            1,
+                                       Source:        "OneChat",
+                                       Target:        "Garden",
+                                       Kind:          "work",
+                                       Text:          "[WhatsApp] Keep the gate plan.",
+                                       LogID:         tc.id,
+                                       WhatsAppBatch: 1,
+                               }},
+                       }
+                       data, err := json.Marshal(file)
+                       require.NoError(t, err)
+                       require.NoError(t, os.WriteFile(path, data, 0o600))
+
+                       second := newDaemonMessageTest(t, dir, server)
+                       require.Equal(t, 1, second.agents[1].memory.Len())
+                       require.True(t, second.loops[1].waiting())
+                       require.Equal(t, "[WhatsApp] Keep the gate plan.", second.events[1].Last("work"))
+                       require.Equal(t, whatsappSource, second.events[1].LastEvent("work").Source)
+                       second.Close(t)
+
+                       third := newDaemonMessageTest(t, dir, server)
+                       require.Equal(t, 1, third.agents[1].memory.Len())
+                       require.True(t, third.loops[1].waiting())
+                       require.Zero(t, third.agents[0].memory.Len())
+               })
+       }
+}
+
+func TestQueuedWhatsAppBatchUsesRootIfItsChildIsRemoved(t *testing.T) {
+       dir := t.TempDir()
+       server := newModelFake(t)
+       first := newDaemonMessageTest(t, dir, server)
+       inbox := whatsappRouteTestInbox(t, dir)
+       chat := WhatsAppChat{
+               ID:   whatsappTestAna.String(),
+               Name: "Ana",
+       }
+       require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
+       whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
+       require.NoError(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true))
+       first.Close(t)
+       require.NoError(t, os.RemoveAll(filepath.Join(dir, daemonAgentDirName, "children", "Garden")))
+
+       second := newDaemonMessageTest(t, dir, server)
+       require.Equal(t, 1, second.agents[0].memory.Len())
+       require.True(t, second.loops[0].waiting())
+       require.Equal(t, whatsappSource, second.events[0].LastEvent("work").Source)
+       require.Empty(t, second.events[0].LastEvent("work").Status)
+       require.Contains(t, second.events[0].Last("work"), "Keep the gate plan.")
+       require.False(t, second.loops[1].waiting())
+       second.Close(t)
+
+       third := newDaemonMessageTest(t, dir, server)
+       require.Equal(t, 1, third.agents[0].memory.Len())
+       require.True(t, third.loops[0].waiting())
+}
+
+func TestRoutedWhatsAppBatchesCanCloseOutOfNumberOrder(t *testing.T) {
+       dir := t.TempDir()
+       test := newDaemonMessageTest(t, dir, newModelFake(t))
+       inbox := whatsappRouteTestInbox(t, dir)
+       first := WhatsAppChat{
+               ID:   whatsappTestAna.String(),
+               Name: "Ana",
+       }
+       second := WhatsAppChat{
+               ID:    whatsappTestFamily.String(),
+               Name:  "Garden plans",
+               Group: true,
+       }
+       require.NoError(t, test.messages.routes.Set("whatsapp:"+first.ID, "Garden"))
+       require.NoError(t, test.messages.routes.Set("whatsapp:"+second.ID, "Garden"))
+
+       now := time.Now()
+       _, err := inbox.state.Add(first, WhatsAppMessage{
+               ID:     "first",
+               Text:   "The first chat is still open.",
+               Sender: "Ana",
+               Time:   now,
+       }, now)
+       require.NoError(t, err)
+
+       _, err = inbox.state.Add(second, WhatsAppMessage{
+               ID:     "second",
+               Text:   "The second chat is quiet.",
+               Sender: "Bruno",
+               Time:   now,
+       }, now.Add(-time.Minute))
+       require.NoError(t, err)
+       require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
+       require.Equal(t, 1, test.agents[1].memory.Len())
+       require.Contains(t, test.events[1].Last("work"), "The second chat is quiet.")
+       require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
+       require.Equal(t, 2, test.agents[1].memory.Len())
+       require.Contains(t, test.events[1].Last("work"), "The first chat is still open.")
+       require.Zero(t, test.agents[0].memory.Len())
+       require.Empty(t, test.Queue(t).WhatsAppBatches)
+}
+
+func TestWhatsAppCatchupSeparatesChildAndRootBatches(t *testing.T) {
+       dir := t.TempDir()
+       test := newDaemonMessageTest(t, dir, newModelFake(t))
+       inbox := whatsappRouteTestInbox(t, dir)
+       garden := WhatsAppChat{
+               ID:    whatsappTestFamily.String(),
+               Name:  "Garden plans",
+               Group: true,
+       }
+       library := WhatsAppChat{
+               ID:   whatsappTestBruno.String(),
+               Name: "Bruno",
+       }
+       root := WhatsAppChat{
+               ID:   whatsappTestAna.String(),
+               Name: "Ana",
+       }
+       require.NoError(t, test.messages.routes.Set("whatsapp:"+garden.ID, "Garden"))
+       require.NoError(t, test.messages.routes.Set("whatsapp:"+library.ID, "Library"))
+       inbox.waitForCatchup()
+       whatsappRouteTestAdd(t, inbox, garden, "Check the gate.")
+       whatsappRouteTestAdd(t, inbox, library, "Check the book.")
+       whatsappRouteTestAdd(t, inbox, root, "Check the mixed chat.")
+       require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
+
+       for _, agent := range test.agents {
+               require.Zero(t, agent.memory.Len())
+       }
+
+       inbox.endCatchup()
+       require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
+
+       for _, loop := range test.loops {
+               require.Equal(t, 1, loop.memory.Len())
+               require.True(t, loop.waiting())
+       }
+
+       require.Contains(t, test.events[0].Last("work"), "Check the mixed chat.")
+       require.Contains(t, test.events[1].Last("work"), "Check the gate.")
+       require.Contains(t, test.events[2].Last("work"), "Check the book.")
+       require.Empty(t, inbox.state.Closed())
+       require.Empty(t, test.Queue(t).WhatsAppBatches)
+}
+
+func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) {
+       dir := t.TempDir()
+       server := newModelFake(t)
+       first := newDaemonMessageTest(t, dir, server)
+       inbox := whatsappRouteTestInbox(t, dir)
+       chat := WhatsAppChat{
+               ID:   whatsappTestAna.String(),
+               Name: "Ana",
+       }
+       require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
+       require.NoError(t, os.Mkdir(first.messages.path, 0o700))
+       whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
+       require.ErrorContains(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true), "save agent messages")
+       require.Len(t, inbox.state.Closed(), 1)
+
+       for _, loop := range first.loops {
+               require.ErrorContains(t, loop.failed(), "save agent messages")
+               require.Zero(t, loop.memory.Len())
+       }
+
+       path := first.messages.path
+       first.Close(t)
+       require.NoError(t, os.Remove(path))
+
+       second := newDaemonMessageTest(t, dir, server)
+       restored := whatsappRouteTestInbox(t, dir)
+       require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
+       require.Equal(t, 1, second.agents[1].memory.Len())
+       require.True(t, second.loops[1].waiting())
+       require.Zero(t, second.agents[0].memory.Len())
+       require.Empty(t, restored.state.Closed())
+}
+
+func whatsappRouteTestInbox(t *testing.T, dir string) *WhatsAppInbox {
+       t.Helper()
+
+       state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName), whatsappBatchQuiet, whatsappBatchWindow)
+       require.NoError(t, err)
+
+       return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()))
+}
+
+func whatsappRouteTestAdd(t *testing.T, inbox *WhatsAppInbox, chat WhatsAppChat, text string) {
+       t.Helper()
+
+       at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
+       _, err := inbox.state.Add(chat, WhatsAppMessage{
+               ID:     text,
+               Sender: "Ana",
+               Text:   text,
+               Time:   at,
+       }, at)
+       require.NoError(t, err)
+}