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

   1 package main
   2
   3 import (
   4         "context"
   5         "encoding/json"
   6         "errors"
   7         "fmt"
   8         "net/http"
   9         "os"
  10         "path/filepath"
  11         "strings"
  12         "sync"
  13         "testing"
  14         "time"
  15
  16         "github.com/stretchr/testify/require"
  17 )
  18
  19 func TestNotificationWaitsForTheNextTurnAfterEveryActiveToolRequest(t *testing.T) {
  20         server := newModelFake(t)
  21         arrived := make(chan struct{})
  22         release := make(chan struct{})
  23         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
  24                 if call.number == 0 {
  25                         close(arrived)
  26                         <-release
  27                         modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`)}, 100, 0, 5))
  28
  29                         return
  30                 }
  31
  32                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
  33         }
  34
  35         test := newDaemonMessageTest(t, t.TempDir(), server)
  36         test.Start(t, 0, 1, 2)
  37
  38         _, err := test.loops[0].Submit(terminalSource, "Start a tool chain.")
  39         require.NoError(t, err)
  40         <-arrived
  41
  42         before := test.agents[0].memory.Len()
  43         input := &NotifyInput{
  44                 Source: "test-check",
  45                 ID:     "event-1",
  46                 Text:   "Unique queued external data.",
  47         }
  48
  49         receipt, err := test.messages.Notify(input)
  50         require.NoError(t, err)
  51         require.Equal(t, "accepted", receipt.Status)
  52         require.Equal(t, before, test.agents[0].memory.Len())
  53         require.True(t, test.loops[0].Busy())
  54         close(release)
  55         test.events[0].WaitFor(t, "idle", 2)
  56         require.Equal(t, 3, server.Calls())
  57         require.NotContains(t, strings.Join(turnInputTexts(t, server.Call(1)), "\n"), input.Text)
  58         require.Contains(t, strings.Join(turnInputTexts(t, server.Call(2)), "\n"), input.Text)
  59         require.Equal(t, notifySource, test.events[0].LastEvent("work").Source)
  60         require.Empty(t, test.Queue(t).Pending)
  61         require.Len(t, test.Queue(t).Notifications, 1)
  62         require.Zero(t, test.agents[1].memory.Len())
  63         require.Zero(t, test.agents[2].memory.Len())
  64         require.NotContains(t, test.logs.String(), input.Source)
  65         require.NotContains(t, test.logs.String(), input.Text)
  66         require.NotContains(t, test.logs.String(), input.ID)
  67 }
  68
  69 func TestNotificationRetriesSurviveRestartBeforeAndAfterDelivery(t *testing.T) {
  70         directory := t.TempDir()
  71         server := newModelFake(t)
  72         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
  73                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
  74         }
  75
  76         input := &NotifyInput{
  77                 Source: "check",
  78                 ID:     "result-1",
  79                 Text:   "A new result.",
  80         }
  81
  82         first := newDaemonMessageTest(t, directory, server)
  83         receipt, err := first.messages.Notify(input)
  84         require.NoError(t, err)
  85         require.Zero(t, first.agents[0].memory.Len())
  86         first.Close(t)
  87
  88         second := newDaemonMessageTest(t, directory, server)
  89         duplicate, err := second.messages.Notify(input)
  90         require.NoError(t, err)
  91         require.Equal(t, receipt.ID, duplicate.ID)
  92         require.Equal(t, "duplicate", duplicate.Status)
  93         require.Zero(t, second.agents[0].memory.Len())
  94         second.Start(t, 0)
  95         second.events[0].WaitFor(t, "idle", 1)
  96         require.Empty(t, second.Queue(t).Pending)
  97         second.Close(t)
  98
  99         third := newDaemonMessageTest(t, directory, server)
 100         duplicate, err = third.messages.Notify(input)
 101         require.NoError(t, err)
 102         require.Equal(t, "duplicate", duplicate.Status)
 103         require.Equal(t, receipt.ID, duplicate.ID)
 104         require.False(t, third.loops[0].waiting())
 105         require.Equal(t, 1, server.Calls())
 106
 107         changed := *input
 108         changed.Text = "A different result."
 109         _, err = third.messages.Notify(&changed)
 110         require.ErrorContains(t, err, "conflicts")
 111         require.Len(t, third.Queue(t).Notifications, 1)
 112 }
 113
 114 func TestNotificationAcceptanceIsSerializedForConcurrentRetries(t *testing.T) {
 115         server := newModelFake(t)
 116         test := newDaemonMessageTest(t, t.TempDir(), server)
 117         input := &NotifyInput{
 118                 Source: "check",
 119                 ID:     "1",
 120                 Text:   "A result.",
 121         }
 122
 123         results := make(chan Event, 20)
 124         failures := make(chan error, 20)
 125
 126         var workers sync.WaitGroup
 127
 128         for range 20 {
 129                 workers.Add(1)
 130
 131                 go func() {
 132                         defer workers.Done()
 133
 134                         receipt, err := test.messages.Notify(input)
 135                         results <- receipt
 136
 137                         failures <- err
 138                 }()
 139         }
 140
 141         workers.Wait()
 142         close(results)
 143         close(failures)
 144
 145         accepted := 0
 146         for receipt := range results {
 147                 require.Equal(t, 1, receipt.ID)
 148
 149                 if receipt.Status == "accepted" {
 150                         accepted++
 151                 } else {
 152                         require.Equal(t, "duplicate", receipt.Status)
 153                 }
 154         }
 155
 156         for err := range failures {
 157                 require.NoError(t, err)
 158         }
 159
 160         require.Equal(t, 1, accepted)
 161         require.Len(t, test.Queue(t).Pending, 1)
 162         require.Len(t, test.Queue(t).Notifications, 1)
 163 }
 164
 165 func TestNotificationStorageFailureReturnsNoAcceptedReceipt(t *testing.T) {
 166         server := newModelFake(t)
 167         test := newDaemonMessageTest(t, t.TempDir(), server)
 168         test.messages.path = filepath.Join(t.TempDir(), "missing", "messages.json")
 169         receipt, err := test.messages.Notify(&NotifyInput{
 170                 Source: "check",
 171                 ID:     "1",
 172                 Text:   "A result.",
 173         })
 174         require.ErrorContains(t, err, "accept external event")
 175         require.Empty(t, receipt.Kind)
 176         require.False(t, test.loops[0].waiting())
 177         require.Zero(t, test.agents[0].memory.Len())
 178
 179         _, err = test.messages.Notify(&NotifyInput{
 180                 Source: "check",
 181                 ID:     "2",
 182                 Text:   "Another result.",
 183         })
 184         require.ErrorContains(t, err, "queue stopped")
 185 }
 186
 187 func TestNotificationFullStoreRejectsNewKeysWithoutRemovingReceipts(t *testing.T) {
 188         server := newModelFake(t)
 189         test := newDaemonMessageTest(t, t.TempDir(), server)
 190
 191         for number := range notifyReceipts {
 192                 test.messages.state.Notifications = append(test.messages.state.Notifications, NotifyReceipt{
 193                         Source: "check",
 194                         ID:     fmt.Sprint(number),
 195                         Digest: notifyDigest("data"),
 196                         Number: number + 1,
 197                 })
 198         }
 199
 200         test.messages.state.Next = notifyReceipts
 201         receipt, err := test.messages.Notify(&NotifyInput{
 202                 Source: "check",
 203                 ID:     "new-key",
 204                 Text:   "data",
 205         })
 206         require.ErrorContains(t, err, "store is full")
 207         require.Empty(t, receipt.Kind)
 208         require.Len(t, test.messages.state.Notifications, notifyReceipts)
 209         require.Empty(t, test.messages.state.Pending)
 210
 211         receipt, err = test.messages.Notify(&NotifyInput{
 212                 Source: "check",
 213                 ID:     "0",
 214                 Text:   "data",
 215         })
 216         require.NoError(t, err)
 217         require.Equal(t, "duplicate", receipt.Status)
 218 }
 219
 220 func TestNotificationBatchesKeepAcceptanceOrder(t *testing.T) {
 221         server := newModelFake(t)
 222         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 223                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 224         }
 225
 226         test := newDaemonMessageTest(t, t.TempDir(), server)
 227         test.agents[0].memory.summarizer = newFakeSummarizer()
 228         for number := range 21 {
 229                 _, err := test.messages.Notify(&NotifyInput{
 230                         Source: "check",
 231                         ID:     fmt.Sprint(number),
 232                         Text:   fmt.Sprintf("result-%02d", number),
 233                 })
 234                 require.NoError(t, err)
 235         }
 236
 237         test.Start(t, 0)
 238         test.events[0].WaitFor(t, "idle", 2)
 239         require.Equal(t, 2, server.Calls())
 240
 241         first := turnInputTexts(t, server.Call(0))
 242         require.Len(t, first, 22)
 243
 244         for number := range 20 {
 245                 require.Contains(t, first[number+2], fmt.Sprintf("result-%02d", number))
 246         }
 247
 248         second := turnInputTexts(t, server.Call(1))
 249         require.Contains(t, second[len(second)-1], "result-20")
 250         require.Empty(t, test.Queue(t).Pending)
 251 }
 252
 253 func TestNotificationRecoversItsPreparedLogIDWithoutAnotherAppend(t *testing.T) {
 254         for _, appended := range []bool{false, true} {
 255                 t.Run(fmt.Sprintf("log append completed %t", appended), func(t *testing.T) {
 256                         directory := t.TempDir()
 257                         server := newModelFake(t)
 258                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 259                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 260                         }
 261
 262                         first := newDaemonMessageTest(t, directory, server)
 263                         _, err := first.messages.Notify(&NotifyInput{
 264                                 Source: "check",
 265                                 ID:     "1",
 266                                 Text:   "A result.",
 267                         })
 268                         require.NoError(t, err)
 269
 270                         if appended {
 271                                 require.NoError(t, first.messages.Notifications(first.loops[0]))
 272                         } else {
 273                                 id := 0
 274                                 first.messages.state.Pending[0].LogID = &id
 275                                 require.NoError(t, first.messages.save())
 276                         }
 277
 278                         first.Close(t)
 279
 280                         second := newDaemonMessageTest(t, directory, server)
 281                         second.Start(t, 0)
 282                         second.events[0].WaitFor(t, "idle", 1)
 283                         require.Equal(t, 1, server.Calls())
 284                         require.Equal(t, 1, second.agents[0].memory.Len())
 285                         require.Empty(t, second.Queue(t).Pending)
 286                 })
 287         }
 288 }
 289
 290 func TestPreparedNotificationKeepsItsLogIDBeforeStartupWriters(t *testing.T) {
 291         testCases := []struct {
 292                 name   string
 293                 resume bool
 294         }{
 295                 {
 296                         name: "keeps its reservation before a startup user message",
 297                 },
 298                 {
 299                         name:   "keeps its reservation before a worker recovery note",
 300                         resume: true,
 301                 },
 302         }
 303
 304         for _, tc := range testCases {
 305                 t.Run(tc.name, func(t *testing.T) {
 306                         directory := t.TempDir()
 307                         server := newModelFake(t)
 308                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 309                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 310                         }
 311
 312                         first := newDaemonMessageTest(t, directory, server)
 313                         input := &NotifyInput{
 314                                 Source: "check",
 315                                 ID:     "reserved",
 316                                 Text:   "A result with a prepared log ID.",
 317                         }
 318
 319                         receipt, err := first.messages.Notify(input)
 320                         require.NoError(t, err)
 321
 322                         id := 0
 323                         first.messages.state.Pending[0].LogID = &id
 324                         require.NoError(t, first.messages.save())
 325
 326                         expected := first.Queue(t).Pending[0].Text
 327                         first.Close(t)
 328
 329                         second := newDaemonMessageTest(t, directory, server)
 330                         require.False(t, second.loops[0].waiting())
 331                         require.Empty(t, second.events[0].Kinds())
 332
 333                         if tc.resume {
 334                                 path := filepath.Join(second.agents[0].agent.dir, subagentWorkFileName)
 335                                 require.NoError(t, os.WriteFile(path, []byte(`{"running":["BlueFox"]}`), 0o600))
 336                                 require.NoError(t, second.loops[0].runs.Resume(t.Context()))
 337
 338                                 note, err := second.agents[0].memory.Message(1)
 339                                 require.NoError(t, err)
 340                                 require.Equal(t, "note", note.Kind)
 341                                 require.Contains(t, note.Text, "The daemon restarted, cutting BlueFox")
 342                         } else {
 343                                 submitted, err := second.loops[0].Submit(terminalSource, "A user message during startup.")
 344                                 require.NoError(t, err)
 345                                 require.Equal(t, 1, submitted.ID)
 346                         }
 347
 348                         restored, err := second.agents[0].memory.Message(id)
 349                         require.NoError(t, err)
 350                         require.Equal(t, "work", restored.Kind)
 351                         require.Equal(t, expected, restored.Text)
 352                         require.Len(t, second.Queue(t).Pending, 1)
 353
 354                         second.Start(t, 0)
 355                         second.events[0].WaitFor(t, "idle", 1)
 356                         require.Empty(t, second.Queue(t).Pending)
 357                         require.Equal(t, 1, server.Calls())
 358                         require.Equal(t, id, second.events[0].LastEvent("work").ID)
 359
 360                         duplicate, err := second.messages.Notify(input)
 361                         require.NoError(t, err)
 362                         require.Equal(t, "duplicate", duplicate.Status)
 363                         require.Equal(t, receipt.ID, duplicate.ID)
 364                         require.Len(t, second.Queue(t).Notifications, 1)
 365                 })
 366         }
 367 }
 368
 369 func TestNotificationBatchMakesProgressWithEscapedControlCharacters(t *testing.T) {
 370         server := newModelFake(t)
 371         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 372                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 373         }
 374
 375         test := newDaemonMessageTest(t, t.TempDir(), server)
 376         _, err := test.messages.Notify(&NotifyInput{
 377                 Source: "check",
 378                 ID:     "escaped",
 379                 Text:   "data" + strings.Repeat("\x00", notifyTextBytes-4),
 380         })
 381         require.NoError(t, err)
 382         require.Greater(t, len(test.Queue(t).Pending[0].Text), 64<<10)
 383         test.Start(t, 0)
 384         test.events[0].WaitFor(t, "idle", 1)
 385         require.Empty(t, test.Queue(t).Pending)
 386 }
 387
 388 func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
 389         server := newModelFake(t)
 390         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 391                 text := "I will ask the user."
 392                 if call.body["model"] == "gpt-test-garden" {
 393                         text = "Approve the gate size."
 394                 }
 395
 396                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 397         }
 398
 399         test := newDaemonMessageTest(t, t.TempDir(), server)
 400         test.Start(t, 0, 1, 2)
 401         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
 402         test.events[0].WaitFor(t, "idle", 1)
 403         test.events[1].WaitFor(t, "idle", 1)
 404
 405         require.Equal(t, "OneChat", test.events[1].LastEvent("user").Source)
 406         require.Equal(t, "Check the gate size.", test.events[1].Last("user"))
 407         require.Equal(t, "Garden", test.events[0].LastEvent("work").Source)
 408         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
 409         require.Equal(t, []string{"user", "busy", "reply", "idle"}, test.events[1].Kinds())
 410         require.Equal(t, []string{"work", "busy", "reply", "idle"}, test.events[0].Kinds())
 411         require.Equal(t, 2, server.Calls())
 412         require.Zero(t, test.agents[2].memory.Len())
 413         require.Empty(t, test.events[2].Kinds())
 414
 415         first, err := test.agents[1].memory.Message(0)
 416         require.NoError(t, err)
 417         require.Equal(t, "user", first.Kind)
 418
 419         report, err := test.agents[0].memory.Message(0)
 420         require.NoError(t, err)
 421         require.Equal(t, "work", report.Kind)
 422         require.Equal(t, "[Garden] Approve the gate size.", report.Text)
 423         require.Contains(t, test.logs.String(), "turn started name=Garden")
 424         require.Contains(t, test.logs.String(), "turn done name=Garden")
 425         require.NotContains(t, test.logs.String(), "gate size")
 426         require.Empty(t, test.Queue(t).Pending)
 427
 428         info, err := os.Stat(test.messages.path)
 429         require.NoError(t, err)
 430         require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
 431 }
 432
 433 func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
 434         server := newModelFake(t)
 435         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 436                 text := "I will ask the user."
 437                 if call.body["model"] == "gpt-test-garden" {
 438                         text = "Approve the gate size."
 439                 }
 440
 441                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 442         }
 443
 444         test := newDaemonMessageTest(t, t.TempDir(), server)
 445         telegram := newTelegramFake(t)
 446         bot := newTelegramBot(newTelegramAPI(telegram.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), discardLog())
 447         test.loops[0].notify = func(event Event) {
 448                 test.events[0].Add(event)
 449                 bot.Notify(event)
 450         }
 451
 452         ctx, cancel := context.WithCancel(t.Context())
 453         defer cancel()
 454
 455         done := make(chan struct{})
 456         go func() {
 457                 defer close(done)
 458
 459                 bot.send(ctx)
 460         }()
 461
 462         t.Cleanup(func() {
 463                 cancel()
 464                 <-done
 465         })
 466
 467         test.Start(t, 0, 1)
 468         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
 469         test.events[0].WaitFor(t, "idle", 1)
 470         require.Eventually(t, func() bool {
 471                 return len(telegram.Sent()) >= 2
 472         }, 10*time.Second, time.Millisecond)
 473
 474         require.Equal(t, []string{"[Garden] finished", "I will ask the user."}, telegram.Sent())
 475         require.Equal(t, subagentFinished, test.events[0].LastEvent("work").Status)
 476         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
 477 }
 478
 479 func TestQueuedAgentMessageFailureKeepsItsOperationAndCause(t *testing.T) {
 480         server := newModelFake(t)
 481         test := newDaemonMessageTest(t, t.TempDir(), server)
 482         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 483
 484         failure := errors.New("fake queued message failure")
 485         test.loops[1].fail(failure)
 486
 487         err := test.loops[1].Run(t.Context())
 488         require.ErrorIs(t, err, failure)
 489         require.ErrorContains(t, err, "check turn loop failure")
 490         require.Zero(t, server.Calls())
 491 }
 492
 493 func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.T) {
 494         testCases := []struct {
 495                 name string
 496                 text string
 497         }{
 498                 {
 499                         name: "sends nothing for an empty final reply",
 500                         text: "",
 501                 },
 502                 {
 503                         name: "sends nothing for a whitespace final reply",
 504                         text: " \n\t ",
 505                 },
 506         }
 507
 508         for _, tc := range testCases {
 509                 t.Run(tc.name, func(t *testing.T) {
 510                         server := newModelFake(t)
 511                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 512                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(tc.text)}, 100, 0, 5))
 513                         }
 514
 515                         test := newDaemonMessageTest(t, t.TempDir(), server)
 516                         test.Start(t, 0, 1, 2)
 517                         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 518                         test.events[1].WaitFor(t, "idle", 1)
 519                         require.Equal(t, 1, server.Calls())
 520                         require.Zero(t, test.agents[0].memory.Len())
 521                         require.Empty(t, test.events[0].Kinds())
 522                         require.False(t, test.loops[0].waiting())
 523                         require.Empty(t, test.events[2].Kinds())
 524                         require.Empty(t, test.Queue(t).Pending)
 525                 })
 526         }
 527 }
 528
 529 func TestBusyChildQueuesTheNextParentMessage(t *testing.T) {
 530         server := newModelFake(t)
 531         arrived := make(chan struct{})
 532         release := make(chan struct{})
 533         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 534                 if call.number == 0 {
 535                         close(arrived)
 536                         <-release
 537                 }
 538
 539                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 540         }
 541
 542         test := newDaemonMessageTest(t, t.TempDir(), server)
 543         test.Start(t, 0, 1)
 544         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the first plan."))
 545         <-arrived
 546         require.True(t, test.loops[1].Busy())
 547         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the second plan too."))
 548         require.Equal(t, 1, server.Calls())
 549         require.Len(t, test.Queue(t).Pending, 2)
 550         close(release)
 551         test.events[1].WaitFor(t, "idle", 2)
 552         require.Equal(t, 2, server.Calls())
 553
 554         texts := turnInputTexts(t, server.Call(1))
 555         require.Equal(t, "Check the second plan too.", texts[len(texts)-1])
 556         require.Empty(t, test.Queue(t).Pending)
 557 }
 558
 559 func TestBusyParentQueuesItsChildReport(t *testing.T) {
 560         server := newModelFake(t)
 561         arrived := make(chan struct{})
 562         release := make(chan struct{})
 563         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 564                 text := "The plan is ready."
 565                 if call.number == 0 {
 566                         close(arrived)
 567                         <-release
 568
 569                         text = "I will continue."
 570                 }
 571
 572                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 573         }
 574
 575         test := newDaemonMessageTest(t, t.TempDir(), server)
 576         test.Start(t, 0, 1)
 577
 578         _, err := test.loops[0].Submit("terminal", "Start the work.")
 579         require.NoError(t, err)
 580         <-arrived
 581         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Prepare the plan."))
 582         test.events[1].WaitFor(t, "idle", 1)
 583         require.Equal(t, "[Garden] The plan is ready.", test.events[0].Last("work"))
 584         require.True(t, test.loops[0].Busy())
 585         require.Equal(t, 2, server.Calls())
 586         close(release)
 587         test.events[0].WaitFor(t, "idle", 2)
 588         require.Equal(t, 3, server.Calls())
 589
 590         texts := turnInputTexts(t, server.Call(2))
 591         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
 592         require.Empty(t, test.Queue(t).Pending)
 593 }
 594
 595 func TestChildReportsOnlyTheFinalReplyAndTakesQueuedToolInput(t *testing.T) {
 596         server := newModelFake(t)
 597         arrived := make(chan struct{})
 598         release := make(chan struct{})
 599         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 600                 if call.number == 0 {
 601                         close(arrived)
 602                         <-release
 603                         modelWriteEvents(w, modelCompletedEvent([]any{
 604                                 modelMessageItem("I will check the plan."),
 605                                 modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`),
 606                         }, 100, 0, 5))
 607
 608                         return
 609                 }
 610
 611                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan needs approval.")}, 100, 0, 5))
 612         }
 613
 614         test := newDaemonMessageTest(t, t.TempDir(), server)
 615         test.Start(t, 1)
 616         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the plan."))
 617         <-arrived
 618         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Also check the date."))
 619         close(release)
 620         test.events[1].WaitFor(t, "idle", 1)
 621         require.Equal(t, 2, server.Calls())
 622         require.Equal(t, 1, test.events[1].Count("busy"))
 623         require.Equal(t, 1, test.events[0].Count("work"))
 624         require.Equal(t, "[Garden] The plan needs approval.", test.events[0].Last("work"))
 625         require.NotContains(t, test.events[0].Last("work"), "I will check")
 626
 627         texts := turnInputTexts(t, server.Call(1))
 628         require.Equal(t, "Also check the date.", texts[len(texts)-1])
 629
 630         queue := test.Queue(t)
 631         require.Len(t, queue.Pending, 1)
 632         require.Equal(t, "work", queue.Pending[0].Kind)
 633 }
 634
 635 func TestQueuedParentMessageSurvivesDaemonRestart(t *testing.T) {
 636         dir := t.TempDir()
 637         server := newModelFake(t)
 638         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 639                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 640         }
 641
 642         first := newDaemonMessageTest(t, dir, server)
 643         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Remember the garden plan."))
 644         require.Equal(t, 1, first.agents[1].memory.Len())
 645         first.Close(t)
 646
 647         second := newDaemonMessageTest(t, dir, server)
 648         require.Equal(t, 1, second.agents[1].memory.Len())
 649         require.Equal(t, "OneChat", second.events[1].LastEvent("user").Source)
 650         second.Start(t, 0, 1)
 651         second.events[1].WaitFor(t, "idle", 1)
 652         require.Equal(t, 1, server.Calls())
 653         require.Empty(t, second.Queue(t).Pending)
 654         second.Close(t)
 655
 656         third := newDaemonMessageTest(t, dir, server)
 657         require.Equal(t, 1, third.agents[1].memory.Len())
 658         require.False(t, third.loops[1].waiting())
 659         require.Empty(t, third.events[1].Kinds())
 660 }
 661
 662 func TestQueuedChildReportSurvivesDaemonRestart(t *testing.T) {
 663         dir := t.TempDir()
 664         server := newModelFake(t)
 665         first := newDaemonMessageTest(t, dir, server)
 666         first.Start(t, 1)
 667         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
 668         first.events[1].WaitFor(t, "idle", 1)
 669         require.Equal(t, "[Garden] a line", first.events[0].Last("work"))
 670         require.Len(t, first.Queue(t).Pending, 1)
 671         first.Close(t)
 672
 673         second := newDaemonMessageTest(t, dir, server)
 674         require.Equal(t, 1, second.agents[0].memory.Len())
 675         require.False(t, second.loops[1].waiting())
 676         second.Start(t, 0, 1)
 677         second.events[0].WaitFor(t, "idle", 1)
 678         require.Equal(t, 2, server.Calls())
 679         require.Equal(t, 2, second.agents[0].memory.Len())
 680         require.Empty(t, second.Queue(t).Pending)
 681         second.Close(t)
 682
 683         third := newDaemonMessageTest(t, dir, server)
 684         require.Equal(t, 2, third.agents[0].memory.Len())
 685         require.False(t, third.loops[0].waiting())
 686         require.Empty(t, third.events[0].Kinds())
 687 }
 688
 689 func TestInterruptedChildDeliveryResumesAfterDaemonRestart(t *testing.T) {
 690         dir := t.TempDir()
 691         server := newModelFake(t)
 692         arrived := make(chan struct{})
 693         release := make(chan struct{})
 694         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 695                 if call.number == 0 {
 696                         close(arrived)
 697                         <-release
 698                 }
 699
 700                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 701         }
 702
 703         first := newDaemonMessageTest(t, dir, server)
 704         first.Start(t, 1)
 705         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Finish the garden plan."))
 706         <-arrived
 707         first.Close(t)
 708         close(release)
 709
 710         second := newDaemonMessageTest(t, dir, server)
 711         require.Equal(t, 1, second.agents[1].memory.Len())
 712         second.Start(t, 1)
 713         second.events[1].WaitFor(t, "idle", 1)
 714         require.Equal(t, 2, server.Calls())
 715         require.Equal(t, 1, second.agents[1].memory.Len())
 716         require.Empty(t, second.Queue(t).Pending)
 717 }
 718
 719 func TestInterruptedParentReportResumesAfterDaemonRestart(t *testing.T) {
 720         dir := t.TempDir()
 721         server := newModelFake(t)
 722         arrived := make(chan struct{})
 723         release := make(chan struct{})
 724         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 725                 if call.number == 1 {
 726                         close(arrived)
 727                         <-release
 728                 }
 729
 730                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan is ready.")}, 100, 0, 5))
 731         }
 732
 733         first := newDaemonMessageTest(t, dir, server)
 734         first.Start(t, 0, 1)
 735         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
 736         <-arrived
 737         first.Close(t)
 738         close(release)
 739
 740         second := newDaemonMessageTest(t, dir, server)
 741         require.Equal(t, 1, second.agents[0].memory.Len())
 742         require.False(t, second.loops[1].waiting())
 743         second.Start(t, 0, 1)
 744         second.events[0].WaitFor(t, "idle", 1)
 745         require.Equal(t, 3, server.Calls())
 746         require.Equal(t, 2, second.agents[0].memory.Len())
 747         require.Empty(t, second.Queue(t).Pending)
 748
 749         texts := turnInputTexts(t, server.Call(2))
 750         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
 751 }
 752
 753 func TestAgentMessageWriteFailureStopsDelivery(t *testing.T) {
 754         server := newModelFake(t)
 755         test := newDaemonMessageTest(t, t.TempDir(), server)
 756         require.NoError(t, os.Mkdir(test.messages.path, 0o700))
 757         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."), "save agent messages")
 758         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Do not send again."), "save agent messages")
 759
 760         for _, loop := range test.loops {
 761                 require.ErrorContains(t, loop.failed(), "save agent messages")
 762                 require.Zero(t, loop.memory.Len())
 763         }
 764 }
 765
 766 func TestAgentMessageSnapshotBlocksDelivery(t *testing.T) {
 767         server := newModelFake(t)
 768         test := newDaemonMessageTest(t, t.TempDir(), server)
 769         entered := make(chan struct{})
 770         release := make(chan struct{})
 771
 772         var once sync.Once
 773
 774         defer once.Do(func() {
 775                 close(release)
 776         })
 777
 778         snapshot := make(chan error, 1)
 779         go func() {
 780                 snapshot <- test.messages.Snapshot(func() error {
 781                         return daemonSnapshot(t.Context(), test.loops, func() error {
 782                                 close(entered)
 783                                 <-release
 784
 785                                 return nil
 786                         })
 787                 })
 788         }()
 789
 790         <-entered
 791
 792         sent := make(chan error, 1)
 793         started := make(chan struct{})
 794         go func() {
 795                 close(started)
 796
 797                 sent <- test.messages.Tell("OneChat", "Garden", "Record the plan after the copy.")
 798         }()
 799
 800         <-started
 801         require.Never(t, func() bool {
 802                 return len(sent) > 0
 803         }, 50*time.Millisecond, time.Millisecond)
 804
 805         once.Do(func() {
 806                 close(release)
 807         })
 808         require.NoError(t, <-snapshot)
 809         require.NoError(t, <-sent)
 810         require.Equal(t, 1, test.agents[1].memory.Len())
 811         require.Len(t, test.Queue(t).Pending, 1)
 812 }
 813
 814 func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) {
 815         server := newModelFake(t)
 816
 817         test := newDaemonMessageTest(t, t.TempDir(), server)
 818         testCases := []struct {
 819                 name   string
 820                 source string
 821                 target string
 822         }{
 823                 {
 824                         name:   "no sibling messages",
 825                         source: "Garden",
 826                         target: "Library",
 827                 },
 828                 {
 829                         name:   "no arbitrary child reports",
 830                         source: "Garden",
 831                         target: "OneChat",
 832                 },
 833                 {
 834                         name:   "no self messages",
 835                         source: "OneChat",
 836                         target: "OneChat",
 837                 },
 838                 {
 839                         name:   "refuses an unknown sender",
 840                         source: "Other",
 841                         target: "Garden",
 842                 },
 843                 {
 844                         name:   "refuses an unknown recipient",
 845                         source: "OneChat",
 846                         target: "Other",
 847                 },
 848         }
 849
 850         for _, tc := range testCases {
 851                 t.Run(tc.name, func(t *testing.T) {
 852                         require.ErrorContains(t, test.messages.Tell(tc.source, tc.target, "Do not deliver."), "not a child")
 853                 })
 854         }
 855
 856         for _, agent := range test.agents {
 857                 require.Zero(t, agent.memory.Len())
 858         }
 859 }
 860
 861 func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) {
 862         server := newModelFake(t)
 863         testCases := []struct {
 864                 name string
 865                 file string
 866         }{
 867                 {
 868                         name: "stops startup for broken JSON",
 869                         file: "{",
 870                 },
 871                 {
 872                         name: "stops startup for unknown sender",
 873                         file: `{"next":1,"pending":[{"id":1,"source":"Other","target":"Garden","kind":"user"}]}`,
 874                 },
 875                 {
 876                         name: "stops startup for sibling delivery",
 877                         file: `{"next":1,"pending":[{"id":1,"source":"Library","target":"Garden","kind":"user"}]}`,
 878                 },
 879                 {
 880                         name: "stops startup for invalid sequence",
 881                         file: `{"next":0,"pending":[{"id":1,"source":"OneChat","target":"Garden","kind":"user"}]}`,
 882                 },
 883         }
 884
 885         for _, tc := range testCases {
 886                 t.Run(tc.name, func(t *testing.T) {
 887                         dir := t.TempDir()
 888                         test := newDaemonMessageTest(t, dir, server)
 889                         path := test.messages.path
 890                         test.Close(t)
 891                         require.NoError(t, os.WriteFile(path, []byte(tc.file), 0o600))
 892
 893                         agents, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
 894                         require.NoError(t, err)
 895
 896                         defer func() {
 897                                 require.NoError(t, agents.Close())
 898                         }()
 899
 900                         var loops []*TurnLoop
 901
 902                         for _, agent := range agents {
 903                                 loop, err := agent.Loop(nil, func(Event) {
 904                                 })
 905                                 require.NoError(t, err)
 906
 907                                 loops = append(loops, loop)
 908                         }
 909
 910                         _, err = openDaemonMessages(loops)
 911                         require.Error(t, err)
 912
 913                         for _, agent := range agents {
 914                                 require.Zero(t, agent.memory.Len())
 915                         }
 916                 })
 917         }
 918 }
 919
 920 func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
 921         server := newModelFake(t)
 922         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 923                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 924         }
 925
 926         test := newDaemonMessageTest(t, t.TempDir(), server)
 927         child := test.agents[1]
 928         at := child.clock().Add(time.Minute).Format(scheduleAtLayout)
 929         output := test.loops[1].tools.run(t.Context(), "schedule_set", `{"text":"Check the plants.","at":"`+at+`","cron":null}`)
 930         require.Contains(t, output, "Schedule 1 set.")
 931         require.Equal(t, "Garden", child.schedules.sorted()[0].Owner)
 932
 933         child.schedules.clock = func() time.Time {
 934                 return child.clock().Add(time.Minute)
 935         }
 936         require.ErrorContains(t, child.schedules.deliver(t.Context(), test.loops[0]), "does not match")
 937         require.NoError(t, child.schedules.deliver(t.Context(), test.loops[1]))
 938         test.Start(t, 0, 1, 2)
 939         test.events[1].WaitFor(t, "idle", 1)
 940         require.Equal(t, 1, server.Calls())
 941         require.Equal(t, child.agent.CacheKey(), server.Call(0).body["prompt_cache_key"])
 942         require.Zero(t, test.agents[0].memory.Len())
 943         require.Zero(t, test.agents[2].memory.Len())
 944         require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.")
 945         require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source)
 946 }
 947
 948 func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) {
 949         testCases := []struct {
 950                 name string
 951                 chat WhatsAppChat
 952         }{
 953                 {
 954                         name: "routes a private chat only to its child",
 955                         chat: WhatsAppChat{
 956                                 ID:   whatsappTestAna.String(),
 957                                 Name: "Ana",
 958                         },
 959                 },
 960                 {
 961                         name: "routes an unmentioned group only to its child",
 962                         chat: WhatsAppChat{
 963                                 ID:    whatsappTestFamily.String(),
 964                                 Name:  "Garden plans",
 965                                 Group: true,
 966                         },
 967                 },
 968         }
 969
 970         for _, tc := range testCases {
 971                 t.Run(tc.name, func(t *testing.T) {
 972                         server := newModelFake(t)
 973                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 974                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 975                         }
 976
 977                         dir := t.TempDir()
 978                         test := newDaemonMessageTest(t, dir, server)
 979                         inbox := whatsappRouteTestInbox(t, dir)
 980                         require.NoError(t, test.messages.routes.Set("whatsapp:"+tc.chat.ID, "Garden"))
 981                         whatsappRouteTestAdd(t, inbox, tc.chat, "The gate needs a new latch.")
 982                         test.Start(t, 0, 1, 2)
 983                         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 984                         test.events[1].WaitFor(t, "idle", 1)
 985                         require.Equal(t, 1, server.Calls())
 986                         require.Equal(t, "gpt-test-garden", server.Call(0).body["model"])
 987                         require.Contains(t, string(server.Call(0).raw), "The gate needs a new latch.")
 988                         require.Zero(t, test.agents[0].memory.Len())
 989                         require.Empty(t, test.events[0].Kinds())
 990                         require.Zero(t, test.agents[2].memory.Len())
 991                         require.False(t, test.loops[0].waiting())
 992                         require.Equal(t, whatsappSource, test.events[1].LastEvent("work").Source)
 993                         require.Empty(t, test.events[1].LastEvent("work").Status)
 994                         require.Equal(t, 1, test.agents[1].memory.Len())
 995                         require.Empty(t, inbox.state.Closed())
 996                         require.Empty(t, test.Queue(t).Pending)
 997                         require.Contains(t, test.logs.String(), "batch routed name=Garden kind=whatsapp id=1 to=Garden")
 998                         require.NotContains(t, test.logs.String(), tc.chat.ID)
 999                         require.NotContains(t, test.logs.String(), "new latch")
1000                 })
1001         }
1002 }
1003
1004 func TestWhatsAppUnrouteRestoresRootDelivery(t *testing.T) {
1005         dir := t.TempDir()
1006         test := newDaemonMessageTest(t, dir, newModelFake(t))
1007         inbox := whatsappRouteTestInbox(t, dir)
1008         chat := WhatsAppChat{
1009                 ID:   whatsappTestAna.String(),
1010                 Name: "Ana",
1011         }
1012
1013         source := "whatsapp:" + chat.ID
1014         require.NoError(t, test.messages.routes.Set(source, "Garden"))
1015         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
1016         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1017         require.Zero(t, test.agents[0].memory.Len())
1018         require.True(t, test.loops[1].waiting())
1019         require.NoError(t, test.messages.routes.Set(source, ""))
1020         whatsappRouteTestAdd(t, inbox, chat, "Please check the other plan.")
1021         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1022         require.Equal(t, 1, test.agents[0].memory.Len())
1023         require.Equal(t, 1, test.agents[1].memory.Len())
1024         require.True(t, test.loops[0].waiting())
1025         require.Contains(t, test.events[0].Last("work"), "Please check the other plan.")
1026         require.NotContains(t, test.events[0].Last("work"), "Keep the gate plan.")
1027         require.Empty(t, inbox.state.Closed())
1028 }
1029
1030 func TestMissingWhatsAppRouteChildFallsBackToRoot(t *testing.T) {
1031         dir := t.TempDir()
1032         test := newDaemonMessageTest(t, dir, newModelFake(t))
1033         inbox := whatsappRouteTestInbox(t, dir)
1034         chat := WhatsAppChat{
1035                 ID:   whatsappTestAna.String(),
1036                 Name: "Ana",
1037         }
1038
1039         require.NoError(t, test.messages.routes.Set("whatsapp:"+chat.ID, "Missing"))
1040         whatsappRouteTestAdd(t, inbox, chat, "Please check the plan.")
1041         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1042         require.Equal(t, 1, test.agents[0].memory.Len())
1043         require.True(t, test.loops[0].waiting())
1044         require.Zero(t, test.agents[1].memory.Len())
1045         require.False(t, test.loops[1].waiting())
1046         require.Empty(t, test.events[0].LastEvent("work").Status)
1047         require.Equal(t, whatsappSource, test.events[0].LastEvent("work").Source)
1048         require.Empty(t, inbox.state.Closed())
1049 }
1050
1051 func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) {
1052         testCases := []struct {
1053                 name   string
1054                 finish bool
1055         }{
1056                 {
1057                         name:   "queued turn resumes",
1058                         finish: false,
1059                 },
1060                 {
1061                         name:   "finished turn stays finished",
1062                         finish: true,
1063                 },
1064         }
1065
1066         for _, tc := range testCases {
1067                 t.Run(tc.name, func(t *testing.T) {
1068                         dir := t.TempDir()
1069                         server := newModelFake(t)
1070                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
1071                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1072                         }
1073
1074                         first := newDaemonMessageTest(t, dir, server)
1075                         inbox := whatsappRouteTestInbox(t, dir)
1076                         chat := WhatsAppChat{
1077                                 ID:   whatsappTestAna.String(),
1078                                 Name: "Ana",
1079                         }
1080
1081                         source := "whatsapp:" + chat.ID
1082                         require.NoError(t, first.messages.routes.Set(source, "Garden"))
1083                         whatsappRouteTestAdd(t, inbox, chat, "Remember the gate plan.")
1084                         require.NoError(t, inbox.state.CloseAll())
1085
1086                         batch := inbox.state.Closed()[0]
1087                         routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
1088                         require.NoError(t, err)
1089                         require.True(t, routed)
1090
1091                         if tc.finish {
1092                                 first.Start(t, 1)
1093                                 first.events[1].WaitFor(t, "idle", 1)
1094                                 require.Empty(t, first.Queue(t).Pending)
1095                         }
1096
1097                         first.Close(t)
1098
1099                         second := newDaemonMessageTest(t, dir, server)
1100                         require.Equal(t, source+" -> Garden", second.messages.routes.List())
1101                         require.Equal(t, 1, second.agents[1].memory.Len())
1102                         require.Equal(t, !tc.finish, second.loops[1].waiting())
1103                         require.NoError(t, second.messages.routes.Set(source, ""))
1104
1105                         restored := whatsappRouteTestInbox(t, dir)
1106                         require.Len(t, restored.state.Closed(), 1)
1107                         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
1108                         require.Empty(t, restored.state.Closed())
1109                         require.Zero(t, second.agents[0].memory.Len())
1110                         require.Equal(t, 1, second.agents[1].memory.Len())
1111
1112                         if !tc.finish {
1113                                 second.Start(t, 1)
1114                                 second.events[1].WaitFor(t, "idle", 1)
1115                         }
1116
1117                         require.Equal(t, 1, server.Calls())
1118                         require.Empty(t, second.Queue(t).Pending)
1119                         second.Close(t)
1120
1121                         third := newDaemonMessageTest(t, dir, server)
1122                         require.False(t, third.loops[1].waiting())
1123                         require.Empty(t, third.events[1].Kinds())
1124                         require.Equal(t, 1, third.agents[1].memory.Len())
1125                 })
1126         }
1127 }
1128
1129 func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) {
1130         testCases := []struct {
1131                 name string
1132                 id   *int
1133         }{
1134                 {
1135                         name: "recovers one message before its log ID is stored",
1136                         id:   nil,
1137                 },
1138                 {
1139                         name: "recovers one message after its log ID is stored",
1140                         id:   new(0),
1141                 },
1142         }
1143
1144         for _, tc := range testCases {
1145                 t.Run(tc.name, func(t *testing.T) {
1146                         dir := t.TempDir()
1147                         server := newModelFake(t)
1148                         first := newDaemonMessageTest(t, dir, server)
1149                         path := first.messages.path
1150                         first.Close(t)
1151
1152                         file := DaemonMessageFile{
1153                                 Next:            1,
1154                                 WhatsAppBatches: []int{1},
1155                                 Pending: []DaemonMessage{{
1156                                         ID:            1,
1157                                         Source:        "OneChat",
1158                                         Target:        "Garden",
1159                                         Kind:          "work",
1160                                         Text:          "[WhatsApp] Keep the gate plan.",
1161                                         LogID:         tc.id,
1162                                         WhatsAppBatch: 1,
1163                                 }},
1164                         }
1165
1166                         data, err := json.Marshal(file)
1167                         require.NoError(t, err)
1168                         require.NoError(t, os.WriteFile(path, data, 0o600))
1169
1170                         second := newDaemonMessageTest(t, dir, server)
1171                         require.Equal(t, 1, second.agents[1].memory.Len())
1172                         require.True(t, second.loops[1].waiting())
1173                         require.Equal(t, "[WhatsApp] Keep the gate plan.", second.events[1].Last("work"))
1174                         require.Equal(t, whatsappSource, second.events[1].LastEvent("work").Source)
1175                         second.Close(t)
1176
1177                         third := newDaemonMessageTest(t, dir, server)
1178                         require.Equal(t, 1, third.agents[1].memory.Len())
1179                         require.True(t, third.loops[1].waiting())
1180                         require.Zero(t, third.agents[0].memory.Len())
1181                 })
1182         }
1183 }
1184
1185 func TestQueuedWhatsAppBatchUsesRootIfItsChildIsRemoved(t *testing.T) {
1186         dir := t.TempDir()
1187         server := newModelFake(t)
1188         first := newDaemonMessageTest(t, dir, server)
1189         inbox := whatsappRouteTestInbox(t, dir)
1190         chat := WhatsAppChat{
1191                 ID:   whatsappTestAna.String(),
1192                 Name: "Ana",
1193         }
1194
1195         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
1196         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
1197         require.NoError(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true))
1198         first.Close(t)
1199         require.NoError(t, os.RemoveAll(filepath.Join(dir, daemonAgentDirName, "children", "Garden")))
1200
1201         second := newDaemonMessageTest(t, dir, server)
1202         require.Equal(t, 1, second.agents[0].memory.Len())
1203         require.True(t, second.loops[0].waiting())
1204         require.Equal(t, whatsappSource, second.events[0].LastEvent("work").Source)
1205         require.Empty(t, second.events[0].LastEvent("work").Status)
1206         require.Contains(t, second.events[0].Last("work"), "Keep the gate plan.")
1207         require.False(t, second.loops[1].waiting())
1208         second.Close(t)
1209
1210         third := newDaemonMessageTest(t, dir, server)
1211         require.Equal(t, 1, third.agents[0].memory.Len())
1212         require.True(t, third.loops[0].waiting())
1213 }
1214
1215 func TestRoutedWhatsAppBatchesCanCloseOutOfNumberOrder(t *testing.T) {
1216         dir := t.TempDir()
1217         test := newDaemonMessageTest(t, dir, newModelFake(t))
1218         inbox := whatsappRouteTestInbox(t, dir)
1219         first := WhatsAppChat{
1220                 ID:   whatsappTestAna.String(),
1221                 Name: "Ana",
1222         }
1223
1224         second := WhatsAppChat{
1225                 ID:    whatsappTestFamily.String(),
1226                 Name:  "Garden plans",
1227                 Group: true,
1228         }
1229
1230         require.NoError(t, test.messages.routes.Set("whatsapp:"+first.ID, "Garden"))
1231         require.NoError(t, test.messages.routes.Set("whatsapp:"+second.ID, "Garden"))
1232
1233         now := time.Now()
1234         _, err := inbox.state.Add(first, WhatsAppMessage{
1235                 ID:     "first",
1236                 Text:   "The first chat is still open.",
1237                 Sender: "Ana",
1238                 Time:   now,
1239         }, now)
1240         require.NoError(t, err)
1241
1242         _, err = inbox.state.Add(second, WhatsAppMessage{
1243                 ID:     "second",
1244                 Text:   "The second chat is quiet.",
1245                 Sender: "Bruno",
1246                 Time:   now,
1247         }, now.Add(-time.Minute))
1248         require.NoError(t, err)
1249         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
1250         require.Equal(t, 1, test.agents[1].memory.Len())
1251         require.Contains(t, test.events[1].Last("work"), "The second chat is quiet.")
1252         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1253         require.Equal(t, 2, test.agents[1].memory.Len())
1254         require.Contains(t, test.events[1].Last("work"), "The first chat is still open.")
1255         require.Zero(t, test.agents[0].memory.Len())
1256         require.Empty(t, test.Queue(t).WhatsAppBatches)
1257 }
1258
1259 func TestWhatsAppCatchupSeparatesChildAndRootBatches(t *testing.T) {
1260         dir := t.TempDir()
1261         test := newDaemonMessageTest(t, dir, newModelFake(t))
1262         inbox := whatsappRouteTestInbox(t, dir)
1263         garden := WhatsAppChat{
1264                 ID:    whatsappTestFamily.String(),
1265                 Name:  "Garden plans",
1266                 Group: true,
1267         }
1268
1269         library := WhatsAppChat{
1270                 ID:   whatsappTestBruno.String(),
1271                 Name: "Bruno",
1272         }
1273
1274         root := WhatsAppChat{
1275                 ID:   whatsappTestAna.String(),
1276                 Name: "Ana",
1277         }
1278
1279         require.NoError(t, test.messages.routes.Set("whatsapp:"+garden.ID, "Garden"))
1280         require.NoError(t, test.messages.routes.Set("whatsapp:"+library.ID, "Library"))
1281         inbox.waitForCatchup()
1282         whatsappRouteTestAdd(t, inbox, garden, "Check the gate.")
1283         whatsappRouteTestAdd(t, inbox, library, "Check the book.")
1284         whatsappRouteTestAdd(t, inbox, root, "Check the mixed chat.")
1285         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
1286
1287         for _, agent := range test.agents {
1288                 require.Zero(t, agent.memory.Len())
1289         }
1290
1291         inbox.endCatchup()
1292         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
1293
1294         for _, loop := range test.loops {
1295                 require.Equal(t, 1, loop.memory.Len())
1296                 require.True(t, loop.waiting())
1297         }
1298
1299         require.Contains(t, test.events[0].Last("work"), "Check the mixed chat.")
1300         require.Contains(t, test.events[1].Last("work"), "Check the gate.")
1301         require.Contains(t, test.events[2].Last("work"), "Check the book.")
1302         require.Empty(t, inbox.state.Closed())
1303         require.Empty(t, test.Queue(t).WhatsAppBatches)
1304 }
1305
1306 func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) {
1307         dir := t.TempDir()
1308         server := newModelFake(t)
1309         first := newDaemonMessageTest(t, dir, server)
1310         inbox := whatsappRouteTestInbox(t, dir)
1311         chat := WhatsAppChat{
1312                 ID:   whatsappTestAna.String(),
1313                 Name: "Ana",
1314         }
1315
1316         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
1317         require.NoError(t, os.Mkdir(first.messages.path, 0o700))
1318         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
1319         require.ErrorContains(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true), "save agent messages")
1320         require.Len(t, inbox.state.Closed(), 1)
1321
1322         for _, loop := range first.loops {
1323                 require.ErrorContains(t, loop.failed(), "save agent messages")
1324                 require.Zero(t, loop.memory.Len())
1325         }
1326
1327         path := first.messages.path
1328         first.Close(t)
1329         require.NoError(t, os.Remove(path))
1330
1331         second := newDaemonMessageTest(t, dir, server)
1332         restored := whatsappRouteTestInbox(t, dir)
1333         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
1334         require.Equal(t, 1, second.agents[1].memory.Len())
1335         require.True(t, second.loops[1].waiting())
1336         require.Zero(t, second.agents[0].memory.Len())
1337         require.Empty(t, restored.state.Closed())
1338 }
1339
1340 func whatsappRouteTestInbox(t *testing.T, dir string) *WhatsAppInbox {
1341         t.Helper()
1342
1343         state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName))
1344         require.NoError(t, err)
1345
1346         return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()), &ImageStore{dir: filepath.Join(dir, imageDirName)}, discardLog())
1347 }
1348
1349 func whatsappRouteTestAdd(t *testing.T, inbox *WhatsAppInbox, chat WhatsAppChat, text string) {
1350         t.Helper()
1351
1352         at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
1353         _, err := inbox.state.Add(chat, WhatsAppMessage{
1354                 ID:     text,
1355                 Sender: "Ana",
1356                 Text:   text,
1357                 Time:   at,
1358         }, at)
1359         require.NoError(t, err)
1360 }