X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/723f6e18adc6482ed913e1618c8a76e981720c2b..c781f79f3c8cb9a7577ee1574d196988da2719b3:/daemon_message_test.go diff --git a/daemon_message_test.go b/daemon_message_test.go index 5574172..27138d3 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" @@ -23,6 +24,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.")) @@ -67,6 +69,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()) @@ -96,6 +99,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")) @@ -129,12 +133,14 @@ func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.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)) } + test := newDaemonMessageTest(t, t.TempDir(), server) test.Start(t, 0, 1, 2) require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan.")) @@ -161,6 +167,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.")) @@ -193,6 +200,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) @@ -231,6 +239,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.")) @@ -258,6 +267,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()) @@ -318,6 +328,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.")) @@ -347,6 +358,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.")) @@ -418,6 +430,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) }) @@ -462,6 +475,7 @@ func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) { 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") @@ -496,6 +510,7 @@ func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) { 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() @@ -536,6 +551,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) @@ -557,3 +573,412 @@ 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) +}