]> git.giacomolli.uk - onechat/blob - whatsapp_route_test.go
Receive images in WhatsApp batches
[onechat] / whatsapp_route_test.go
1 package main
2
3 import (
4 "encoding/json"
5 "net/http"
6 "os"
7 "path/filepath"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/require"
12 )
13
14 func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) {
15 testCases := []struct {
16 name string
17 chat WhatsAppChat
18 }{
19 {
20 name: "private chat",
21 chat: WhatsAppChat{
22 ID: whatsappTestAna.String(),
23 Name: "Ana",
24 },
25 },
26 {
27 name: "group without a mention",
28 chat: WhatsAppChat{
29 ID: whatsappTestFamily.String(),
30 Name: "Garden plans",
31 Group: true,
32 },
33 },
34 }
35 for _, tc := range testCases {
36 t.Run(tc.name, func(t *testing.T) {
37 server := newModelFake(t)
38 server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
39 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
40 }
41 dir := t.TempDir()
42 test := newDaemonMessageTest(t, dir, server)
43 inbox := whatsappRouteTestInbox(t, dir)
44 require.NoError(t, test.messages.routes.Set("whatsapp:"+tc.chat.ID, "Garden"))
45 whatsappRouteTestAdd(t, inbox, tc.chat, "The gate needs a new latch.")
46 test.Start(t, 0, 1, 2)
47 require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
48 test.events[1].WaitFor(t, "idle", 1)
49 require.Equal(t, 1, server.Calls())
50 require.Equal(t, "gpt-test-garden", server.Call(0).body["model"])
51 require.Contains(t, string(server.Call(0).raw), "The gate needs a new latch.")
52 require.Zero(t, test.agents[0].memory.Len())
53 require.Empty(t, test.events[0].Kinds())
54 require.Zero(t, test.agents[2].memory.Len())
55 require.False(t, test.loops[0].waiting())
56 require.Equal(t, whatsappSource, test.events[1].LastEvent("work").Source)
57 require.Empty(t, test.events[1].LastEvent("work").Status)
58 require.Equal(t, 1, test.agents[1].memory.Len())
59 require.Empty(t, inbox.state.Closed())
60 require.Empty(t, test.Queue(t).Pending)
61 require.Contains(t, test.logs.String(), "batch routed name=Garden kind=whatsapp id=1 to=Garden")
62 require.NotContains(t, test.logs.String(), tc.chat.ID)
63 require.NotContains(t, test.logs.String(), "new latch")
64 })
65 }
66 }
67
68 func TestWhatsAppUnrouteRestoresRootDelivery(t *testing.T) {
69 dir := t.TempDir()
70 test := newDaemonMessageTest(t, dir, newModelFake(t))
71 inbox := whatsappRouteTestInbox(t, dir)
72 chat := WhatsAppChat{
73 ID: whatsappTestAna.String(),
74 Name: "Ana",
75 }
76 source := "whatsapp:" + chat.ID
77 require.NoError(t, test.messages.routes.Set(source, "Garden"))
78 whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
79 require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
80 require.Zero(t, test.agents[0].memory.Len())
81 require.True(t, test.loops[1].waiting())
82 require.NoError(t, test.messages.routes.Set(source, ""))
83 whatsappRouteTestAdd(t, inbox, chat, "Please check the other plan.")
84 require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
85 require.Equal(t, 1, test.agents[0].memory.Len())
86 require.Equal(t, 1, test.agents[1].memory.Len())
87 require.True(t, test.loops[0].waiting())
88 require.Contains(t, test.events[0].Last("work"), "Please check the other plan.")
89 require.NotContains(t, test.events[0].Last("work"), "Keep the gate plan.")
90 require.Empty(t, inbox.state.Closed())
91 }
92
93 func TestMissingWhatsAppRouteChildFallsBackToRoot(t *testing.T) {
94 dir := t.TempDir()
95 test := newDaemonMessageTest(t, dir, newModelFake(t))
96 inbox := whatsappRouteTestInbox(t, dir)
97 chat := WhatsAppChat{
98 ID: whatsappTestAna.String(),
99 Name: "Ana",
100 }
101 require.NoError(t, test.messages.routes.Set("whatsapp:"+chat.ID, "Missing"))
102 whatsappRouteTestAdd(t, inbox, chat, "Please check the plan.")
103 require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
104 require.Equal(t, 1, test.agents[0].memory.Len())
105 require.True(t, test.loops[0].waiting())
106 require.Zero(t, test.agents[1].memory.Len())
107 require.False(t, test.loops[1].waiting())
108 require.Empty(t, test.events[0].LastEvent("work").Status)
109 require.Equal(t, whatsappSource, test.events[0].LastEvent("work").Source)
110 require.Empty(t, inbox.state.Closed())
111 }
112
113 func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) {
114 testCases := []struct {
115 name string
116 finish bool
117 }{
118 {
119 name: "queued turn resumes",
120 finish: false,
121 },
122 {
123 name: "finished turn stays finished",
124 finish: true,
125 },
126 }
127 for _, tc := range testCases {
128 t.Run(tc.name, func(t *testing.T) {
129 dir := t.TempDir()
130 server := newModelFake(t)
131 server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
132 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
133 }
134 first := newDaemonMessageTest(t, dir, server)
135 inbox := whatsappRouteTestInbox(t, dir)
136 chat := WhatsAppChat{
137 ID: whatsappTestAna.String(),
138 Name: "Ana",
139 }
140 source := "whatsapp:" + chat.ID
141 require.NoError(t, first.messages.routes.Set(source, "Garden"))
142 whatsappRouteTestAdd(t, inbox, chat, "Remember the gate plan.")
143 require.NoError(t, inbox.state.CloseAll())
144
145 batch := inbox.state.Closed()[0]
146 routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
147 require.NoError(t, err)
148 require.True(t, routed)
149
150 if tc.finish {
151 first.Start(t, 1)
152 first.events[1].WaitFor(t, "idle", 1)
153 require.Empty(t, first.Queue(t).Pending)
154 }
155
156 first.Close(t)
157
158 second := newDaemonMessageTest(t, dir, server)
159 require.Equal(t, source+" -> Garden", second.messages.routes.List())
160 require.Equal(t, 1, second.agents[1].memory.Len())
161 require.Equal(t, !tc.finish, second.loops[1].waiting())
162 require.NoError(t, second.messages.routes.Set(source, ""))
163
164 restored := whatsappRouteTestInbox(t, dir)
165 require.Len(t, restored.state.Closed(), 1)
166 require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
167 require.Empty(t, restored.state.Closed())
168 require.Zero(t, second.agents[0].memory.Len())
169 require.Equal(t, 1, second.agents[1].memory.Len())
170
171 if !tc.finish {
172 second.Start(t, 1)
173 second.events[1].WaitFor(t, "idle", 1)
174 }
175
176 require.Equal(t, 1, server.Calls())
177 require.Empty(t, second.Queue(t).Pending)
178 second.Close(t)
179
180 third := newDaemonMessageTest(t, dir, server)
181 require.False(t, third.loops[1].waiting())
182 require.Empty(t, third.events[1].Kinds())
183 require.Equal(t, 1, third.agents[1].memory.Len())
184 })
185 }
186 }
187
188 func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) {
189 testCases := []struct {
190 name string
191 id *int
192 }{
193 {
194 name: "before log ID is stored",
195 id: nil,
196 },
197 {
198 name: "after log ID is stored",
199 id: new(0),
200 },
201 }
202 for _, tc := range testCases {
203 t.Run(tc.name, func(t *testing.T) {
204 dir := t.TempDir()
205 server := newModelFake(t)
206 first := newDaemonMessageTest(t, dir, server)
207 path := first.messages.path
208 first.Close(t)
209
210 file := DaemonMessageFile{
211 Next: 1,
212 WhatsAppBatches: []int{1},
213 Pending: []DaemonMessage{{
214 ID: 1,
215 Source: "OneChat",
216 Target: "Garden",
217 Kind: "work",
218 Text: "[WhatsApp] Keep the gate plan.",
219 LogID: tc.id,
220 WhatsAppBatch: 1,
221 }},
222 }
223 data, err := json.Marshal(file)
224 require.NoError(t, err)
225 require.NoError(t, os.WriteFile(path, data, 0o600))
226
227 second := newDaemonMessageTest(t, dir, server)
228 require.Equal(t, 1, second.agents[1].memory.Len())
229 require.True(t, second.loops[1].waiting())
230 require.Equal(t, "[WhatsApp] Keep the gate plan.", second.events[1].Last("work"))
231 require.Equal(t, whatsappSource, second.events[1].LastEvent("work").Source)
232 second.Close(t)
233
234 third := newDaemonMessageTest(t, dir, server)
235 require.Equal(t, 1, third.agents[1].memory.Len())
236 require.True(t, third.loops[1].waiting())
237 require.Zero(t, third.agents[0].memory.Len())
238 })
239 }
240 }
241
242 func TestQueuedWhatsAppBatchUsesRootIfItsChildIsRemoved(t *testing.T) {
243 dir := t.TempDir()
244 server := newModelFake(t)
245 first := newDaemonMessageTest(t, dir, server)
246 inbox := whatsappRouteTestInbox(t, dir)
247 chat := WhatsAppChat{
248 ID: whatsappTestAna.String(),
249 Name: "Ana",
250 }
251 require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
252 whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
253 require.NoError(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true))
254 first.Close(t)
255 require.NoError(t, os.RemoveAll(filepath.Join(dir, daemonAgentDirName, "children", "Garden")))
256
257 second := newDaemonMessageTest(t, dir, server)
258 require.Equal(t, 1, second.agents[0].memory.Len())
259 require.True(t, second.loops[0].waiting())
260 require.Equal(t, whatsappSource, second.events[0].LastEvent("work").Source)
261 require.Empty(t, second.events[0].LastEvent("work").Status)
262 require.Contains(t, second.events[0].Last("work"), "Keep the gate plan.")
263 require.False(t, second.loops[1].waiting())
264 second.Close(t)
265
266 third := newDaemonMessageTest(t, dir, server)
267 require.Equal(t, 1, third.agents[0].memory.Len())
268 require.True(t, third.loops[0].waiting())
269 }
270
271 func TestRoutedWhatsAppBatchesCanCloseOutOfNumberOrder(t *testing.T) {
272 dir := t.TempDir()
273 test := newDaemonMessageTest(t, dir, newModelFake(t))
274 inbox := whatsappRouteTestInbox(t, dir)
275 first := WhatsAppChat{
276 ID: whatsappTestAna.String(),
277 Name: "Ana",
278 }
279 second := WhatsAppChat{
280 ID: whatsappTestFamily.String(),
281 Name: "Garden plans",
282 Group: true,
283 }
284 require.NoError(t, test.messages.routes.Set("whatsapp:"+first.ID, "Garden"))
285 require.NoError(t, test.messages.routes.Set("whatsapp:"+second.ID, "Garden"))
286
287 now := time.Now()
288 _, err := inbox.state.Add(first, WhatsAppMessage{
289 ID: "first",
290 Text: "The first chat is still open.",
291 Sender: "Ana",
292 Time: now,
293 }, now)
294 require.NoError(t, err)
295
296 _, err = inbox.state.Add(second, WhatsAppMessage{
297 ID: "second",
298 Text: "The second chat is quiet.",
299 Sender: "Bruno",
300 Time: now,
301 }, now.Add(-time.Minute))
302 require.NoError(t, err)
303 require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
304 require.Equal(t, 1, test.agents[1].memory.Len())
305 require.Contains(t, test.events[1].Last("work"), "The second chat is quiet.")
306 require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
307 require.Equal(t, 2, test.agents[1].memory.Len())
308 require.Contains(t, test.events[1].Last("work"), "The first chat is still open.")
309 require.Zero(t, test.agents[0].memory.Len())
310 require.Empty(t, test.Queue(t).WhatsAppBatches)
311 }
312
313 func TestWhatsAppCatchupSeparatesChildAndRootBatches(t *testing.T) {
314 dir := t.TempDir()
315 test := newDaemonMessageTest(t, dir, newModelFake(t))
316 inbox := whatsappRouteTestInbox(t, dir)
317 garden := WhatsAppChat{
318 ID: whatsappTestFamily.String(),
319 Name: "Garden plans",
320 Group: true,
321 }
322 library := WhatsAppChat{
323 ID: whatsappTestBruno.String(),
324 Name: "Bruno",
325 }
326 root := WhatsAppChat{
327 ID: whatsappTestAna.String(),
328 Name: "Ana",
329 }
330 require.NoError(t, test.messages.routes.Set("whatsapp:"+garden.ID, "Garden"))
331 require.NoError(t, test.messages.routes.Set("whatsapp:"+library.ID, "Library"))
332 inbox.waitForCatchup()
333 whatsappRouteTestAdd(t, inbox, garden, "Check the gate.")
334 whatsappRouteTestAdd(t, inbox, library, "Check the book.")
335 whatsappRouteTestAdd(t, inbox, root, "Check the mixed chat.")
336 require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
337
338 for _, agent := range test.agents {
339 require.Zero(t, agent.memory.Len())
340 }
341
342 inbox.endCatchup()
343 require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
344
345 for _, loop := range test.loops {
346 require.Equal(t, 1, loop.memory.Len())
347 require.True(t, loop.waiting())
348 }
349
350 require.Contains(t, test.events[0].Last("work"), "Check the mixed chat.")
351 require.Contains(t, test.events[1].Last("work"), "Check the gate.")
352 require.Contains(t, test.events[2].Last("work"), "Check the book.")
353 require.Empty(t, inbox.state.Closed())
354 require.Empty(t, test.Queue(t).WhatsAppBatches)
355 }
356
357 func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) {
358 dir := t.TempDir()
359 server := newModelFake(t)
360 first := newDaemonMessageTest(t, dir, server)
361 inbox := whatsappRouteTestInbox(t, dir)
362 chat := WhatsAppChat{
363 ID: whatsappTestAna.String(),
364 Name: "Ana",
365 }
366 require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
367 require.NoError(t, os.Mkdir(first.messages.path, 0o700))
368 whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
369 require.ErrorContains(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true), "save agent messages")
370 require.Len(t, inbox.state.Closed(), 1)
371
372 for _, loop := range first.loops {
373 require.ErrorContains(t, loop.failed(), "save agent messages")
374 require.Zero(t, loop.memory.Len())
375 }
376
377 path := first.messages.path
378 first.Close(t)
379 require.NoError(t, os.Remove(path))
380
381 second := newDaemonMessageTest(t, dir, server)
382 restored := whatsappRouteTestInbox(t, dir)
383 require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
384 require.Equal(t, 1, second.agents[1].memory.Len())
385 require.True(t, second.loops[1].waiting())
386 require.Zero(t, second.agents[0].memory.Len())
387 require.Empty(t, restored.state.Closed())
388 }
389
390 func whatsappRouteTestInbox(t *testing.T, dir string) *WhatsAppInbox {
391 t.Helper()
392
393 state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName), whatsappBatchQuiet, whatsappBatchWindow)
394 require.NoError(t, err)
395
396 return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()), &ImageStore{dir: filepath.Join(dir, imageDirName)}, discardLog())
397 }
398
399 func whatsappRouteTestAdd(t *testing.T, inbox *WhatsAppInbox, chat WhatsAppChat, text string) {
400 t.Helper()
401
402 at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
403 _, err := inbox.state.Add(chat, WhatsAppMessage{
404 ID: text,
405 Sender: "Ana",
406 Text: text,
407 Time: at,
408 }, at)
409 require.NoError(t, err)
410 }