Order daemon and storage declarations

[onechat] / daemon_message_test.go

diff --git a/daemon_message_test.go b/daemon_message_test.go
index f9c25f13ad8e231e9c180a8f496bf9e818c48618..432a6406f2fa9882aea2279898edea74c373f4b2 100644 (file)
--- 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)
        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 {
 
        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)
 
        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)

 }
 }