X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/2124c559da8f164a43a3da987b399b6ac6125035..98d137c3e945a09f76e85bd18238558c668fdca4:/daemon_message_test.go?ds=sidebyside

diff --git a/daemon_message_test.go b/daemon_message_test.go
index 060faf5..cfb41c6 100644
--- a/daemon_message_test.go
+++ b/daemon_message_test.go
@@ -4,9 +4,11 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"net/http"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -14,6 +16,388 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+func TestNotificationWaitsForTheNextTurnAfterEveryActiveToolRequest(t *testing.T) {
+	server := newModelFake(t)
+	arrived := make(chan struct{})
+	release := make(chan struct{})
+	server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+		if call.number == 0 {
+			close(arrived)
+			<-release
+			modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`)}, 100, 0, 5))
+
+			return
+		}
+
+		modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+	}
+
+	test := newDaemonMessageTest(t, t.TempDir(), server)
+	test.Start(t, 0, 1, 2)
+
+	_, err := test.loops[0].Submit(terminalSource, "Start a tool chain.")
+	require.NoError(t, err)
+	<-arrived
+
+	before := test.agents[0].memory.Len()
+	input := &NotifyInput{
+		Source: "test-check",
+		ID:     "event-1",
+		Text:   "Unique queued external data.",
+	}
+
+	receipt, err := test.messages.Notify(input)
+	require.NoError(t, err)
+	require.Equal(t, "accepted", receipt.Status)
+	require.Equal(t, before, test.agents[0].memory.Len())
+	require.True(t, test.loops[0].Busy())
+	close(release)
+	test.events[0].WaitFor(t, "idle", 2)
+	require.Equal(t, 3, server.Calls())
+	require.NotContains(t, strings.Join(turnInputTexts(t, server.Call(1)), "\n"), input.Text)
+	require.Contains(t, strings.Join(turnInputTexts(t, server.Call(2)), "\n"), input.Text)
+	require.Equal(t, notifySource, test.events[0].LastEvent("work").Source)
+	require.Empty(t, test.Queue(t).Pending)
+	require.Len(t, test.Queue(t).Notifications, 1)
+	require.Zero(t, test.agents[1].memory.Len())
+	require.Zero(t, test.agents[2].memory.Len())
+	require.NotContains(t, test.logs.String(), input.Source)
+	require.NotContains(t, test.logs.String(), input.Text)
+	require.NotContains(t, test.logs.String(), input.ID)
+}
+
+func TestNotificationRetriesSurviveRestartBeforeAndAfterDelivery(t *testing.T) {
+	directory := t.TempDir()
+	server := newModelFake(t)
+	server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+		modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+	}
+
+	input := &NotifyInput{
+		Source: "check",
+		ID:     "result-1",
+		Text:   "A new result.",
+	}
+
+	first := newDaemonMessageTest(t, directory, server)
+	receipt, err := first.messages.Notify(input)
+	require.NoError(t, err)
+	require.Zero(t, first.agents[0].memory.Len())
+	first.Close(t)
+
+	second := newDaemonMessageTest(t, directory, server)
+	duplicate, err := second.messages.Notify(input)
+	require.NoError(t, err)
+	require.Equal(t, receipt.ID, duplicate.ID)
+	require.Equal(t, "duplicate", duplicate.Status)
+	require.Zero(t, second.agents[0].memory.Len())
+	second.Start(t, 0)
+	second.events[0].WaitFor(t, "idle", 1)
+	require.Empty(t, second.Queue(t).Pending)
+	second.Close(t)
+
+	third := newDaemonMessageTest(t, directory, server)
+	duplicate, err = third.messages.Notify(input)
+	require.NoError(t, err)
+	require.Equal(t, "duplicate", duplicate.Status)
+	require.Equal(t, receipt.ID, duplicate.ID)
+	require.False(t, third.loops[0].waiting())
+	require.Equal(t, 1, server.Calls())
+
+	changed := *input
+	changed.Text = "A different result."
+	_, err = third.messages.Notify(&changed)
+	require.ErrorContains(t, err, "conflicts")
+	require.Len(t, third.Queue(t).Notifications, 1)
+}
+
+func TestNotificationAcceptanceIsSerializedForConcurrentRetries(t *testing.T) {
+	server := newModelFake(t)
+	test := newDaemonMessageTest(t, t.TempDir(), server)
+	input := &NotifyInput{
+		Source: "check",
+		ID:     "1",
+		Text:   "A result.",
+	}
+
+	results := make(chan Event, 20)
+	failures := make(chan error, 20)
+
+	var workers sync.WaitGroup
+
+	for range 20 {
+		workers.Add(1)
+
+		go func() {
+			defer workers.Done()
+
+			receipt, err := test.messages.Notify(input)
+			results <- receipt
+
+			failures <- err
+		}()
+	}
+
+	workers.Wait()
+	close(results)
+	close(failures)
+
+	accepted := 0
+	for receipt := range results {
+		require.Equal(t, 1, receipt.ID)
+
+		if receipt.Status == "accepted" {
+			accepted++
+		} else {
+			require.Equal(t, "duplicate", receipt.Status)
+		}
+	}
+
+	for err := range failures {
+		require.NoError(t, err)
+	}
+
+	require.Equal(t, 1, accepted)
+	require.Len(t, test.Queue(t).Pending, 1)
+	require.Len(t, test.Queue(t).Notifications, 1)
+}
+
+func TestNotificationStorageFailureReturnsNoAcceptedReceipt(t *testing.T) {
+	server := newModelFake(t)
+	test := newDaemonMessageTest(t, t.TempDir(), server)
+	test.messages.path = filepath.Join(t.TempDir(), "missing", "messages.json")
+	receipt, err := test.messages.Notify(&NotifyInput{
+		Source: "check",
+		ID:     "1",
+		Text:   "A result.",
+	})
+	require.ErrorContains(t, err, "accept external event")
+	require.Empty(t, receipt.Kind)
+	require.False(t, test.loops[0].waiting())
+	require.Zero(t, test.agents[0].memory.Len())
+
+	_, err = test.messages.Notify(&NotifyInput{
+		Source: "check",
+		ID:     "2",
+		Text:   "Another result.",
+	})
+	require.ErrorContains(t, err, "queue stopped")
+}
+
+func TestNotificationFullStoreRejectsNewKeysWithoutRemovingReceipts(t *testing.T) {
+	server := newModelFake(t)
+	test := newDaemonMessageTest(t, t.TempDir(), server)
+
+	for number := range notifyReceipts {
+		test.messages.state.Notifications = append(test.messages.state.Notifications, NotifyReceipt{
+			Source: "check",
+			ID:     fmt.Sprint(number),
+			Digest: notifyDigest("data"),
+			Number: number + 1,
+		})
+	}
+
+	test.messages.state.Next = notifyReceipts
+	receipt, err := test.messages.Notify(&NotifyInput{
+		Source: "check",
+		ID:     "new-key",
+		Text:   "data",
+	})
+	require.ErrorContains(t, err, "store is full")
+	require.Empty(t, receipt.Kind)
+	require.Len(t, test.messages.state.Notifications, notifyReceipts)
+	require.Empty(t, test.messages.state.Pending)
+
+	receipt, err = test.messages.Notify(&NotifyInput{
+		Source: "check",
+		ID:     "0",
+		Text:   "data",
+	})
+	require.NoError(t, err)
+	require.Equal(t, "duplicate", receipt.Status)
+}
+
+func TestNotificationBatchesKeepAcceptanceOrder(t *testing.T) {
+	server := newModelFake(t)
+	server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+		modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+	}
+
+	test := newDaemonMessageTest(t, t.TempDir(), server)
+	test.agents[0].memory.summarizer = newFakeSummarizer()
+	for number := range 21 {
+		_, err := test.messages.Notify(&NotifyInput{
+			Source: "check",
+			ID:     fmt.Sprint(number),
+			Text:   fmt.Sprintf("result-%02d", number),
+		})
+		require.NoError(t, err)
+	}
+
+	test.Start(t, 0)
+	test.events[0].WaitFor(t, "idle", 2)
+	require.Equal(t, 2, server.Calls())
+
+	first := turnInputTexts(t, server.Call(0))
+	require.Len(t, first, 22)
+
+	for number := range 20 {
+		require.Contains(t, first[number+2], fmt.Sprintf("result-%02d", number))
+	}
+
+	second := turnInputTexts(t, server.Call(1))
+	require.Contains(t, second[len(second)-1], "result-20")
+	require.Empty(t, test.Queue(t).Pending)
+}
+
+func TestNotificationRecoversItsPreparedLogIDWithoutAnotherAppend(t *testing.T) {
+	testCases := []struct {
+		name     string
+		appended bool
+	}{
+		{
+			name: "recovers a saved ID before the log append",
+		},
+		{
+			name:     "reuses the log entry after the append",
+			appended: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			directory := t.TempDir()
+			server := newModelFake(t)
+			server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+				modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+			}
+
+			first := newDaemonMessageTest(t, directory, server)
+			_, err := first.messages.Notify(&NotifyInput{
+				Source: "check",
+				ID:     "1",
+				Text:   "A result.",
+			})
+			require.NoError(t, err)
+
+			if tc.appended {
+				require.NoError(t, first.messages.Notifications(first.loops[0]))
+			} else {
+				id := 0
+				first.messages.state.Pending[0].LogID = &id
+				require.NoError(t, first.messages.save())
+			}
+
+			first.Close(t)
+
+			second := newDaemonMessageTest(t, directory, server)
+			second.Start(t, 0)
+			second.events[0].WaitFor(t, "idle", 1)
+			require.Equal(t, 1, server.Calls())
+			require.Equal(t, 1, second.agents[0].memory.Len())
+			require.Empty(t, second.Queue(t).Pending)
+		})
+	}
+}
+
+func TestPreparedNotificationKeepsItsLogIDBeforeStartupWriters(t *testing.T) {
+	testCases := []struct {
+		name   string
+		resume bool
+	}{
+		{
+			name: "keeps its reservation before a startup user message",
+		},
+		{
+			name:   "keeps its reservation before a worker recovery note",
+			resume: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			directory := t.TempDir()
+			server := newModelFake(t)
+			server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+				modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+			}
+
+			first := newDaemonMessageTest(t, directory, server)
+			input := &NotifyInput{
+				Source: "check",
+				ID:     "reserved",
+				Text:   "A result with a prepared log ID.",
+			}
+
+			receipt, err := first.messages.Notify(input)
+			require.NoError(t, err)
+
+			id := 0
+			first.messages.state.Pending[0].LogID = &id
+			require.NoError(t, first.messages.save())
+
+			expected := first.Queue(t).Pending[0].Text
+			first.Close(t)
+
+			second := newDaemonMessageTest(t, directory, server)
+			require.False(t, second.loops[0].waiting())
+			require.Empty(t, second.events[0].Kinds())
+
+			if tc.resume {
+				path := filepath.Join(second.agents[0].agent.dir, subagentWorkFileName)
+				require.NoError(t, os.WriteFile(path, []byte(`{"running":["BlueFox"]}`), 0o600))
+				require.NoError(t, second.loops[0].runs.Resume(t.Context()))
+
+				note, err := second.agents[0].memory.Message(1)
+				require.NoError(t, err)
+				require.Equal(t, "note", note.Kind)
+				require.Contains(t, note.Text, "The daemon restarted, cutting BlueFox")
+			} else {
+				submitted, err := second.loops[0].Submit(terminalSource, "A user message during startup.")
+				require.NoError(t, err)
+				require.Equal(t, 1, submitted.ID)
+			}
+
+			restored, err := second.agents[0].memory.Message(id)
+			require.NoError(t, err)
+			require.Equal(t, "work", restored.Kind)
+			require.Equal(t, expected, restored.Text)
+			require.Len(t, second.Queue(t).Pending, 1)
+
+			second.Start(t, 0)
+			second.events[0].WaitFor(t, "idle", 1)
+			require.Empty(t, second.Queue(t).Pending)
+			require.Equal(t, 1, server.Calls())
+			require.Equal(t, id, second.events[0].LastEvent("work").ID)
+
+			duplicate, err := second.messages.Notify(input)
+			require.NoError(t, err)
+			require.Equal(t, "duplicate", duplicate.Status)
+			require.Equal(t, receipt.ID, duplicate.ID)
+			require.Len(t, second.Queue(t).Notifications, 1)
+		})
+	}
+}
+
+func TestNotificationBatchMakesProgressWithEscapedControlCharacters(t *testing.T) {
+	server := newModelFake(t)
+	server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+		modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+	}
+
+	test := newDaemonMessageTest(t, t.TempDir(), server)
+	_, err := test.messages.Notify(&NotifyInput{
+		Source: "check",
+		ID:     "escaped",
+		Text:   "data" + strings.Repeat("\x00", notifyTextBytes-4),
+	})
+	require.NoError(t, err)
+	require.Greater(t, len(test.Queue(t).Pending[0].Text), 64<<10)
+	test.Start(t, 0)
+	test.events[0].WaitFor(t, "idle", 1)
+	require.Empty(t, test.Queue(t).Pending)
+}
+
 func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
 	server := newModelFake(t)
 	server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
@@ -24,6 +408,7 @@ func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
 
 		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."))
@@ -68,6 +453,7 @@ func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
 
 		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())
@@ -97,6 +483,7 @@ func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
 	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"))
@@ -122,20 +509,22 @@ func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.T) {
 		text string
 	}{
 		{
-			name: "empty final reply",
+			name: "sends nothing for an empty final reply",
 			text: "",
 		},
 		{
-			name: "whitespace final reply",
+			name: "sends nothing for a whitespace final reply",
 			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))
 			}
+
 			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 +551,7 @@ func TestBusyChildQueuesTheNextParentMessage(t *testing.T) {
 
 		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."))
@@ -194,6 +584,7 @@ func TestBusyParentQueuesItsChildReport(t *testing.T) {
 
 		modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 	}
+
 	test := newDaemonMessageTest(t, t.TempDir(), server)
 	test.Start(t, 0, 1)
 
@@ -232,6 +623,7 @@ func TestChildReportsOnlyTheFinalReplyAndTakesQueuedToolInput(t *testing.T) {
 
 		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."))
@@ -259,6 +651,7 @@ func TestQueuedParentMessageSurvivesDaemonRestart(t *testing.T) {
 	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())
@@ -319,6 +712,7 @@ func TestInterruptedChildDeliveryResumesAfterDaemonRestart(t *testing.T) {
 
 		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."))
@@ -348,6 +742,7 @@ func TestInterruptedParentReportResumesAfterDaemonRestart(t *testing.T) {
 
 		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."))
@@ -419,6 +814,7 @@ func TestAgentMessageSnapshotBlocksDelivery(t *testing.T) {
 	require.Never(t, func() bool {
 		return len(sent) > 0
 	}, 50*time.Millisecond, time.Millisecond)
+
 	once.Do(func() {
 		close(release)
 	})
@@ -453,16 +849,17 @@ func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) {
 			target: "OneChat",
 		},
 		{
-			name:   "unknown sender",
+			name:   "refuses an unknown sender",
 			source: "Other",
 			target: "Garden",
 		},
 		{
-			name:   "unknown recipient",
+			name:   "refuses an unknown recipient",
 			source: "OneChat",
 			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")
@@ -481,22 +878,23 @@ func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) {
 		file string
 	}{
 		{
-			name: "broken JSON",
+			name: "stops startup for broken JSON",
 			file: "{",
 		},
 		{
-			name: "unknown sender",
+			name: "stops startup for unknown sender",
 			file: `{"next":1,"pending":[{"id":1,"source":"Other","target":"Garden","kind":"user"}]}`,
 		},
 		{
-			name: "sibling delivery",
+			name: "stops startup for sibling delivery",
 			file: `{"next":1,"pending":[{"id":1,"source":"Library","target":"Garden","kind":"user"}]}`,
 		},
 		{
-			name: "invalid sequence",
+			name: "stops startup for invalid sequence",
 			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()
@@ -537,6 +935,7 @@ func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
 	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)
@@ -559,94 +958,416 @@ func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
 	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: "routes a private chat only to its child",
+			chat: WhatsAppChat{
+				ID:   whatsappTestAna.String(),
+				Name: "Ana",
+			},
+		},
+		{
+			name: "routes an unmentioned group only to its child",
+			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())
 
-		r.cancel = nil
-		r.done = nil
+			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: "recovers one message before its log ID is stored",
+			id:   nil,
+		},
+		{
+			name: "recovers one message after its log ID is stored",
+			id:   new(0),
+		},
 	}
 
-	require.NoError(t, r.agents.Close())
+	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,
+				}},
+			}
 
-	r.agents = nil
+			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",
+	}
 
-	_, 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)
+	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",
 	}
 
-	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)
-	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 {
-		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))
 
-		test.events = append(test.events, events)
-		test.loops = append(test.loops, loop)
+	for _, loop := range test.loops {
+		require.Equal(t, 1, loop.memory.Len())
+		require.True(t, loop.waiting())
 	}
 
-	test.messages, err = openDaemonMessages(test.loops)
+	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))
 	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)
 }