From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 07:21:17 +0000 (-0300) Subject: Deliver routed WhatsApp batches to children X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/effd32b0f2e4d146b219b1727601b0b7f6593638 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 36db09c..7912d0d 100644 --- 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 -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 @@ -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 -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, diff --git a/daemon_message.go b/daemon_message.go index 40a99a9..8ee26e8 100644 --- 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. +// 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 { @@ -66,15 +67,32 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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 - 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] + 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 (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") } @@ -91,10 +109,11 @@ func openDaemonMessages(loops []*TurnLoop) (*DaemonMessages, error) { 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 @@ -118,7 +137,10 @@ func (m *DaemonMessages) deliver(message *DaemonMessage) error { } status := "" - if message.Kind == "work" { + source := message.Source + if message.WhatsAppBatch > 0 { + source = whatsappSource + } else if message.Kind == "work" { status = subagentFinished } @@ -126,11 +148,16 @@ func (m *DaemonMessages) deliver(message *DaemonMessage) error { Kind: message.Kind, ID: logged.ID, Text: message.Text, - Source: message.Source, + Source: source, 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 } @@ -257,6 +284,82 @@ func (m *DaemonMessages) Tell(parent, child, text string) error { 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++ @@ -339,19 +442,24 @@ func (m *DaemonMessages) Finish(loop *TurnLoop, active []Message, reply string) 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 { - 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 -// 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 { - 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 c7fbf51..adff79e 100644 --- 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 -// 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 @@ -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. +// 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() @@ -110,14 +114,39 @@ func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppL 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 - 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) } - 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) } @@ -146,6 +175,13 @@ func (i *WhatsAppInbox) deliver(ctx context.Context, l *TurnLoop, link WhatsAppL 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 } diff --git a/whatsapp_route_test.go b/whatsapp_route_test.go new file mode 100644 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) +}