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: "empty final reply",
 129                         text: "",
 130                 },
 131                 {
 132                         name: "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:   "unknown sender",
 469                         source: "Other",
 470                         target: "Garden",
 471                 },
 472                 {
 473                         name:   "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: "broken JSON",
 498                         file: "{",
 499                 },
 500                 {
 501                         name: "unknown sender",
 502                         file: `{"next":1,"pending":[{"id":1,"source":"Other","target":"Garden","kind":"user"}]}`,
 503                 },
 504                 {
 505                         name: "sibling delivery",
 506                         file: `{"next":1,"pending":[{"id":1,"source":"Library","target":"Garden","kind":"user"}]}`,
 507                 },
 508                 {
 509                         name: "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: "private chat",
 584                         chat: WhatsAppChat{
 585                                 ID:   whatsappTestAna.String(),
 586                                 Name: "Ana",
 587                         },
 588                 },
 589                 {
 590                         name: "group without a mention",
 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         require.NoError(t, test.messages.routes.Set("whatsapp:"+chat.ID, "Missing"))
 668         whatsappRouteTestAdd(t, inbox, chat, "Please check the plan.")
 669         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 670         require.Equal(t, 1, test.agents[0].memory.Len())
 671         require.True(t, test.loops[0].waiting())
 672         require.Zero(t, test.agents[1].memory.Len())
 673         require.False(t, test.loops[1].waiting())
 674         require.Empty(t, test.events[0].LastEvent("work").Status)
 675         require.Equal(t, whatsappSource, test.events[0].LastEvent("work").Source)
 676         require.Empty(t, inbox.state.Closed())
 677 }
 678
 679 func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) {
 680         testCases := []struct {
 681                 name   string
 682                 finish bool
 683         }{
 684                 {
 685                         name:   "queued turn resumes",
 686                         finish: false,
 687                 },
 688                 {
 689                         name:   "finished turn stays finished",
 690                         finish: true,
 691                 },
 692         }
 693
 694         for _, tc := range testCases {
 695                 t.Run(tc.name, func(t *testing.T) {
 696                         dir := t.TempDir()
 697                         server := newModelFake(t)
 698                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 699                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 700                         }
 701
 702                         first := newDaemonMessageTest(t, dir, server)
 703                         inbox := whatsappRouteTestInbox(t, dir)
 704                         chat := WhatsAppChat{
 705                                 ID:   whatsappTestAna.String(),
 706                                 Name: "Ana",
 707                         }
 708
 709                         source := "whatsapp:" + chat.ID
 710                         require.NoError(t, first.messages.routes.Set(source, "Garden"))
 711                         whatsappRouteTestAdd(t, inbox, chat, "Remember the gate plan.")
 712                         require.NoError(t, inbox.state.CloseAll())
 713
 714                         batch := inbox.state.Closed()[0]
 715                         routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
 716                         require.NoError(t, err)
 717                         require.True(t, routed)
 718
 719                         if tc.finish {
 720                                 first.Start(t, 1)
 721                                 first.events[1].WaitFor(t, "idle", 1)
 722                                 require.Empty(t, first.Queue(t).Pending)
 723                         }
 724
 725                         first.Close(t)
 726
 727                         second := newDaemonMessageTest(t, dir, server)
 728                         require.Equal(t, source+" -> Garden", second.messages.routes.List())
 729                         require.Equal(t, 1, second.agents[1].memory.Len())
 730                         require.Equal(t, !tc.finish, second.loops[1].waiting())
 731                         require.NoError(t, second.messages.routes.Set(source, ""))
 732
 733                         restored := whatsappRouteTestInbox(t, dir)
 734                         require.Len(t, restored.state.Closed(), 1)
 735                         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
 736                         require.Empty(t, restored.state.Closed())
 737                         require.Zero(t, second.agents[0].memory.Len())
 738                         require.Equal(t, 1, second.agents[1].memory.Len())
 739
 740                         if !tc.finish {
 741                                 second.Start(t, 1)
 742                                 second.events[1].WaitFor(t, "idle", 1)
 743                         }
 744
 745                         require.Equal(t, 1, server.Calls())
 746                         require.Empty(t, second.Queue(t).Pending)
 747                         second.Close(t)
 748
 749                         third := newDaemonMessageTest(t, dir, server)
 750                         require.False(t, third.loops[1].waiting())
 751                         require.Empty(t, third.events[1].Kinds())
 752                         require.Equal(t, 1, third.agents[1].memory.Len())
 753                 })
 754         }
 755 }
 756
 757 func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) {
 758         testCases := []struct {
 759                 name string
 760                 id   *int
 761         }{
 762                 {
 763                         name: "before log ID is stored",
 764                         id:   nil,
 765                 },
 766                 {
 767                         name: "after log ID is stored",
 768                         id:   new(0),
 769                 },
 770         }
 771
 772         for _, tc := range testCases {
 773                 t.Run(tc.name, func(t *testing.T) {
 774                         dir := t.TempDir()
 775                         server := newModelFake(t)
 776                         first := newDaemonMessageTest(t, dir, server)
 777                         path := first.messages.path
 778                         first.Close(t)
 779
 780                         file := DaemonMessageFile{
 781                                 Next:            1,
 782                                 WhatsAppBatches: []int{1},
 783                                 Pending: []DaemonMessage{{
 784                                         ID:            1,
 785                                         Source:        "OneChat",
 786                                         Target:        "Garden",
 787                                         Kind:          "work",
 788                                         Text:          "[WhatsApp] Keep the gate plan.",
 789                                         LogID:         tc.id,
 790                                         WhatsAppBatch: 1,
 791                                 }},
 792                         }
 793
 794                         data, err := json.Marshal(file)
 795                         require.NoError(t, err)
 796                         require.NoError(t, os.WriteFile(path, data, 0o600))
 797
 798                         second := newDaemonMessageTest(t, dir, server)
 799                         require.Equal(t, 1, second.agents[1].memory.Len())
 800                         require.True(t, second.loops[1].waiting())
 801                         require.Equal(t, "[WhatsApp] Keep the gate plan.", second.events[1].Last("work"))
 802                         require.Equal(t, whatsappSource, second.events[1].LastEvent("work").Source)
 803                         second.Close(t)
 804
 805                         third := newDaemonMessageTest(t, dir, server)
 806                         require.Equal(t, 1, third.agents[1].memory.Len())
 807                         require.True(t, third.loops[1].waiting())
 808                         require.Zero(t, third.agents[0].memory.Len())
 809                 })
 810         }
 811 }
 812
 813 func TestQueuedWhatsAppBatchUsesRootIfItsChildIsRemoved(t *testing.T) {
 814         dir := t.TempDir()
 815         server := newModelFake(t)
 816         first := newDaemonMessageTest(t, dir, server)
 817         inbox := whatsappRouteTestInbox(t, dir)
 818         chat := WhatsAppChat{
 819                 ID:   whatsappTestAna.String(),
 820                 Name: "Ana",
 821         }
 822         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
 823         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
 824         require.NoError(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true))
 825         first.Close(t)
 826         require.NoError(t, os.RemoveAll(filepath.Join(dir, daemonAgentDirName, "children", "Garden")))
 827
 828         second := newDaemonMessageTest(t, dir, server)
 829         require.Equal(t, 1, second.agents[0].memory.Len())
 830         require.True(t, second.loops[0].waiting())
 831         require.Equal(t, whatsappSource, second.events[0].LastEvent("work").Source)
 832         require.Empty(t, second.events[0].LastEvent("work").Status)
 833         require.Contains(t, second.events[0].Last("work"), "Keep the gate plan.")
 834         require.False(t, second.loops[1].waiting())
 835         second.Close(t)
 836
 837         third := newDaemonMessageTest(t, dir, server)
 838         require.Equal(t, 1, third.agents[0].memory.Len())
 839         require.True(t, third.loops[0].waiting())
 840 }
 841
 842 func TestRoutedWhatsAppBatchesCanCloseOutOfNumberOrder(t *testing.T) {
 843         dir := t.TempDir()
 844         test := newDaemonMessageTest(t, dir, newModelFake(t))
 845         inbox := whatsappRouteTestInbox(t, dir)
 846         first := WhatsAppChat{
 847                 ID:   whatsappTestAna.String(),
 848                 Name: "Ana",
 849         }
 850
 851         second := WhatsAppChat{
 852                 ID:    whatsappTestFamily.String(),
 853                 Name:  "Garden plans",
 854                 Group: true,
 855         }
 856         require.NoError(t, test.messages.routes.Set("whatsapp:"+first.ID, "Garden"))
 857         require.NoError(t, test.messages.routes.Set("whatsapp:"+second.ID, "Garden"))
 858
 859         now := time.Now()
 860         _, err := inbox.state.Add(first, WhatsAppMessage{
 861                 ID:     "first",
 862                 Text:   "The first chat is still open.",
 863                 Sender: "Ana",
 864                 Time:   now,
 865         }, now)
 866         require.NoError(t, err)
 867
 868         _, err = inbox.state.Add(second, WhatsAppMessage{
 869                 ID:     "second",
 870                 Text:   "The second chat is quiet.",
 871                 Sender: "Bruno",
 872                 Time:   now,
 873         }, now.Add(-time.Minute))
 874         require.NoError(t, err)
 875         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
 876         require.Equal(t, 1, test.agents[1].memory.Len())
 877         require.Contains(t, test.events[1].Last("work"), "The second chat is quiet.")
 878         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 879         require.Equal(t, 2, test.agents[1].memory.Len())
 880         require.Contains(t, test.events[1].Last("work"), "The first chat is still open.")
 881         require.Zero(t, test.agents[0].memory.Len())
 882         require.Empty(t, test.Queue(t).WhatsAppBatches)
 883 }
 884
 885 func TestWhatsAppCatchupSeparatesChildAndRootBatches(t *testing.T) {
 886         dir := t.TempDir()
 887         test := newDaemonMessageTest(t, dir, newModelFake(t))
 888         inbox := whatsappRouteTestInbox(t, dir)
 889         garden := WhatsAppChat{
 890                 ID:    whatsappTestFamily.String(),
 891                 Name:  "Garden plans",
 892                 Group: true,
 893         }
 894
 895         library := WhatsAppChat{
 896                 ID:   whatsappTestBruno.String(),
 897                 Name: "Bruno",
 898         }
 899
 900         root := WhatsAppChat{
 901                 ID:   whatsappTestAna.String(),
 902                 Name: "Ana",
 903         }
 904         require.NoError(t, test.messages.routes.Set("whatsapp:"+garden.ID, "Garden"))
 905         require.NoError(t, test.messages.routes.Set("whatsapp:"+library.ID, "Library"))
 906         inbox.waitForCatchup()
 907         whatsappRouteTestAdd(t, inbox, garden, "Check the gate.")
 908         whatsappRouteTestAdd(t, inbox, library, "Check the book.")
 909         whatsappRouteTestAdd(t, inbox, root, "Check the mixed chat.")
 910         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
 911
 912         for _, agent := range test.agents {
 913                 require.Zero(t, agent.memory.Len())
 914         }
 915
 916         inbox.endCatchup()
 917         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
 918
 919         for _, loop := range test.loops {
 920                 require.Equal(t, 1, loop.memory.Len())
 921                 require.True(t, loop.waiting())
 922         }
 923
 924         require.Contains(t, test.events[0].Last("work"), "Check the mixed chat.")
 925         require.Contains(t, test.events[1].Last("work"), "Check the gate.")
 926         require.Contains(t, test.events[2].Last("work"), "Check the book.")
 927         require.Empty(t, inbox.state.Closed())
 928         require.Empty(t, test.Queue(t).WhatsAppBatches)
 929 }
 930
 931 func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) {
 932         dir := t.TempDir()
 933         server := newModelFake(t)
 934         first := newDaemonMessageTest(t, dir, server)
 935         inbox := whatsappRouteTestInbox(t, dir)
 936         chat := WhatsAppChat{
 937                 ID:   whatsappTestAna.String(),
 938                 Name: "Ana",
 939         }
 940         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
 941         require.NoError(t, os.Mkdir(first.messages.path, 0o700))
 942         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
 943         require.ErrorContains(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true), "save agent messages")
 944         require.Len(t, inbox.state.Closed(), 1)
 945
 946         for _, loop := range first.loops {
 947                 require.ErrorContains(t, loop.failed(), "save agent messages")
 948                 require.Zero(t, loop.memory.Len())
 949         }
 950
 951         path := first.messages.path
 952         first.Close(t)
 953         require.NoError(t, os.Remove(path))
 954
 955         second := newDaemonMessageTest(t, dir, server)
 956         restored := whatsappRouteTestInbox(t, dir)
 957         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
 958         require.Equal(t, 1, second.agents[1].memory.Len())
 959         require.True(t, second.loops[1].waiting())
 960         require.Zero(t, second.agents[0].memory.Len())
 961         require.Empty(t, restored.state.Closed())
 962 }
 963
 964 func whatsappRouteTestInbox(t *testing.T, dir string) *WhatsAppInbox {
 965         t.Helper()
 966
 967         state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName))
 968         require.NoError(t, err)
 969
 970         return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()), &ImageStore{dir: filepath.Join(dir, imageDirName)}, discardLog())
 971 }
 972
 973 func whatsappRouteTestAdd(t *testing.T, inbox *WhatsAppInbox, chat WhatsAppChat, text string) {
 974         t.Helper()
 975
 976         at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
 977         _, err := inbox.state.Add(chat, WhatsAppMessage{
 978                 ID:     text,
 979                 Sender: "Ana",
 980                 Text:   text,
 981                 Time:   at,
 982         }, at)
 983         require.NoError(t, err)
 984 }