git.giacomolli.uk - onechat/blob - daemon_message_test.go

   1 package main
   2
   3 import (
   4         "context"
   5         "encoding/json"
   6         "errors"
   7         "net/http"
   8         "os"
   9         "path/filepath"
  10         "sync"
  11         "testing"
  12         "time"
  13
  14         "github.com/stretchr/testify/require"
  15 )
  16
  17 func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
  18         server := newModelFake(t)
  19         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
  20                 text := "I will ask the user."
  21                 if call.body["model"] == "gpt-test-garden" {
  22                         text = "Approve the gate size."
  23                 }
  24
  25                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
  26         }
  27
  28         test := newDaemonMessageTest(t, t.TempDir(), server)
  29         test.Start(t, 0, 1, 2)
  30         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
  31         test.events[0].WaitFor(t, "idle", 1)
  32         test.events[1].WaitFor(t, "idle", 1)
  33
  34         require.Equal(t, "OneChat", test.events[1].LastEvent("user").Source)
  35         require.Equal(t, "Check the gate size.", test.events[1].Last("user"))
  36         require.Equal(t, "Garden", test.events[0].LastEvent("work").Source)
  37         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
  38         require.Equal(t, []string{"user", "busy", "reply", "idle"}, test.events[1].Kinds())
  39         require.Equal(t, []string{"work", "busy", "reply", "idle"}, test.events[0].Kinds())
  40         require.Equal(t, 2, server.Calls())
  41         require.Zero(t, test.agents[2].memory.Len())
  42         require.Empty(t, test.events[2].Kinds())
  43
  44         first, err := test.agents[1].memory.Message(0)
  45         require.NoError(t, err)
  46         require.Equal(t, "user", first.Kind)
  47
  48         report, err := test.agents[0].memory.Message(0)
  49         require.NoError(t, err)
  50         require.Equal(t, "work", report.Kind)
  51         require.Equal(t, "[Garden] Approve the gate size.", report.Text)
  52         require.Contains(t, test.logs.String(), "turn started name=Garden")
  53         require.Contains(t, test.logs.String(), "turn done name=Garden")
  54         require.NotContains(t, test.logs.String(), "gate size")
  55         require.Empty(t, test.Queue(t).Pending)
  56
  57         info, err := os.Stat(test.messages.path)
  58         require.NoError(t, err)
  59         require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
  60 }
  61
  62 func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
  63         server := newModelFake(t)
  64         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
  65                 text := "I will ask the user."
  66                 if call.body["model"] == "gpt-test-garden" {
  67                         text = "Approve the gate size."
  68                 }
  69
  70                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
  71         }
  72
  73         test := newDaemonMessageTest(t, t.TempDir(), server)
  74         telegram := newTelegramFake(t)
  75         bot := newTelegramBot(newTelegramAPI(telegram.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), discardLog())
  76         test.loops[0].notify = func(event Event) {
  77                 test.events[0].Add(event)
  78                 bot.Notify(event)
  79         }
  80
  81         ctx, cancel := context.WithCancel(t.Context())
  82         defer cancel()
  83
  84         done := make(chan struct{})
  85         go func() {
  86                 defer close(done)
  87
  88                 bot.send(ctx)
  89         }()
  90
  91         t.Cleanup(func() {
  92                 cancel()
  93                 <-done
  94         })
  95
  96         test.Start(t, 0, 1)
  97         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
  98         test.events[0].WaitFor(t, "idle", 1)
  99         require.Eventually(t, func() bool {
 100                 return len(telegram.Sent()) >= 2
 101         }, 10*time.Second, time.Millisecond)
 102
 103         require.Equal(t, []string{"[Garden] finished", "I will ask the user."}, telegram.Sent())
 104         require.Equal(t, subagentFinished, test.events[0].LastEvent("work").Status)
 105         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
 106 }
 107
 108 func TestQueuedAgentMessageFailureKeepsItsOperationAndCause(t *testing.T) {
 109         server := newModelFake(t)
 110         test := newDaemonMessageTest(t, t.TempDir(), server)
 111         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 112
 113         failure := errors.New("fake queued message failure")
 114         test.loops[1].fail(failure)
 115
 116         err := test.loops[1].Run(t.Context())
 117         require.ErrorIs(t, err, failure)
 118         require.ErrorContains(t, err, "check turn loop failure")
 119         require.Zero(t, server.Calls())
 120 }
 121
 122 func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.T) {
 123         testCases := []struct {
 124                 name string
 125                 text string
 126         }{
 127                 {
 128                         name: "sends nothing for an empty final reply",
 129                         text: "",
 130                 },
 131                 {
 132                         name: "sends nothing for a whitespace final reply",
 133                         text: " \n\t ",
 134                 },
 135         }
 136
 137         for _, tc := range testCases {
 138                 t.Run(tc.name, func(t *testing.T) {
 139                         server := newModelFake(t)
 140                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 141                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(tc.text)}, 100, 0, 5))
 142                         }
 143
 144                         test := newDaemonMessageTest(t, t.TempDir(), server)
 145                         test.Start(t, 0, 1, 2)
 146                         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 147                         test.events[1].WaitFor(t, "idle", 1)
 148                         require.Equal(t, 1, server.Calls())
 149                         require.Zero(t, test.agents[0].memory.Len())
 150                         require.Empty(t, test.events[0].Kinds())
 151                         require.False(t, test.loops[0].waiting())
 152                         require.Empty(t, test.events[2].Kinds())
 153                         require.Empty(t, test.Queue(t).Pending)
 154                 })
 155         }
 156 }
 157
 158 func TestBusyChildQueuesTheNextParentMessage(t *testing.T) {
 159         server := newModelFake(t)
 160         arrived := make(chan struct{})
 161         release := make(chan struct{})
 162         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 163                 if call.number == 0 {
 164                         close(arrived)
 165                         <-release
 166                 }
 167
 168                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 169         }
 170
 171         test := newDaemonMessageTest(t, t.TempDir(), server)
 172         test.Start(t, 0, 1)
 173         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the first plan."))
 174         <-arrived
 175         require.True(t, test.loops[1].Busy())
 176         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the second plan too."))
 177         require.Equal(t, 1, server.Calls())
 178         require.Len(t, test.Queue(t).Pending, 2)
 179         close(release)
 180         test.events[1].WaitFor(t, "idle", 2)
 181         require.Equal(t, 2, server.Calls())
 182
 183         texts := turnInputTexts(t, server.Call(1))
 184         require.Equal(t, "Check the second plan too.", texts[len(texts)-1])
 185         require.Empty(t, test.Queue(t).Pending)
 186 }
 187
 188 func TestBusyParentQueuesItsChildReport(t *testing.T) {
 189         server := newModelFake(t)
 190         arrived := make(chan struct{})
 191         release := make(chan struct{})
 192         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 193                 text := "The plan is ready."
 194                 if call.number == 0 {
 195                         close(arrived)
 196                         <-release
 197
 198                         text = "I will continue."
 199                 }
 200
 201                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 202         }
 203
 204         test := newDaemonMessageTest(t, t.TempDir(), server)
 205         test.Start(t, 0, 1)
 206
 207         _, err := test.loops[0].Submit("terminal", "Start the work.")
 208         require.NoError(t, err)
 209         <-arrived
 210         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Prepare the plan."))
 211         test.events[1].WaitFor(t, "idle", 1)
 212         require.Equal(t, "[Garden] The plan is ready.", test.events[0].Last("work"))
 213         require.True(t, test.loops[0].Busy())
 214         require.Equal(t, 2, server.Calls())
 215         close(release)
 216         test.events[0].WaitFor(t, "idle", 2)
 217         require.Equal(t, 3, server.Calls())
 218
 219         texts := turnInputTexts(t, server.Call(2))
 220         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
 221         require.Empty(t, test.Queue(t).Pending)
 222 }
 223
 224 func TestChildReportsOnlyTheFinalReplyAndTakesQueuedToolInput(t *testing.T) {
 225         server := newModelFake(t)
 226         arrived := make(chan struct{})
 227         release := make(chan struct{})
 228         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 229                 if call.number == 0 {
 230                         close(arrived)
 231                         <-release
 232                         modelWriteEvents(w, modelCompletedEvent([]any{
 233                                 modelMessageItem("I will check the plan."),
 234                                 modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`),
 235                         }, 100, 0, 5))
 236
 237                         return
 238                 }
 239
 240                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan needs approval.")}, 100, 0, 5))
 241         }
 242
 243         test := newDaemonMessageTest(t, t.TempDir(), server)
 244         test.Start(t, 1)
 245         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the plan."))
 246         <-arrived
 247         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Also check the date."))
 248         close(release)
 249         test.events[1].WaitFor(t, "idle", 1)
 250         require.Equal(t, 2, server.Calls())
 251         require.Equal(t, 1, test.events[1].Count("busy"))
 252         require.Equal(t, 1, test.events[0].Count("work"))
 253         require.Equal(t, "[Garden] The plan needs approval.", test.events[0].Last("work"))
 254         require.NotContains(t, test.events[0].Last("work"), "I will check")
 255
 256         texts := turnInputTexts(t, server.Call(1))
 257         require.Equal(t, "Also check the date.", texts[len(texts)-1])
 258
 259         queue := test.Queue(t)
 260         require.Len(t, queue.Pending, 1)
 261         require.Equal(t, "work", queue.Pending[0].Kind)
 262 }
 263
 264 func TestQueuedParentMessageSurvivesDaemonRestart(t *testing.T) {
 265         dir := t.TempDir()
 266         server := newModelFake(t)
 267         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 268                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 269         }
 270
 271         first := newDaemonMessageTest(t, dir, server)
 272         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Remember the garden plan."))
 273         require.Equal(t, 1, first.agents[1].memory.Len())
 274         first.Close(t)
 275
 276         second := newDaemonMessageTest(t, dir, server)
 277         require.Equal(t, 1, second.agents[1].memory.Len())
 278         require.Equal(t, "OneChat", second.events[1].LastEvent("user").Source)
 279         second.Start(t, 0, 1)
 280         second.events[1].WaitFor(t, "idle", 1)
 281         require.Equal(t, 1, server.Calls())
 282         require.Empty(t, second.Queue(t).Pending)
 283         second.Close(t)
 284
 285         third := newDaemonMessageTest(t, dir, server)
 286         require.Equal(t, 1, third.agents[1].memory.Len())
 287         require.False(t, third.loops[1].waiting())
 288         require.Empty(t, third.events[1].Kinds())
 289 }
 290
 291 func TestQueuedChildReportSurvivesDaemonRestart(t *testing.T) {
 292         dir := t.TempDir()
 293         server := newModelFake(t)
 294         first := newDaemonMessageTest(t, dir, server)
 295         first.Start(t, 1)
 296         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
 297         first.events[1].WaitFor(t, "idle", 1)
 298         require.Equal(t, "[Garden] a line", first.events[0].Last("work"))
 299         require.Len(t, first.Queue(t).Pending, 1)
 300         first.Close(t)
 301
 302         second := newDaemonMessageTest(t, dir, server)
 303         require.Equal(t, 1, second.agents[0].memory.Len())
 304         require.False(t, second.loops[1].waiting())
 305         second.Start(t, 0, 1)
 306         second.events[0].WaitFor(t, "idle", 1)
 307         require.Equal(t, 2, server.Calls())
 308         require.Equal(t, 2, second.agents[0].memory.Len())
 309         require.Empty(t, second.Queue(t).Pending)
 310         second.Close(t)
 311
 312         third := newDaemonMessageTest(t, dir, server)
 313         require.Equal(t, 2, third.agents[0].memory.Len())
 314         require.False(t, third.loops[0].waiting())
 315         require.Empty(t, third.events[0].Kinds())
 316 }
 317
 318 func TestInterruptedChildDeliveryResumesAfterDaemonRestart(t *testing.T) {
 319         dir := t.TempDir()
 320         server := newModelFake(t)
 321         arrived := make(chan struct{})
 322         release := make(chan struct{})
 323         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 324                 if call.number == 0 {
 325                         close(arrived)
 326                         <-release
 327                 }
 328
 329                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 330         }
 331
 332         first := newDaemonMessageTest(t, dir, server)
 333         first.Start(t, 1)
 334         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Finish the garden plan."))
 335         <-arrived
 336         first.Close(t)
 337         close(release)
 338
 339         second := newDaemonMessageTest(t, dir, server)
 340         require.Equal(t, 1, second.agents[1].memory.Len())
 341         second.Start(t, 1)
 342         second.events[1].WaitFor(t, "idle", 1)
 343         require.Equal(t, 2, server.Calls())
 344         require.Equal(t, 1, second.agents[1].memory.Len())
 345         require.Empty(t, second.Queue(t).Pending)
 346 }
 347
 348 func TestInterruptedParentReportResumesAfterDaemonRestart(t *testing.T) {
 349         dir := t.TempDir()
 350         server := newModelFake(t)
 351         arrived := make(chan struct{})
 352         release := make(chan struct{})
 353         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 354                 if call.number == 1 {
 355                         close(arrived)
 356                         <-release
 357                 }
 358
 359                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan is ready.")}, 100, 0, 5))
 360         }
 361
 362         first := newDaemonMessageTest(t, dir, server)
 363         first.Start(t, 0, 1)
 364         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
 365         <-arrived
 366         first.Close(t)
 367         close(release)
 368
 369         second := newDaemonMessageTest(t, dir, server)
 370         require.Equal(t, 1, second.agents[0].memory.Len())
 371         require.False(t, second.loops[1].waiting())
 372         second.Start(t, 0, 1)
 373         second.events[0].WaitFor(t, "idle", 1)
 374         require.Equal(t, 3, server.Calls())
 375         require.Equal(t, 2, second.agents[0].memory.Len())
 376         require.Empty(t, second.Queue(t).Pending)
 377
 378         texts := turnInputTexts(t, server.Call(2))
 379         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
 380 }
 381
 382 func TestAgentMessageWriteFailureStopsDelivery(t *testing.T) {
 383         server := newModelFake(t)
 384         test := newDaemonMessageTest(t, t.TempDir(), server)
 385         require.NoError(t, os.Mkdir(test.messages.path, 0o700))
 386         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."), "save agent messages")
 387         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Do not send again."), "save agent messages")
 388
 389         for _, loop := range test.loops {
 390                 require.ErrorContains(t, loop.failed(), "save agent messages")
 391                 require.Zero(t, loop.memory.Len())
 392         }
 393 }
 394
 395 func TestAgentMessageSnapshotBlocksDelivery(t *testing.T) {
 396         server := newModelFake(t)
 397         test := newDaemonMessageTest(t, t.TempDir(), server)
 398         entered := make(chan struct{})
 399         release := make(chan struct{})
 400
 401         var once sync.Once
 402
 403         defer once.Do(func() {
 404                 close(release)
 405         })
 406
 407         snapshot := make(chan error, 1)
 408         go func() {
 409                 snapshot <- test.messages.Snapshot(func() error {
 410                         return daemonSnapshot(t.Context(), test.loops, func() error {
 411                                 close(entered)
 412                                 <-release
 413
 414                                 return nil
 415                         })
 416                 })
 417         }()
 418
 419         <-entered
 420
 421         sent := make(chan error, 1)
 422         started := make(chan struct{})
 423         go func() {
 424                 close(started)
 425
 426                 sent <- test.messages.Tell("OneChat", "Garden", "Record the plan after the copy.")
 427         }()
 428
 429         <-started
 430         require.Never(t, func() bool {
 431                 return len(sent) > 0
 432         }, 50*time.Millisecond, time.Millisecond)
 433
 434         once.Do(func() {
 435                 close(release)
 436         })
 437         require.NoError(t, <-snapshot)
 438         require.NoError(t, <-sent)
 439         require.Equal(t, 1, test.agents[1].memory.Len())
 440         require.Len(t, test.Queue(t).Pending, 1)
 441 }
 442
 443 func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) {
 444         server := newModelFake(t)
 445
 446         test := newDaemonMessageTest(t, t.TempDir(), server)
 447         testCases := []struct {
 448                 name   string
 449                 source string
 450                 target string
 451         }{
 452                 {
 453                         name:   "no sibling messages",
 454                         source: "Garden",
 455                         target: "Library",
 456                 },
 457                 {
 458                         name:   "no arbitrary child reports",
 459                         source: "Garden",
 460                         target: "OneChat",
 461                 },
 462                 {
 463                         name:   "no self messages",
 464                         source: "OneChat",
 465                         target: "OneChat",
 466                 },
 467                 {
 468                         name:   "refuses an unknown sender",
 469                         source: "Other",
 470                         target: "Garden",
 471                 },
 472                 {
 473                         name:   "refuses an unknown recipient",
 474                         source: "OneChat",
 475                         target: "Other",
 476                 },
 477         }
 478
 479         for _, tc := range testCases {
 480                 t.Run(tc.name, func(t *testing.T) {
 481                         require.ErrorContains(t, test.messages.Tell(tc.source, tc.target, "Do not deliver."), "not a child")
 482                 })
 483         }
 484
 485         for _, agent := range test.agents {
 486                 require.Zero(t, agent.memory.Len())
 487         }
 488 }
 489
 490 func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) {
 491         server := newModelFake(t)
 492         testCases := []struct {
 493                 name string
 494                 file string
 495         }{
 496                 {
 497                         name: "stops startup for broken JSON",
 498                         file: "{",
 499                 },
 500                 {
 501                         name: "stops startup for unknown sender",
 502                         file: `{"next":1,"pending":[{"id":1,"source":"Other","target":"Garden","kind":"user"}]}`,
 503                 },
 504                 {
 505                         name: "stops startup for sibling delivery",
 506                         file: `{"next":1,"pending":[{"id":1,"source":"Library","target":"Garden","kind":"user"}]}`,
 507                 },
 508                 {
 509                         name: "stops startup for invalid sequence",
 510                         file: `{"next":0,"pending":[{"id":1,"source":"OneChat","target":"Garden","kind":"user"}]}`,
 511                 },
 512         }
 513
 514         for _, tc := range testCases {
 515                 t.Run(tc.name, func(t *testing.T) {
 516                         dir := t.TempDir()
 517                         test := newDaemonMessageTest(t, dir, server)
 518                         path := test.messages.path
 519                         test.Close(t)
 520                         require.NoError(t, os.WriteFile(path, []byte(tc.file), 0o600))
 521
 522                         agents, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
 523                         require.NoError(t, err)
 524
 525                         defer func() {
 526                                 require.NoError(t, agents.Close())
 527                         }()
 528
 529                         var loops []*TurnLoop
 530
 531                         for _, agent := range agents {
 532                                 loop, err := agent.Loop(nil, func(Event) {
 533                                 })
 534                                 require.NoError(t, err)
 535
 536                                 loops = append(loops, loop)
 537                         }
 538
 539                         _, err = openDaemonMessages(loops)
 540                         require.Error(t, err)
 541
 542                         for _, agent := range agents {
 543                                 require.Zero(t, agent.memory.Len())
 544                         }
 545                 })
 546         }
 547 }
 548
 549 func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
 550         server := newModelFake(t)
 551         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 552                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 553         }
 554
 555         test := newDaemonMessageTest(t, t.TempDir(), server)
 556         child := test.agents[1]
 557         at := child.clock().Add(time.Minute).Format(scheduleAtLayout)
 558         output := test.loops[1].tools.run(t.Context(), "schedule_set", `{"text":"Check the plants.","at":"`+at+`","cron":null}`)
 559         require.Contains(t, output, "Schedule 1 set.")
 560         require.Equal(t, "Garden", child.schedules.sorted()[0].Owner)
 561
 562         child.schedules.clock = func() time.Time {
 563                 return child.clock().Add(time.Minute)
 564         }
 565         require.ErrorContains(t, child.schedules.deliver(t.Context(), test.loops[0]), "does not match")
 566         require.NoError(t, child.schedules.deliver(t.Context(), test.loops[1]))
 567         test.Start(t, 0, 1, 2)
 568         test.events[1].WaitFor(t, "idle", 1)
 569         require.Equal(t, 1, server.Calls())
 570         require.Equal(t, child.agent.CacheKey(), server.Call(0).body["prompt_cache_key"])
 571         require.Zero(t, test.agents[0].memory.Len())
 572         require.Zero(t, test.agents[2].memory.Len())
 573         require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.")
 574         require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source)
 575 }
 576
 577 func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) {
 578         testCases := []struct {
 579                 name string
 580                 chat WhatsAppChat
 581         }{
 582                 {
 583                         name: "routes a private chat only to its child",
 584                         chat: WhatsAppChat{
 585                                 ID:   whatsappTestAna.String(),
 586                                 Name: "Ana",
 587                         },
 588                 },
 589                 {
 590                         name: "routes an unmentioned group only to its child",
 591                         chat: WhatsAppChat{
 592                                 ID:    whatsappTestFamily.String(),
 593                                 Name:  "Garden plans",
 594                                 Group: true,
 595                         },
 596                 },
 597         }
 598
 599         for _, tc := range testCases {
 600                 t.Run(tc.name, func(t *testing.T) {
 601                         server := newModelFake(t)
 602                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 603                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 604                         }
 605
 606                         dir := t.TempDir()
 607                         test := newDaemonMessageTest(t, dir, server)
 608                         inbox := whatsappRouteTestInbox(t, dir)
 609                         require.NoError(t, test.messages.routes.Set("whatsapp:"+tc.chat.ID, "Garden"))
 610                         whatsappRouteTestAdd(t, inbox, tc.chat, "The gate needs a new latch.")
 611                         test.Start(t, 0, 1, 2)
 612                         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 613                         test.events[1].WaitFor(t, "idle", 1)
 614                         require.Equal(t, 1, server.Calls())
 615                         require.Equal(t, "gpt-test-garden", server.Call(0).body["model"])
 616                         require.Contains(t, string(server.Call(0).raw), "The gate needs a new latch.")
 617                         require.Zero(t, test.agents[0].memory.Len())
 618                         require.Empty(t, test.events[0].Kinds())
 619                         require.Zero(t, test.agents[2].memory.Len())
 620                         require.False(t, test.loops[0].waiting())
 621                         require.Equal(t, whatsappSource, test.events[1].LastEvent("work").Source)
 622                         require.Empty(t, test.events[1].LastEvent("work").Status)
 623                         require.Equal(t, 1, test.agents[1].memory.Len())
 624                         require.Empty(t, inbox.state.Closed())
 625                         require.Empty(t, test.Queue(t).Pending)
 626                         require.Contains(t, test.logs.String(), "batch routed name=Garden kind=whatsapp id=1 to=Garden")
 627                         require.NotContains(t, test.logs.String(), tc.chat.ID)
 628                         require.NotContains(t, test.logs.String(), "new latch")
 629                 })
 630         }
 631 }
 632
 633 func TestWhatsAppUnrouteRestoresRootDelivery(t *testing.T) {
 634         dir := t.TempDir()
 635         test := newDaemonMessageTest(t, dir, newModelFake(t))
 636         inbox := whatsappRouteTestInbox(t, dir)
 637         chat := WhatsAppChat{
 638                 ID:   whatsappTestAna.String(),
 639                 Name: "Ana",
 640         }
 641
 642         source := "whatsapp:" + chat.ID
 643         require.NoError(t, test.messages.routes.Set(source, "Garden"))
 644         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
 645         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 646         require.Zero(t, test.agents[0].memory.Len())
 647         require.True(t, test.loops[1].waiting())
 648         require.NoError(t, test.messages.routes.Set(source, ""))
 649         whatsappRouteTestAdd(t, inbox, chat, "Please check the other plan.")
 650         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 651         require.Equal(t, 1, test.agents[0].memory.Len())
 652         require.Equal(t, 1, test.agents[1].memory.Len())
 653         require.True(t, test.loops[0].waiting())
 654         require.Contains(t, test.events[0].Last("work"), "Please check the other plan.")
 655         require.NotContains(t, test.events[0].Last("work"), "Keep the gate plan.")
 656         require.Empty(t, inbox.state.Closed())
 657 }
 658
 659 func TestMissingWhatsAppRouteChildFallsBackToRoot(t *testing.T) {
 660         dir := t.TempDir()
 661         test := newDaemonMessageTest(t, dir, newModelFake(t))
 662         inbox := whatsappRouteTestInbox(t, dir)
 663         chat := WhatsAppChat{
 664                 ID:   whatsappTestAna.String(),
 665                 Name: "Ana",
 666         }
 667
 668         require.NoError(t, test.messages.routes.Set("whatsapp:"+chat.ID, "Missing"))
 669         whatsappRouteTestAdd(t, inbox, chat, "Please check the plan.")
 670         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 671         require.Equal(t, 1, test.agents[0].memory.Len())
 672         require.True(t, test.loops[0].waiting())
 673         require.Zero(t, test.agents[1].memory.Len())
 674         require.False(t, test.loops[1].waiting())
 675         require.Empty(t, test.events[0].LastEvent("work").Status)
 676         require.Equal(t, whatsappSource, test.events[0].LastEvent("work").Source)
 677         require.Empty(t, inbox.state.Closed())
 678 }
 679
 680 func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) {
 681         testCases := []struct {
 682                 name   string
 683                 finish bool
 684         }{
 685                 {
 686                         name:   "queued turn resumes",
 687                         finish: false,
 688                 },
 689                 {
 690                         name:   "finished turn stays finished",
 691                         finish: true,
 692                 },
 693         }
 694
 695         for _, tc := range testCases {
 696                 t.Run(tc.name, func(t *testing.T) {
 697                         dir := t.TempDir()
 698                         server := newModelFake(t)
 699                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 700                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 701                         }
 702
 703                         first := newDaemonMessageTest(t, dir, server)
 704                         inbox := whatsappRouteTestInbox(t, dir)
 705                         chat := WhatsAppChat{
 706                                 ID:   whatsappTestAna.String(),
 707                                 Name: "Ana",
 708                         }
 709
 710                         source := "whatsapp:" + chat.ID
 711                         require.NoError(t, first.messages.routes.Set(source, "Garden"))
 712                         whatsappRouteTestAdd(t, inbox, chat, "Remember the gate plan.")
 713                         require.NoError(t, inbox.state.CloseAll())
 714
 715                         batch := inbox.state.Closed()[0]
 716                         routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
 717                         require.NoError(t, err)
 718                         require.True(t, routed)
 719
 720                         if tc.finish {
 721                                 first.Start(t, 1)
 722                                 first.events[1].WaitFor(t, "idle", 1)
 723                                 require.Empty(t, first.Queue(t).Pending)
 724                         }
 725
 726                         first.Close(t)
 727
 728                         second := newDaemonMessageTest(t, dir, server)
 729                         require.Equal(t, source+" -> Garden", second.messages.routes.List())
 730                         require.Equal(t, 1, second.agents[1].memory.Len())
 731                         require.Equal(t, !tc.finish, second.loops[1].waiting())
 732                         require.NoError(t, second.messages.routes.Set(source, ""))
 733
 734                         restored := whatsappRouteTestInbox(t, dir)
 735                         require.Len(t, restored.state.Closed(), 1)
 736                         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
 737                         require.Empty(t, restored.state.Closed())
 738                         require.Zero(t, second.agents[0].memory.Len())
 739                         require.Equal(t, 1, second.agents[1].memory.Len())
 740
 741                         if !tc.finish {
 742                                 second.Start(t, 1)
 743                                 second.events[1].WaitFor(t, "idle", 1)
 744                         }
 745
 746                         require.Equal(t, 1, server.Calls())
 747                         require.Empty(t, second.Queue(t).Pending)
 748                         second.Close(t)
 749
 750                         third := newDaemonMessageTest(t, dir, server)
 751                         require.False(t, third.loops[1].waiting())
 752                         require.Empty(t, third.events[1].Kinds())
 753                         require.Equal(t, 1, third.agents[1].memory.Len())
 754                 })
 755         }
 756 }
 757
 758 func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) {
 759         testCases := []struct {
 760                 name string
 761                 id   *int
 762         }{
 763                 {
 764                         name: "recovers one message before its log ID is stored",
 765                         id:   nil,
 766                 },
 767                 {
 768                         name: "recovers one message after its log ID is stored",
 769                         id:   new(0),
 770                 },
 771         }
 772
 773         for _, tc := range testCases {
 774                 t.Run(tc.name, func(t *testing.T) {
 775                         dir := t.TempDir()
 776                         server := newModelFake(t)
 777                         first := newDaemonMessageTest(t, dir, server)
 778                         path := first.messages.path
 779                         first.Close(t)
 780
 781                         file := DaemonMessageFile{
 782                                 Next:            1,
 783                                 WhatsAppBatches: []int{1},
 784                                 Pending: []DaemonMessage{{
 785                                         ID:            1,
 786                                         Source:        "OneChat",
 787                                         Target:        "Garden",
 788                                         Kind:          "work",
 789                                         Text:          "[WhatsApp] Keep the gate plan.",
 790                                         LogID:         tc.id,
 791                                         WhatsAppBatch: 1,
 792                                 }},
 793                         }
 794
 795                         data, err := json.Marshal(file)
 796                         require.NoError(t, err)
 797                         require.NoError(t, os.WriteFile(path, data, 0o600))
 798
 799                         second := newDaemonMessageTest(t, dir, server)
 800                         require.Equal(t, 1, second.agents[1].memory.Len())
 801                         require.True(t, second.loops[1].waiting())
 802                         require.Equal(t, "[WhatsApp] Keep the gate plan.", second.events[1].Last("work"))
 803                         require.Equal(t, whatsappSource, second.events[1].LastEvent("work").Source)
 804                         second.Close(t)
 805
 806                         third := newDaemonMessageTest(t, dir, server)
 807                         require.Equal(t, 1, third.agents[1].memory.Len())
 808                         require.True(t, third.loops[1].waiting())
 809                         require.Zero(t, third.agents[0].memory.Len())
 810                 })
 811         }
 812 }
 813
 814 func TestQueuedWhatsAppBatchUsesRootIfItsChildIsRemoved(t *testing.T) {
 815         dir := t.TempDir()
 816         server := newModelFake(t)
 817         first := newDaemonMessageTest(t, dir, server)
 818         inbox := whatsappRouteTestInbox(t, dir)
 819         chat := WhatsAppChat{
 820                 ID:   whatsappTestAna.String(),
 821                 Name: "Ana",
 822         }
 823
 824         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
 825         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
 826         require.NoError(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true))
 827         first.Close(t)
 828         require.NoError(t, os.RemoveAll(filepath.Join(dir, daemonAgentDirName, "children", "Garden")))
 829
 830         second := newDaemonMessageTest(t, dir, server)
 831         require.Equal(t, 1, second.agents[0].memory.Len())
 832         require.True(t, second.loops[0].waiting())
 833         require.Equal(t, whatsappSource, second.events[0].LastEvent("work").Source)
 834         require.Empty(t, second.events[0].LastEvent("work").Status)
 835         require.Contains(t, second.events[0].Last("work"), "Keep the gate plan.")
 836         require.False(t, second.loops[1].waiting())
 837         second.Close(t)
 838
 839         third := newDaemonMessageTest(t, dir, server)
 840         require.Equal(t, 1, third.agents[0].memory.Len())
 841         require.True(t, third.loops[0].waiting())
 842 }
 843
 844 func TestRoutedWhatsAppBatchesCanCloseOutOfNumberOrder(t *testing.T) {
 845         dir := t.TempDir()
 846         test := newDaemonMessageTest(t, dir, newModelFake(t))
 847         inbox := whatsappRouteTestInbox(t, dir)
 848         first := WhatsAppChat{
 849                 ID:   whatsappTestAna.String(),
 850                 Name: "Ana",
 851         }
 852
 853         second := WhatsAppChat{
 854                 ID:    whatsappTestFamily.String(),
 855                 Name:  "Garden plans",
 856                 Group: true,
 857         }
 858
 859         require.NoError(t, test.messages.routes.Set("whatsapp:"+first.ID, "Garden"))
 860         require.NoError(t, test.messages.routes.Set("whatsapp:"+second.ID, "Garden"))
 861
 862         now := time.Now()
 863         _, err := inbox.state.Add(first, WhatsAppMessage{
 864                 ID:     "first",
 865                 Text:   "The first chat is still open.",
 866                 Sender: "Ana",
 867                 Time:   now,
 868         }, now)
 869         require.NoError(t, err)
 870
 871         _, err = inbox.state.Add(second, WhatsAppMessage{
 872                 ID:     "second",
 873                 Text:   "The second chat is quiet.",
 874                 Sender: "Bruno",
 875                 Time:   now,
 876         }, now.Add(-time.Minute))
 877         require.NoError(t, err)
 878         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
 879         require.Equal(t, 1, test.agents[1].memory.Len())
 880         require.Contains(t, test.events[1].Last("work"), "The second chat is quiet.")
 881         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 882         require.Equal(t, 2, test.agents[1].memory.Len())
 883         require.Contains(t, test.events[1].Last("work"), "The first chat is still open.")
 884         require.Zero(t, test.agents[0].memory.Len())
 885         require.Empty(t, test.Queue(t).WhatsAppBatches)
 886 }
 887
 888 func TestWhatsAppCatchupSeparatesChildAndRootBatches(t *testing.T) {
 889         dir := t.TempDir()
 890         test := newDaemonMessageTest(t, dir, newModelFake(t))
 891         inbox := whatsappRouteTestInbox(t, dir)
 892         garden := WhatsAppChat{
 893                 ID:    whatsappTestFamily.String(),
 894                 Name:  "Garden plans",
 895                 Group: true,
 896         }
 897
 898         library := WhatsAppChat{
 899                 ID:   whatsappTestBruno.String(),
 900                 Name: "Bruno",
 901         }
 902
 903         root := WhatsAppChat{
 904                 ID:   whatsappTestAna.String(),
 905                 Name: "Ana",
 906         }
 907
 908         require.NoError(t, test.messages.routes.Set("whatsapp:"+garden.ID, "Garden"))
 909         require.NoError(t, test.messages.routes.Set("whatsapp:"+library.ID, "Library"))
 910         inbox.waitForCatchup()
 911         whatsappRouteTestAdd(t, inbox, garden, "Check the gate.")
 912         whatsappRouteTestAdd(t, inbox, library, "Check the book.")
 913         whatsappRouteTestAdd(t, inbox, root, "Check the mixed chat.")
 914         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
 915
 916         for _, agent := range test.agents {
 917                 require.Zero(t, agent.memory.Len())
 918         }
 919
 920         inbox.endCatchup()
 921         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
 922
 923         for _, loop := range test.loops {
 924                 require.Equal(t, 1, loop.memory.Len())
 925                 require.True(t, loop.waiting())
 926         }
 927
 928         require.Contains(t, test.events[0].Last("work"), "Check the mixed chat.")
 929         require.Contains(t, test.events[1].Last("work"), "Check the gate.")
 930         require.Contains(t, test.events[2].Last("work"), "Check the book.")
 931         require.Empty(t, inbox.state.Closed())
 932         require.Empty(t, test.Queue(t).WhatsAppBatches)
 933 }
 934
 935 func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) {
 936         dir := t.TempDir()
 937         server := newModelFake(t)
 938         first := newDaemonMessageTest(t, dir, server)
 939         inbox := whatsappRouteTestInbox(t, dir)
 940         chat := WhatsAppChat{
 941                 ID:   whatsappTestAna.String(),
 942                 Name: "Ana",
 943         }
 944
 945         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
 946         require.NoError(t, os.Mkdir(first.messages.path, 0o700))
 947         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
 948         require.ErrorContains(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true), "save agent messages")
 949         require.Len(t, inbox.state.Closed(), 1)
 950
 951         for _, loop := range first.loops {
 952                 require.ErrorContains(t, loop.failed(), "save agent messages")
 953                 require.Zero(t, loop.memory.Len())
 954         }
 955
 956         path := first.messages.path
 957         first.Close(t)
 958         require.NoError(t, os.Remove(path))
 959
 960         second := newDaemonMessageTest(t, dir, server)
 961         restored := whatsappRouteTestInbox(t, dir)
 962         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
 963         require.Equal(t, 1, second.agents[1].memory.Len())
 964         require.True(t, second.loops[1].waiting())
 965         require.Zero(t, second.agents[0].memory.Len())
 966         require.Empty(t, restored.state.Closed())
 967 }
 968
 969 func whatsappRouteTestInbox(t *testing.T, dir string) *WhatsAppInbox {
 970         t.Helper()
 971
 972         state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName))
 973         require.NoError(t, err)
 974
 975         return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()), &ImageStore{dir: filepath.Join(dir, imageDirName)}, discardLog())
 976 }
 977
 978 func whatsappRouteTestAdd(t *testing.T, inbox *WhatsAppInbox, chat WhatsAppChat, text string) {
 979         t.Helper()
 980
 981         at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
 982         _, err := inbox.state.Add(chat, WhatsAppMessage{
 983                 ID:     text,
 984                 Sender: "Ana",
 985                 Text:   text,
 986                 Time:   at,
 987         }, at)
 988         require.NoError(t, err)
 989 }