X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/2ad578a24f252706976cddb41bee6cc969f251cd..a5c83e92d7feea2e104b5515043a76c8c4c56c60:/daemon_message_test.go diff --git a/daemon_message_test.go b/daemon_message_test.go index f9c25f1..432a640 100644 --- a/daemon_message_test.go +++ b/daemon_message_test.go @@ -532,94 +532,427 @@ func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) { } } -// 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 TestChildScheduleWakesOnlyItsOwner(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) + child := test.agents[1] + at := child.clock().Add(time.Minute).Format(scheduleAtLayout) + output := test.loops[1].tools.run(t.Context(), "schedule_set", `{"text":"Check the plants.","at":"`+at+`","cron":null}`) + require.Contains(t, output, "Schedule 1 set.") + require.Equal(t, "Garden", child.schedules.sorted()[0].Owner) + + child.schedules.clock = func() time.Time { + return child.clock().Add(time.Minute) + } + require.ErrorContains(t, child.schedules.deliver(t.Context(), test.loops[0]), "does not match") + require.NoError(t, child.schedules.deliver(t.Context(), test.loops[1])) + test.Start(t, 0, 1, 2) + test.events[1].WaitFor(t, "idle", 1) + require.Equal(t, 1, server.Calls()) + require.Equal(t, child.agent.CacheKey(), server.Call(0).body["prompt_cache_key"]) + require.Zero(t, test.agents[0].memory.Len()) + require.Zero(t, test.agents[2].memory.Len()) + require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.") + require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source) } -func (r *DaemonMessageTest) Start(t *testing.T, indices ...int) { - t.Helper() - - 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) - }() +func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) { + testCases := []struct { + name string + chat WhatsAppChat + }{ + { + name: "private chat", + chat: WhatsAppChat{ + ID: whatsappTestAna.String(), + Name: "Ana", + }, + }, + { + name: "group without a mention", + chat: WhatsAppChat{ + ID: whatsappTestFamily.String(), + Name: "Garden plans", + Group: true, + }, + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + server := newModelFake(t) + server.reply = func(w http.ResponseWriter, _ ModelFakeCall) { + modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5)) + } + dir := t.TempDir() + test := newDaemonMessageTest(t, dir, server) + inbox := whatsappRouteTestInbox(t, dir) + require.NoError(t, test.messages.routes.Set("whatsapp:"+tc.chat.ID, "Garden")) + whatsappRouteTestAdd(t, inbox, tc.chat, "The gate needs a new latch.") + test.Start(t, 0, 1, 2) + require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true)) + test.events[1].WaitFor(t, "idle", 1) + require.Equal(t, 1, server.Calls()) + require.Equal(t, "gpt-test-garden", server.Call(0).body["model"]) + require.Contains(t, string(server.Call(0).raw), "The gate needs a new latch.") + require.Zero(t, test.agents[0].memory.Len()) + require.Empty(t, test.events[0].Kinds()) + require.Zero(t, test.agents[2].memory.Len()) + require.False(t, test.loops[0].waiting()) + require.Equal(t, whatsappSource, test.events[1].LastEvent("work").Source) + require.Empty(t, test.events[1].LastEvent("work").Status) + require.Equal(t, 1, test.agents[1].memory.Len()) + require.Empty(t, inbox.state.Closed()) + require.Empty(t, test.Queue(t).Pending) + require.Contains(t, test.logs.String(), "batch routed name=Garden kind=whatsapp id=1 to=Garden") + require.NotContains(t, test.logs.String(), tc.chat.ID) + require.NotContains(t, test.logs.String(), "new latch") + }) } } -func (r *DaemonMessageTest) Queue(t *testing.T) DaemonMessageFile { - 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", + } + 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()) +} - data, err := os.ReadFile(r.messages.path) - require.NoError(t, err) +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()) +} - var file DaemonMessageFile +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()) - require.NoError(t, json.Unmarshal(data, &file)) + batch := inbox.state.Closed()[0] + routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC)) + require.NoError(t, err) + require.True(t, routed) - return file -} + if tc.finish { + first.Start(t, 1) + first.events[1].WaitFor(t, "idle", 1) + require.Empty(t, first.Queue(t).Pending) + } -func (r *DaemonMessageTest) Close(t *testing.T) { - t.Helper() + first.Close(t) - if r.cancel != nil { - r.cancel() + 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, "")) - for _, done := range r.done { - require.ErrorIs(t, <-done, context.Canceled) - } + 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.NoError(t, r.agents.Close()) + require.Equal(t, 1, server.Calls()) + require.Empty(t, second.Queue(t).Pending) + second.Close(t) - r.agents = nil + third := newDaemonMessageTest(t, dir, server) + require.False(t, third.loops[1].waiting()) + require.Empty(t, third.events[1].Kinds()) + require.Equal(t, 1, third.agents[1].memory.Len()) + }) + } } -func newDaemonMessageTest(t *testing.T, dir string, server *ModelFake) *DaemonMessageTest { - t.Helper() +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()) + }) + } +} - _, 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) +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"))) - test := &DaemonMessageTest{ - logs: &SyncBuffer{}, + 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, } - 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()) + } - test.events = append(test.events, events) - test.loops = append(test.loops, loop) + inbox.endCatchup() + require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false)) + + for _, loop := range test.loops { + require.Equal(t, 1, loop.memory.Len()) + require.True(t, loop.waiting()) + } + + require.Contains(t, test.events[0].Last("work"), "Check the mixed chat.") + require.Contains(t, test.events[1].Last("work"), "Check the gate.") + require.Contains(t, test.events[2].Last("work"), "Check the book.") + require.Empty(t, inbox.state.Closed()) + require.Empty(t, test.Queue(t).WhatsAppBatches) +} + +func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) { + dir := t.TempDir() + server := newModelFake(t) + first := newDaemonMessageTest(t, dir, server) + inbox := whatsappRouteTestInbox(t, dir) + chat := WhatsAppChat{ + ID: whatsappTestAna.String(), + Name: "Ana", } + require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden")) + require.NoError(t, os.Mkdir(first.messages.path, 0o700)) + whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.") + require.ErrorContains(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true), "save agent messages") + require.Len(t, inbox.state.Closed(), 1) - test.messages, err = openDaemonMessages(test.loops) + 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) }