Separate multiline Go statements

[onechat] / daemon_message_test.go

diff --git a/daemon_message_test.go b/daemon_message_test.go
index 060faf5e0a7d4310e043e40afe3208acd3f40038..27138d3f895fa57fb24bbe666fda7efecc2daa47 100644 (file)
--- a/daemon_message_test.go
+++ b/daemon_message_test.go
@@ -24,6 +24,7 @@ func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
        }
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
        }

+

        test := newDaemonMessageTest(t, t.TempDir(), server)
        test.Start(t, 0, 1, 2)
        require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
        test := newDaemonMessageTest(t, t.TempDir(), server)
        test.Start(t, 0, 1, 2)
        require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))


@@ -68,6 +69,7 @@ func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
        }
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
        }

+

        test := newDaemonMessageTest(t, t.TempDir(), server)
        telegram := newTelegramFake(t)
        bot := newTelegramBot(newTelegramAPI(telegram.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), discardLog())
        test := newDaemonMessageTest(t, t.TempDir(), server)
        telegram := newTelegramFake(t)
        bot := newTelegramBot(newTelegramAPI(telegram.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), discardLog())


@@ -97,6 +99,7 @@ func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
        require.Eventually(t, func() bool {
                return len(telegram.Sent()) >= 2
        }, 10*time.Second, time.Millisecond)
        require.Eventually(t, func() bool {
                return len(telegram.Sent()) >= 2
        }, 10*time.Second, time.Millisecond)

+

        require.Equal(t, []string{"[Garden] finished", "I will ask the user."}, telegram.Sent())
        require.Equal(t, subagentFinished, test.events[0].LastEvent("work").Status)
        require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
        require.Equal(t, []string{"[Garden] finished", "I will ask the user."}, telegram.Sent())
        require.Equal(t, subagentFinished, test.events[0].LastEvent("work").Status)
        require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))


@@ -130,12 +133,14 @@ func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.T) {
                        text: " \n\t ",
                },
        }
                        text: " \n\t ",
                },
        }

+

        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(tc.text)}, 100, 0, 5))
                        }
        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(tc.text)}, 100, 0, 5))
                        }

+

                        test := newDaemonMessageTest(t, t.TempDir(), server)
                        test.Start(t, 0, 1, 2)
                        require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
                        test := newDaemonMessageTest(t, t.TempDir(), server)
                        test.Start(t, 0, 1, 2)
                        require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))


@@ -162,6 +167,7 @@ func TestBusyChildQueuesTheNextParentMessage(t *testing.T) {
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
        }
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
        }

+

        test := newDaemonMessageTest(t, t.TempDir(), server)
        test.Start(t, 0, 1)
        require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the first plan."))
        test := newDaemonMessageTest(t, t.TempDir(), server)
        test.Start(t, 0, 1)
        require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the first plan."))


@@ -194,6 +200,7 @@ func TestBusyParentQueuesItsChildReport(t *testing.T) {
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
        }
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
        }

+

        test := newDaemonMessageTest(t, t.TempDir(), server)
        test.Start(t, 0, 1)
 
        test := newDaemonMessageTest(t, t.TempDir(), server)
        test.Start(t, 0, 1)
 


@@ -232,6 +239,7 @@ func TestChildReportsOnlyTheFinalReplyAndTakesQueuedToolInput(t *testing.T) {
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan needs approval.")}, 100, 0, 5))
        }
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan needs approval.")}, 100, 0, 5))
        }

+

        test := newDaemonMessageTest(t, t.TempDir(), server)
        test.Start(t, 1)
        require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the plan."))
        test := newDaemonMessageTest(t, t.TempDir(), server)
        test.Start(t, 1)
        require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the plan."))


@@ -259,6 +267,7 @@ func TestQueuedParentMessageSurvivesDaemonRestart(t *testing.T) {
        server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
        }
        server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
        }

+

        first := newDaemonMessageTest(t, dir, server)
        require.NoError(t, first.messages.Tell("OneChat", "Garden", "Remember the garden plan."))
        require.Equal(t, 1, first.agents[1].memory.Len())
        first := newDaemonMessageTest(t, dir, server)
        require.NoError(t, first.messages.Tell("OneChat", "Garden", "Remember the garden plan."))
        require.Equal(t, 1, first.agents[1].memory.Len())


@@ -319,6 +328,7 @@ func TestInterruptedChildDeliveryResumesAfterDaemonRestart(t *testing.T) {
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
        }
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
        }

+

        first := newDaemonMessageTest(t, dir, server)
        first.Start(t, 1)
        require.NoError(t, first.messages.Tell("OneChat", "Garden", "Finish the garden plan."))
        first := newDaemonMessageTest(t, dir, server)
        first.Start(t, 1)
        require.NoError(t, first.messages.Tell("OneChat", "Garden", "Finish the garden plan."))


@@ -348,6 +358,7 @@ func TestInterruptedParentReportResumesAfterDaemonRestart(t *testing.T) {
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan is ready.")}, 100, 0, 5))
        }
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan is ready.")}, 100, 0, 5))
        }

+

        first := newDaemonMessageTest(t, dir, server)
        first.Start(t, 0, 1)
        require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
        first := newDaemonMessageTest(t, dir, server)
        first.Start(t, 0, 1)
        require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))


@@ -419,6 +430,7 @@ func TestAgentMessageSnapshotBlocksDelivery(t *testing.T) {
        require.Never(t, func() bool {
                return len(sent) > 0
        }, 50*time.Millisecond, time.Millisecond)
        require.Never(t, func() bool {
                return len(sent) > 0
        }, 50*time.Millisecond, time.Millisecond)

+

        once.Do(func() {
                close(release)
        })
        once.Do(func() {
                close(release)
        })


@@ -463,6 +475,7 @@ func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) {
                        target: "Other",
                },
        }
                        target: "Other",
                },
        }

