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