require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source)
}
-// 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 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) Start(t *testing.T, indices ...int) {
- 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())
+}
- 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 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 (r *DaemonMessageTest) Queue(t *testing.T) DaemonMessageFile {
- t.Helper()
+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())
- data, err := os.ReadFile(r.messages.path)
- require.NoError(t, err)
+ batch := inbox.state.Closed()[0]
+ routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
+ require.NoError(t, err)
+ require.True(t, routed)
- var file DaemonMessageFile
+ if tc.finish {
+ first.Start(t, 1)
+ first.events[1].WaitFor(t, "idle", 1)
+ require.Empty(t, first.Queue(t).Pending)
+ }
- require.NoError(t, json.Unmarshal(data, &file))
+ first.Close(t)
- return file
-}
+ 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, ""))
-func (r *DaemonMessageTest) Close(t *testing.T) {
- t.Helper()
+ 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 r.cancel != nil {
- r.cancel()
+ if !tc.finish {
+ second.Start(t, 1)
+ second.events[1].WaitFor(t, "idle", 1)
+ }
- for _, done := range r.done {
- require.ErrorIs(t, <-done, context.Canceled)
- }
+ require.Equal(t, 1, server.Calls())
+ require.Empty(t, second.Queue(t).Pending)
+ second.Close(t)
- r.cancel = nil
- r.done = 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())
+ })
}
-
- require.NoError(t, r.agents.Close())
-
- r.agents = nil
}
-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())
+ }
+
+ 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.events = append(test.events, events)
- test.loops = append(test.loops, loop)
+ for _, loop := range first.loops {
+ require.ErrorContains(t, loop.failed(), "save agent messages")
+ require.Zero(t, loop.memory.Len())
}
- test.messages, err = openDaemonMessages(test.loops)
+ 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)
}
projects / onechat / blobdiff