X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/f6f0febc2c599ed7bb2a2653215a74d8de1abace..a5c83e92d7feea2e104b5515043a76c8c4c56c60:/daemon_message_test.go diff --git a/daemon_message_test.go b/daemon_message_test.go index 5574172..432a640 100644 --- a/daemon_message_test.go +++ b/daemon_message_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "errors" "net/http" "os" @@ -557,3 +558,401 @@ func TestChildScheduleWakesOnlyItsOwner(t *testing.T) { require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.") require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source) } + +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)) + require.NoError(t, err) + + 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) +}