+

        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
                        require.ErrorContains(t, test.messages.Tell(tc.source, tc.target, "Do not deliver."), "not a child")
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
                        require.ErrorContains(t, test.messages.Tell(tc.source, tc.target, "Do not deliver."), "not a child")


@@ -497,6 +510,7 @@ func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) {
                        file: `{"next":0,"pending":[{"id":1,"source":"OneChat","target":"Garden","kind":"user"}]}`,
                },
        }
                        file: `{"next":0,"pending":[{"id":1,"source":"OneChat","target":"Garden","kind":"user"}]}`,
                },
        }

+

        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
                        dir := t.TempDir()
        for _, tc := range testCases {
                t.Run(tc.name, func(t *testing.T) {
                        dir := t.TempDir()


@@ -537,6 +551,7 @@ func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
        server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
        }
        server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
        }

+

        test := newDaemonMessageTest(t, t.TempDir(), server)
        child := test.agents[1]
        at := child.clock().Add(time.Minute).Format(scheduleAtLayout)
        test := newDaemonMessageTest(t, t.TempDir(), server)
        child := test.agents[1]
        at := child.clock().Add(time.Minute).Format(scheduleAtLayout)


@@ -559,94 +574,411 @@ func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
        require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source)
 }
 
        require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source)
 }
 

-// DaemonMessageTest owns test agents and the loops selected by Start.
-type DaemonMessageTest struct {
-       agents   DaemonAgents
-       loops    []*TurnLoop
-       events   []*TurnEventRecorder
-       messages *DaemonMessages
-       logs     *SyncBuffer
-       cancel   context.CancelFunc
-       done     []chan error
+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 (r *DaemonMessageTest) Start(t *testing.T, indices ...int) {
-       t.Helper()
+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",
+       }

 
 

-       ctx, cancel := context.WithCancel(t.Context())
-       r.cancel = cancel
-       for _, index := range indices {
-               done := make(chan error, 1)
-               r.done = append(r.done, done)
-               go func() {
-                       done <- r.loops[index].Run(ctx)
-               }()
+       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 (r *DaemonMessageTest) Queue(t *testing.T) DaemonMessageFile {
-       t.Helper()
+func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) {
+       testCases := []struct {
+               name   string
+               finish bool
+       }{
+               {
+                       name:   "queued turn resumes",
+                       finish: false,
+               },
+               {
+                       name:   "finished turn stays finished",
+                       finish: true,
+               },
+       }

 
 

-       data, err := os.ReadFile(r.messages.path)
-       require.NoError(t, err)
+       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))
+                       }

 
 

-       var file DaemonMessageFile
+                       first := newDaemonMessageTest(t, dir, server)
+                       inbox := whatsappRouteTestInbox(t, dir)
+                       chat := WhatsAppChat{
+                               ID:   whatsappTestAna.String(),
+                               Name: "Ana",
+                       }

 
 

-       require.NoError(t, json.Unmarshal(data, &file))
+                       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())

 
 

-       return file
-}
+                       batch := inbox.state.Closed()[0]
+                       routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
+                       require.NoError(t, err)
+                       require.True(t, routed)

 
 

-func (r *DaemonMessageTest) Close(t *testing.T) {
-       t.Helper()
+                       if tc.finish {
+                               first.Start(t, 1)
+                               first.events[1].WaitFor(t, "idle", 1)
+                               require.Empty(t, first.Queue(t).Pending)
+                       }

 
 

-       if r.cancel != nil {
-               r.cancel()
+                       first.Close(t)

 
 

-               for _, done := range r.done {
-                       require.ErrorIs(t, <-done, context.Canceled)
-               }
+                       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)

 
 

-               r.cancel = nil
-               r.done = nil
+                       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())
+               })

        }
        }

+}

 
 

-       require.NoError(t, r.agents.Close())
+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),
+               },
+       }

 
 

-       r.agents = nil
+       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 newDaemonMessageTest(t *testing.T, dir string, server *ModelFake) *DaemonMessageTest {
-       t.Helper()
+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")))

 
 

-       _, err := os.Stat(filepath.Join(dir, daemonAgentDirName, "children"))
-       if errors.Is(err, os.ErrNotExist) {
-               daemonChildConfig(t, dir, "Garden")
-               daemonChildConfig(t, dir, "Library")
-       } else {
-               require.NoError(t, err)
+       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",

        }
 
        }
 

-       test := &DaemonMessageTest{
-               logs: &SyncBuffer{},
+       second := WhatsAppChat{
+               ID:    whatsappTestFamily.String(),
+               Name:  "Garden plans",
+               Group: true,

        }
        }

-       test.agents, err = openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, daemonTestLogger(test.logs))
+       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)
        require.NoError(t, err)

-       t.Cleanup(func() {
-               test.Close(t)
-       })
+
+       _, 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 {
 
        for _, agent := range test.agents {

-               events := &TurnEventRecorder{}
-               loop, err := agent.Loop(nil, events.Add)
-               require.NoError(t, err)
+               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)

 
 

-               test.events = append(test.events, events)
-               test.loops = append(test.loops, loop)
+       for _, loop := range first.loops {
+               require.ErrorContains(t, loop.failed(), "save agent messages")
+               require.Zero(t, loop.memory.Len())

        }
 
        }
 

-       test.messages, err = openDaemonMessages(test.loops)
+       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))

        require.NoError(t, err)
 
        require.NoError(t, err)
 

-       return test
+       return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()), &ImageStore{dir: filepath.Join(dir, imageDirName)}, 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)

 }
 }