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                 child  bool
 308         }{
 309                 {
 310                         name: "keeps its reservation before a startup user message",
 311                 },
 312                 {
 313                         name:   "keeps its reservation before a worker recovery note",
 314                         resume: true,
 315                 },
 316                 {
 317                         name:  "keeps a child reservation before startup input",
 318                         child: true,
 319                 },
 320                 {
 321                         name:   "keeps a child reservation before a worker recovery note",
 322                         child:  true,
 323                         resume: true,
 324                 },
 325         }
 326
 327         for _, tc := range testCases {
 328                 t.Run(tc.name, func(t *testing.T) {
 329                         index := 0
 330                         if tc.child {
 331                                 index = 1
 332                         }
 333
 334                         directory := t.TempDir()
 335                         server := newModelFake(t)
 336                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 337                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 338                         }
 339
 340                         first := newDaemonMessageTest(t, directory, server)
 341                         input := &NotifyInput{
 342                                 AgentID: first.agents[index].agent.config.ID,
 343                                 Source:  "check",
 344                                 ID:      "reserved",
 345                                 Text:    "A result with a prepared log ID.",
 346                         }
 347
 348                         receipt, err := first.messages.Notify(input)
 349                         require.NoError(t, err)
 350
 351                         id := 0
 352                         first.messages.state.Pending[0].LogID = &id
 353                         require.NoError(t, first.messages.save())
 354
 355                         expected := first.Queue(t).Pending[0].Text
 356                         first.Close(t)
 357
 358                         second := newDaemonMessageTest(t, directory, server)
 359                         require.False(t, second.loops[index].waiting())
 360                         require.Empty(t, second.events[index].Kinds())
 361
 362                         if tc.resume {
 363                                 path := filepath.Join(second.agents[index].agent.dir, subagentWorkFileName)
 364                                 require.NoError(t, os.WriteFile(path, []byte(`{"running":["BlueFox"]}`), 0o600))
 365                                 require.NoError(t, second.loops[index].runs.Resume(t.Context()))
 366
 367                                 note, err := second.agents[index].memory.Message(1)
 368                                 require.NoError(t, err)
 369                                 require.Equal(t, "note", note.Kind)
 370                                 require.Contains(t, note.Text, "The daemon restarted, cutting BlueFox")
 371                         } else {
 372                                 submitted, err := second.loops[index].Submit(terminalSource, "A user message during startup.")
 373                                 require.NoError(t, err)
 374                                 require.Equal(t, 1, submitted.ID)
 375                         }
 376
 377                         restored, err := second.agents[index].memory.Message(id)
 378                         require.NoError(t, err)
 379                         require.Equal(t, "work", restored.Kind)
 380                         require.Equal(t, expected, restored.Text)
 381                         require.Len(t, second.Queue(t).Pending, 1)
 382
 383                         second.Start(t, index)
 384                         second.events[index].WaitFor(t, "idle", 1)
 385                         require.Empty(t, second.Queue(t).Pending)
 386                         require.Equal(t, 1, server.Calls())
 387                         require.Equal(t, id, second.events[index].LastEvent("work").ID)
 388
 389                         duplicate, err := second.messages.Notify(input)
 390                         require.NoError(t, err)
 391                         require.Equal(t, "duplicate", duplicate.Status)
 392                         require.Equal(t, receipt.ID, duplicate.ID)
 393                         require.Len(t, second.Queue(t).Notifications, 1)
 394                 })
 395         }
 396 }
 397
 398 func TestNotificationBatchMakesProgressWithEscapedControlCharacters(t *testing.T) {
 399         server := newModelFake(t)
 400         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 401                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 402         }
 403
 404         test := newDaemonMessageTest(t, t.TempDir(), server)
 405         _, err := test.messages.Notify(&NotifyInput{
 406                 Source: "check",
 407                 ID:     "escaped",
 408                 Text:   "data" + strings.Repeat("\x00", notifyTextBytes-4),
 409         })
 410         require.NoError(t, err)
 411         require.Greater(t, len(test.Queue(t).Pending[0].Text), 64<<10)
 412         test.Start(t, 0)
 413         test.events[0].WaitFor(t, "idle", 1)
 414         require.Empty(t, test.Queue(t).Pending)
 415 }
 416
 417 func TestNotificationSelectorsBindOnePermanentIdentity(t *testing.T) {
 418         server := newModelFake(t)
 419         test := newDaemonMessageTest(t, t.TempDir(), server)
 420         input := &NotifyInput{
 421                 AgentID: test.agents[1].agent.config.ID,
 422                 Source:  "check",
 423                 ID:      "targeted",
 424                 Text:    "A result for one child.",
 425         }
 426
 427         accepted, err := test.messages.Notify(input)
 428         require.NoError(t, err)
 429
 430         queue := test.Queue(t)
 431         require.Equal(t, test.agents[1].agent.config.ID, queue.Pending[0].TargetID)
 432         require.Equal(t, queue.Pending[0].TargetID, queue.Notifications[0].TargetID)
 433         require.Equal(t, "Garden", queue.Pending[0].Target)
 434         require.Zero(t, test.agents[0].memory.Len())
 435         require.Zero(t, test.agents[1].memory.Len())
 436         require.Zero(t, test.agents[2].memory.Len())
 437
 438         input.AgentID = strings.ToUpper(test.agents[1].agent.config.ID)
 439         duplicate, err := test.messages.Notify(input)
 440         require.NoError(t, err)
 441         require.Equal(t, accepted.ID, duplicate.ID)
 442         require.Equal(t, "duplicate", duplicate.Status)
 443
 444         input.AgentID = test.agents[2].agent.config.ID
 445         _, err = test.messages.Notify(input)
 446         require.ErrorContains(t, err, "conflicts")
 447
 448         input.AgentID = ""
 449         _, err = test.messages.Notify(input)
 450         require.ErrorContains(t, err, "conflicts")
 451         require.Equal(t, queue, test.Queue(t))
 452 }
 453
 454 func TestNotificationRejectsUnknownWorkerAndMalformedIdentities(t *testing.T) {
 455         server := newModelFake(t)
 456         test := newDaemonMessageTest(t, t.TempDir(), server)
 457         test.loops[0].runs.subagents.draw = func() string {
 458                 return "BlueFox"
 459         }
 460
 461         worker, err := test.loops[0].runs.subagents.Create(t.Context())
 462         require.NoError(t, err)
 463
 464         testCases := []NotifyInput{
 465                 {AgentID: worker.name},
 466                 {AgentID: "Missing"},
 467                 {AgentID: "garden"},
 468                 {AgentID: "../Garden"},
 469                 {AgentID: strings.Repeat("f", 2*agentIDBytes)},
 470                 {AgentID: "invalid"},
 471         }
 472
 473         for _, tc := range testCases {
 474                 tc.Source = "check"
 475                 tc.ID = "1"
 476                 tc.Text = "Unchanged data."
 477                 receipt, err := test.messages.Notify(&tc)
 478                 require.Error(t, err)
 479                 require.Empty(t, receipt.Kind)
 480         }
 481
 482         require.Empty(t, test.messages.state.Pending)
 483         require.Empty(t, test.messages.state.Notifications)
 484         require.Zero(t, server.Calls())
 485 }
 486
 487 func TestChildNotificationWaitsForTheWholeActiveChain(t *testing.T) {
 488         server := newModelFake(t)
 489         arrived := make(chan struct{})
 490         release := make(chan struct{})
 491         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 492                 if call.number == 0 {
 493                         close(arrived)
 494                         <-release
 495                         modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`)}, 100, 0, 5))
 496
 497                         return
 498                 }
 499
 500                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 501         }
 502
 503         test := newDaemonMessageTest(t, t.TempDir(), server)
 504         test.Start(t, 0, 1, 2)
 505         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Start the first chain."))
 506         <-arrived
 507
 508         _, err := test.messages.Notify(&NotifyInput{
 509                 AgentID: test.agents[1].agent.config.ID,
 510                 Source:  "check",
 511                 ID:      "next-turn",
 512                 Text:    "Only the next child turn sees this.",
 513         })
 514         require.NoError(t, err)
 515         require.Equal(t, 1, test.agents[1].memory.Len())
 516         close(release)
 517         test.events[1].WaitFor(t, "idle", 2)
 518         require.Equal(t, 3, server.Calls())
 519         require.NotContains(t, strings.Join(turnInputTexts(t, server.Call(1)), "\n"), "Only the next child turn")
 520         require.Contains(t, strings.Join(turnInputTexts(t, server.Call(2)), "\n"), "Only the next child turn")
 521         require.Equal(t, notifySource, test.events[1].LastEvent("work").Source)
 522         require.Zero(t, test.agents[0].memory.Len())
 523         require.Zero(t, test.agents[2].memory.Len())
 524         require.Empty(t, test.Queue(t).Pending)
 525 }
 526
 527 func TestNotificationStableIdentitySurvivesOfflineNameChange(t *testing.T) {
 528         directory := t.TempDir()
 529         server := newModelFake(t)
 530         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 531                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 532         }
 533
 534         first := newDaemonMessageTest(t, directory, server)
 535         input := &NotifyInput{
 536                 AgentID: first.agents[1].agent.config.ID,
 537                 Source:  "check",
 538                 ID:      "rename-proof",
 539                 Text:    "A saved child result.",
 540         }
 541
 542         receipt, err := first.messages.Notify(input)
 543         require.NoError(t, err)
 544
 545         id := 0
 546         first.messages.state.Pending[0].LogID = &id
 547         require.NoError(t, first.messages.save())
 548
 549         envelope := first.Queue(t).Pending[0].Text
 550         config := first.agents[1].agent.config
 551         oldPath := first.agents[1].agent.dir
 552         first.Close(t)
 553
 554         // Change only test fixtures to exercise recovery under the same
 555         // identity.
 556         newPath := filepath.Join(filepath.Dir(oldPath), "Orchard")
 557         require.NoError(t, os.Rename(oldPath, newPath))
 558
 559         config.Name = "Orchard"
 560         agentWriteTestConfig(t, newPath, config)
 561
 562         second := newDaemonMessageTest(t, directory, server)
 563         child := second.messages.Child("OneChat", "Orchard")
 564         require.NotNil(t, child)
 565         require.False(t, child.waiting())
 566
 567         restored, err := child.memory.Message(id)
 568         require.NoError(t, err)
 569         require.Equal(t, envelope, restored.Text)
 570
 571         duplicate, err := second.messages.Notify(input)
 572         require.NoError(t, err)
 573         require.Equal(t, receipt.ID, duplicate.ID)
 574         require.Equal(t, "duplicate", duplicate.Status)
 575         require.NoError(t, second.messages.Notifications(child))
 576         require.NoError(t, child.turn(t.Context()))
 577         require.Empty(t, second.Queue(t).Pending)
 578         require.Zero(t, second.agents[0].memory.Len())
 579
 580         second.Close(t)
 581
 582         third := newDaemonMessageTest(t, directory, server)
 583         duplicate, err = third.messages.Notify(input)
 584         require.NoError(t, err)
 585         require.Equal(t, "duplicate", duplicate.Status)
 586         require.Empty(t, third.Queue(t).Pending)
 587         require.Equal(t, 1, server.Calls())
 588 }
 589
 590 func TestMissingNotificationIdentityStopsStartupWithoutFallback(t *testing.T) {
 591         directory := t.TempDir()
 592         server := newModelFake(t)
 593         first := newDaemonMessageTest(t, directory, server)
 594         _, err := first.messages.Notify(&NotifyInput{
 595                 AgentID: first.agents[1].agent.config.ID,
 596                 Source:  "check",
 597                 ID:      "missing",
 598                 Text:    "Do not redirect this result.",
 599         })
 600         require.NoError(t, err)
 601
 602         path := first.messages.path
 603         before, err := os.ReadFile(path)
 604         require.NoError(t, err)
 605
 606         childPath := first.agents[1].agent.dir
 607         first.Close(t)
 608
 609         require.NoError(t, os.RemoveAll(childPath))
 610         // A new agent with the same name must not receive the old identity's
 611         // work.
 612         daemonChildConfig(t, directory, "Garden")
 613
 614         agents, err := openDaemonAgents(t.Context(), directory, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
 615         require.NoError(t, err)
 616
 617         var loops []*TurnLoop
 618
 619         for _, agent := range agents {
 620                 loop, err := agent.Loop(nil, func(Event) {
 621                 })
 622                 require.NoError(t, err)
 623
 624                 loops = append(loops, loop)
 625         }
 626
 627         _, err = openDaemonMessages(loops)
 628         require.ErrorContains(t, err, "notification target is missing")
 629
 630         after, err := os.ReadFile(path)
 631         require.NoError(t, err)
 632         require.Equal(t, before, after)
 633         require.Zero(t, agents[0].memory.Len())
 634         require.Zero(t, agents[1].memory.Len())
 635         require.NoError(t, agents.Close())
 636 }
 637
 638 func TestCompletedNotificationReceiptDoesNotRequireItsTargetAtStartup(t *testing.T) {
 639         directory := t.TempDir()
 640         server := newModelFake(t)
 641         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 642                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 643         }
 644
 645         first := newDaemonMessageTest(t, directory, server)
 646         _, err := first.messages.Notify(&NotifyInput{
 647                 AgentID: first.agents[1].agent.config.ID,
 648                 Source:  "check",
 649                 ID:      "completed-child",
 650                 Text:    "A completed result.",
 651         })
 652         require.NoError(t, err)
 653         require.NoError(t, first.messages.Notifications(first.loops[1]))
 654         require.NoError(t, first.loops[1].turn(t.Context()))
 655         require.Empty(t, first.Queue(t).Pending)
 656
 657         childPath := first.agents[1].agent.dir
 658         first.Close(t)
 659         require.NoError(t, os.RemoveAll(childPath))
 660
 661         second := newDaemonMessageTest(t, directory, server)
 662         require.Empty(t, second.Queue(t).Pending)
 663         require.Len(t, second.Queue(t).Notifications, 1)
 664         require.Zero(t, second.agents[0].memory.Len())
 665 }
 666
 667 func TestLegacyNotificationReceiptsStillBindTheRoot(t *testing.T) {
 668         directory := t.TempDir()
 669         server := newModelFake(t)
 670         first := newDaemonMessageTest(t, directory, server)
 671         input := &NotifyInput{
 672                 Source: "check",
 673                 ID:     "legacy",
 674                 Text:   "A root result.",
 675         }
 676
 677         receipt, err := first.messages.Notify(input)
 678         require.NoError(t, err)
 679
 680         first.messages.state.Pending[0].TargetID = ""
 681         first.messages.state.Notifications[0].TargetID = ""
 682         require.NoError(t, first.messages.save())
 683
 684         before, err := os.ReadFile(first.messages.path)
 685         require.NoError(t, err)
 686         first.Close(t)
 687
 688         second := newDaemonMessageTest(t, directory, server)
 689         after, err := os.ReadFile(second.messages.path)
 690         require.NoError(t, err)
 691         require.Equal(t, before, after)
 692
 693         duplicate, err := second.messages.Notify(input)
 694         require.NoError(t, err)
 695         require.Equal(t, receipt.ID, duplicate.ID)
 696         require.Equal(t, "duplicate", duplicate.Status)
 697
 698         input.AgentID = second.agents[0].agent.config.ID
 699         duplicate, err = second.messages.Notify(input)
 700         require.NoError(t, err)
 701         require.Equal(t, "duplicate", duplicate.Status)
 702
 703         input.AgentID = second.agents[1].agent.config.ID
 704         _, err = second.messages.Notify(input)
 705         require.ErrorContains(t, err, "conflicts")
 706 }
 707
 708 func TestChildNotificationRunsWhileTheRootIsBusy(t *testing.T) {
 709         server := newModelFake(t)
 710         arrived := make(chan struct{})
 711         release := make(chan struct{})
 712         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 713                 if call.number == 0 {
 714                         close(arrived)
 715                         <-release
 716                 }
 717
 718                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 719         }
 720
 721         test := newDaemonMessageTest(t, t.TempDir(), server)
 722         test.Start(t, 0, 1)
 723
 724         _, err := test.loops[0].Submit(terminalSource, "Keep the root busy.")
 725         require.NoError(t, err)
 726         <-arrived
 727
 728         _, err = test.messages.Notify(&NotifyInput{
 729                 AgentID: test.agents[1].agent.config.ID,
 730                 Source:  "check",
 731                 ID:      "independent",
 732                 Text:    "A result for the idle child.",
 733         })
 734         require.NoError(t, err)
 735         test.events[1].WaitFor(t, "idle", 1)
 736         require.True(t, test.loops[0].Busy())
 737         require.Equal(t, 1, test.agents[0].memory.Len())
 738         require.Empty(t, test.events[0].Last("work"))
 739         close(release)
 740         test.events[0].WaitFor(t, "idle", 1)
 741         require.Empty(t, test.Queue(t).Pending)
 742         require.Equal(t, 2, server.Calls())
 743 }
 744
 745 func TestNotificationBatchesStayOrderedForEachChild(t *testing.T) {
 746         server := newModelFake(t)
 747         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 748                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 749         }
 750
 751         test := newDaemonMessageTest(t, t.TempDir(), server)
 752         test.agents[1].memory.summarizer = newFakeSummarizer()
 753         test.agents[2].memory.summarizer = newFakeSummarizer()
 754
 755         for number := range 21 {
 756                 for index := 1; index <= 2; index++ {
 757                         _, err := test.messages.Notify(&NotifyInput{
 758                                 AgentID: test.agents[index].agent.config.ID,
 759                                 Source:  "check",
 760                                 ID:      fmt.Sprintf("%d-%d", index, number),
 761                                 Text:    fmt.Sprintf("target-%d-result-%02d", index, number),
 762                         })
 763                         require.NoError(t, err)
 764                 }
 765         }
 766
 767         test.Start(t, 0, 1, 2)
 768         test.events[1].WaitFor(t, "idle", 2)
 769         test.events[2].WaitFor(t, "idle", 2)
 770         require.Equal(t, 4, server.Calls())
 771         require.Zero(t, test.agents[0].memory.Len())
 772         require.Empty(t, test.Queue(t).Pending)
 773
 774         for index := 1; index <= 2; index++ {
 775                 for number := range 21 {
 776                         logged, err := test.agents[index].memory.Message(number)
 777                         require.NoError(t, err)
 778                         require.Contains(t, logged.Text, fmt.Sprintf("target-%d-result-%02d", index, number))
 779                 }
 780         }
 781 }
 782
 783 func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
 784         server := newModelFake(t)
 785         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 786                 text := "I will ask the user."
 787                 if call.body["model"] == "gpt-test-garden" {
 788                         text = "Approve the gate size."
 789                 }
 790
 791                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 792         }
 793
 794         test := newDaemonMessageTest(t, t.TempDir(), server)
 795         test.Start(t, 0, 1, 2)
 796         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
 797         test.events[0].WaitFor(t, "idle", 1)
 798         test.events[1].WaitFor(t, "idle", 1)
 799
 800         require.Equal(t, "OneChat", test.events[1].LastEvent("user").Source)
 801         require.Equal(t, "Check the gate size.", test.events[1].Last("user"))
 802         require.Equal(t, "Garden", test.events[0].LastEvent("work").Source)
 803         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
 804         require.Equal(t, []string{"user", "busy", "reply", "idle"}, test.events[1].Kinds())
 805         require.Equal(t, []string{"work", "busy", "reply", "idle"}, test.events[0].Kinds())
 806         require.Equal(t, 2, server.Calls())
 807         require.Zero(t, test.agents[2].memory.Len())
 808         require.Empty(t, test.events[2].Kinds())
 809
 810         first, err := test.agents[1].memory.Message(0)
 811         require.NoError(t, err)
 812         require.Equal(t, "user", first.Kind)
 813
 814         report, err := test.agents[0].memory.Message(0)
 815         require.NoError(t, err)
 816         require.Equal(t, "work", report.Kind)
 817         require.Equal(t, "[Garden] Approve the gate size.", report.Text)
 818         require.Contains(t, test.logs.String(), "turn started name=Garden")
 819         require.Contains(t, test.logs.String(), "turn done name=Garden")
 820         require.NotContains(t, test.logs.String(), "gate size")
 821         require.Empty(t, test.Queue(t).Pending)
 822
 823         info, err := os.Stat(test.messages.path)
 824         require.NoError(t, err)
 825         require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
 826 }
 827
 828 func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
 829         server := newModelFake(t)
 830         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 831                 text := "I will ask the user."
 832                 if call.body["model"] == "gpt-test-garden" {
 833                         text = "Approve the gate size."
 834                 }
 835
 836                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 837         }
 838
 839         test := newDaemonMessageTest(t, t.TempDir(), server)
 840         telegram := newTelegramFake(t)
 841         bot := newTelegramBot(newTelegramAPI(telegram.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), discardLog())
 842         test.loops[0].notify = func(event Event) {
 843                 test.events[0].Add(event)
 844                 bot.Notify(event)
 845         }
 846
 847         ctx, cancel := context.WithCancel(t.Context())
 848         defer cancel()
 849
 850         done := make(chan struct{})
 851         go func() {
 852                 defer close(done)
 853
 854                 bot.send(ctx)
 855         }()
 856
 857         t.Cleanup(func() {
 858                 cancel()
 859                 <-done
 860         })
 861
 862         test.Start(t, 0, 1)
 863         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
 864         test.events[0].WaitFor(t, "idle", 1)
 865         require.Eventually(t, func() bool {
 866                 return len(telegram.Sent()) >= 2
 867         }, 10*time.Second, time.Millisecond)
 868
 869         require.Equal(t, []string{"[Garden] finished", "I will ask the user."}, telegram.Sent())
 870         require.Equal(t, subagentFinished, test.events[0].LastEvent("work").Status)
 871         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
 872 }
 873
 874 func TestQueuedAgentMessageFailureKeepsItsOperationAndCause(t *testing.T) {
 875         server := newModelFake(t)
 876         test := newDaemonMessageTest(t, t.TempDir(), server)
 877         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 878
 879         failure := errors.New("fake queued message failure")
 880         test.loops[1].fail(failure)
 881
 882         err := test.loops[1].Run(t.Context())
 883         require.ErrorIs(t, err, failure)
 884         require.ErrorContains(t, err, "check turn loop failure")
 885         require.Zero(t, server.Calls())
 886 }
 887
 888 func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.T) {
 889         testCases := []struct {
 890                 name string
 891                 text string
 892         }{
 893                 {
 894                         name: "sends nothing for an empty final reply",
 895                         text: "",
 896                 },
 897                 {
 898                         name: "sends nothing for a whitespace final reply",
 899                         text: " \n\t ",
 900                 },
 901         }
 902
 903         for _, tc := range testCases {
 904                 t.Run(tc.name, func(t *testing.T) {
 905                         server := newModelFake(t)
 906                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 907                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(tc.text)}, 100, 0, 5))
 908                         }
 909
 910                         test := newDaemonMessageTest(t, t.TempDir(), server)
 911                         test.Start(t, 0, 1, 2)
 912                         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 913                         test.events[1].WaitFor(t, "idle", 1)
 914                         require.Equal(t, 1, server.Calls())
 915                         require.Zero(t, test.agents[0].memory.Len())
 916                         require.Empty(t, test.events[0].Kinds())
 917                         require.False(t, test.loops[0].waiting())
 918                         require.Empty(t, test.events[2].Kinds())
 919                         require.Empty(t, test.Queue(t).Pending)
 920                 })
 921         }
 922 }
 923
 924 func TestBusyChildQueuesTheNextParentMessage(t *testing.T) {
 925         server := newModelFake(t)
 926         arrived := make(chan struct{})
 927         release := make(chan struct{})
 928         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 929                 if call.number == 0 {
 930                         close(arrived)
 931                         <-release
 932                 }
 933
 934                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 935         }
 936
 937         test := newDaemonMessageTest(t, t.TempDir(), server)
 938         test.Start(t, 0, 1)
 939         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the first plan."))
 940         <-arrived
 941         require.True(t, test.loops[1].Busy())
 942         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the second plan too."))
 943         require.Equal(t, 1, server.Calls())
 944         require.Len(t, test.Queue(t).Pending, 2)
 945         close(release)
 946         test.events[1].WaitFor(t, "idle", 2)
 947         require.Equal(t, 2, server.Calls())
 948
 949         texts := turnInputTexts(t, server.Call(1))
 950         require.Equal(t, "Check the second plan too.", texts[len(texts)-1])
 951         require.Empty(t, test.Queue(t).Pending)
 952 }
 953
 954 func TestBusyParentQueuesItsChildReport(t *testing.T) {
 955         server := newModelFake(t)
 956         arrived := make(chan struct{})
 957         release := make(chan struct{})
 958         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 959                 text := "The plan is ready."
 960                 if call.number == 0 {
 961                         close(arrived)
 962                         <-release
 963
 964                         text = "I will continue."
 965                 }
 966
 967                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 968         }
 969
 970         test := newDaemonMessageTest(t, t.TempDir(), server)
 971         test.Start(t, 0, 1)
 972
 973         _, err := test.loops[0].Submit("terminal", "Start the work.")
 974         require.NoError(t, err)
 975         <-arrived
 976         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Prepare the plan."))
 977         test.events[1].WaitFor(t, "idle", 1)
 978         require.Equal(t, "[Garden] The plan is ready.", test.events[0].Last("work"))
 979         require.True(t, test.loops[0].Busy())
 980         require.Equal(t, 2, server.Calls())
 981         close(release)
 982         test.events[0].WaitFor(t, "idle", 2)
 983         require.Equal(t, 3, server.Calls())
 984
 985         texts := turnInputTexts(t, server.Call(2))
 986         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
 987         require.Empty(t, test.Queue(t).Pending)
 988 }
 989
 990 func TestChildReportsOnlyTheFinalReplyAndTakesQueuedToolInput(t *testing.T) {
 991         server := newModelFake(t)
 992         arrived := make(chan struct{})
 993         release := make(chan struct{})
 994         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 995                 if call.number == 0 {
 996                         close(arrived)
 997                         <-release
 998                         modelWriteEvents(w, modelCompletedEvent([]any{
 999                                 modelMessageItem("I will check the plan."),
1000                                 modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`),
1001                         }, 100, 0, 5))
1002
1003                         return
1004                 }
1005
1006                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan needs approval.")}, 100, 0, 5))
1007         }
1008
1009         test := newDaemonMessageTest(t, t.TempDir(), server)
1010         test.Start(t, 1)
1011         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the plan."))
1012         <-arrived
1013         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Also check the date."))
1014         close(release)
1015         test.events[1].WaitFor(t, "idle", 1)
1016         require.Equal(t, 2, server.Calls())
1017         require.Equal(t, 1, test.events[1].Count("busy"))
1018         require.Equal(t, 1, test.events[0].Count("work"))
1019         require.Equal(t, "[Garden] The plan needs approval.", test.events[0].Last("work"))
1020         require.NotContains(t, test.events[0].Last("work"), "I will check")
1021
1022         texts := turnInputTexts(t, server.Call(1))
1023         require.Equal(t, "Also check the date.", texts[len(texts)-1])
1024
1025         queue := test.Queue(t)
1026         require.Len(t, queue.Pending, 1)
1027         require.Equal(t, "work", queue.Pending[0].Kind)
1028 }
1029
1030 func TestQueuedParentMessageSurvivesDaemonRestart(t *testing.T) {
1031         dir := t.TempDir()
1032         server := newModelFake(t)
1033         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
1034                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1035         }
1036
1037         first := newDaemonMessageTest(t, dir, server)
1038         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Remember the garden plan."))
1039         require.Equal(t, 1, first.agents[1].memory.Len())
1040         first.Close(t)
1041
1042         second := newDaemonMessageTest(t, dir, server)
1043         require.Equal(t, 1, second.agents[1].memory.Len())
1044         require.Equal(t, "OneChat", second.events[1].LastEvent("user").Source)
1045         second.Start(t, 0, 1)
1046         second.events[1].WaitFor(t, "idle", 1)
1047         require.Equal(t, 1, server.Calls())
1048         require.Empty(t, second.Queue(t).Pending)
1049         second.Close(t)
1050
1051         third := newDaemonMessageTest(t, dir, server)
1052         require.Equal(t, 1, third.agents[1].memory.Len())
1053         require.False(t, third.loops[1].waiting())
1054         require.Empty(t, third.events[1].Kinds())
1055 }
1056
1057 func TestQueuedChildReportSurvivesDaemonRestart(t *testing.T) {
1058         dir := t.TempDir()
1059         server := newModelFake(t)
1060         first := newDaemonMessageTest(t, dir, server)
1061         first.Start(t, 1)
1062         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
1063         first.events[1].WaitFor(t, "idle", 1)
1064         require.Equal(t, "[Garden] a line", first.events[0].Last("work"))
1065         require.Len(t, first.Queue(t).Pending, 1)
1066         first.Close(t)
1067
1068         second := newDaemonMessageTest(t, dir, server)
1069         require.Equal(t, 1, second.agents[0].memory.Len())
1070         require.False(t, second.loops[1].waiting())
1071         second.Start(t, 0, 1)
1072         second.events[0].WaitFor(t, "idle", 1)
1073         require.Equal(t, 2, server.Calls())
1074         require.Equal(t, 2, second.agents[0].memory.Len())
1075         require.Empty(t, second.Queue(t).Pending)
1076         second.Close(t)
1077
1078         third := newDaemonMessageTest(t, dir, server)
1079         require.Equal(t, 2, third.agents[0].memory.Len())
1080         require.False(t, third.loops[0].waiting())
1081         require.Empty(t, third.events[0].Kinds())
1082 }
1083
1084 func TestInterruptedChildDeliveryResumesAfterDaemonRestart(t *testing.T) {
1085         dir := t.TempDir()
1086         server := newModelFake(t)
1087         arrived := make(chan struct{})
1088         release := make(chan struct{})
1089         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
1090                 if call.number == 0 {
1091                         close(arrived)
1092                         <-release
1093                 }
1094
1095                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1096         }
1097
1098         first := newDaemonMessageTest(t, dir, server)
1099         first.Start(t, 1)
1100         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Finish the garden plan."))
1101         <-arrived
1102         first.Close(t)
1103         close(release)
1104
1105         second := newDaemonMessageTest(t, dir, server)
1106         require.Equal(t, 1, second.agents[1].memory.Len())
1107         second.Start(t, 1)
1108         second.events[1].WaitFor(t, "idle", 1)
1109         require.Equal(t, 2, server.Calls())
1110         require.Equal(t, 1, second.agents[1].memory.Len())
1111         require.Empty(t, second.Queue(t).Pending)
1112 }
1113
1114 func TestInterruptedParentReportResumesAfterDaemonRestart(t *testing.T) {
1115         dir := t.TempDir()
1116         server := newModelFake(t)
1117         arrived := make(chan struct{})
1118         release := make(chan struct{})
1119         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
1120                 if call.number == 1 {
1121                         close(arrived)
1122                         <-release
1123                 }
1124
1125                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan is ready.")}, 100, 0, 5))
1126         }
1127
1128         first := newDaemonMessageTest(t, dir, server)
1129         first.Start(t, 0, 1)
1130         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
1131         <-arrived
1132         first.Close(t)
1133         close(release)
1134
1135         second := newDaemonMessageTest(t, dir, server)
1136         require.Equal(t, 1, second.agents[0].memory.Len())
1137         require.False(t, second.loops[1].waiting())
1138         second.Start(t, 0, 1)
1139         second.events[0].WaitFor(t, "idle", 1)
1140         require.Equal(t, 3, server.Calls())
1141         require.Equal(t, 2, second.agents[0].memory.Len())
1142         require.Empty(t, second.Queue(t).Pending)
1143
1144         texts := turnInputTexts(t, server.Call(2))
1145         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
1146 }
1147
1148 func TestAgentMessageWriteFailureStopsDelivery(t *testing.T) {
1149         server := newModelFake(t)
1150         test := newDaemonMessageTest(t, t.TempDir(), server)
1151         require.NoError(t, os.Mkdir(test.messages.path, 0o700))
1152         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."), "save agent messages")
1153         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Do not send again."), "save agent messages")
1154
1155         for _, loop := range test.loops {
1156                 require.ErrorContains(t, loop.failed(), "save agent messages")
1157                 require.Zero(t, loop.memory.Len())
1158         }
1159 }
1160
1161 func TestAgentMessageSnapshotBlocksDelivery(t *testing.T) {
1162         server := newModelFake(t)
1163         test := newDaemonMessageTest(t, t.TempDir(), server)
1164         entered := make(chan struct{})
1165         release := make(chan struct{})
1166
1167         var once sync.Once
1168
1169         defer once.Do(func() {
1170                 close(release)
1171         })
1172
1173         snapshot := make(chan error, 1)
1174         go func() {
1175                 snapshot <- test.messages.Snapshot(func() error {
1176                         return daemonSnapshot(t.Context(), test.loops, func() error {
1177                                 close(entered)
1178                                 <-release
1179
1180                                 return nil
1181                         })
1182                 })
1183         }()
1184
1185         <-entered
1186
1187         sent := make(chan error, 1)
1188         started := make(chan struct{})
1189         go func() {
1190                 close(started)
1191
1192                 sent <- test.messages.Tell("OneChat", "Garden", "Record the plan after the copy.")
1193         }()
1194
1195         <-started
1196         require.Never(t, func() bool {
1197                 return len(sent) > 0
1198         }, 50*time.Millisecond, time.Millisecond)
1199
1200         once.Do(func() {
1201                 close(release)
1202         })
1203         require.NoError(t, <-snapshot)
1204         require.NoError(t, <-sent)
1205         require.Equal(t, 1, test.agents[1].memory.Len())
1206         require.Len(t, test.Queue(t).Pending, 1)
1207 }
1208
1209 func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) {
1210         server := newModelFake(t)
1211
1212         test := newDaemonMessageTest(t, t.TempDir(), server)
1213         testCases := []struct {
1214                 name   string
1215                 source string
1216                 target string
1217         }{
1218                 {
1219                         name:   "no sibling messages",
1220                         source: "Garden",
1221                         target: "Library",
1222                 },
1223                 {
1224                         name:   "no arbitrary child reports",
1225                         source: "Garden",
1226                         target: "OneChat",
1227                 },
1228                 {
1229                         name:   "no self messages",
1230                         source: "OneChat",
1231                         target: "OneChat",
1232                 },
1233                 {
1234                         name:   "refuses an unknown sender",
1235                         source: "Other",
1236                         target: "Garden",
1237                 },
1238                 {
1239                         name:   "refuses an unknown recipient",
1240                         source: "OneChat",
1241                         target: "Other",
1242                 },
1243         }
1244
1245         for _, tc := range testCases {
1246                 t.Run(tc.name, func(t *testing.T) {
1247                         require.ErrorContains(t, test.messages.Tell(tc.source, tc.target, "Do not deliver."), "not a child")
1248                 })
1249         }
1250
1251         for _, agent := range test.agents {
1252                 require.Zero(t, agent.memory.Len())
1253         }
1254 }
1255
1256 func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) {
1257         server := newModelFake(t)
1258         testCases := []struct {
1259                 name string
1260                 file string
1261         }{
1262                 {
1263                         name: "stops startup for broken JSON",
1264                         file: "{",
1265                 },
1266                 {
1267                         name: "stops startup for unknown sender",
1268                         file: `{"next":1,"pending":[{"id":1,"source":"Other","target":"Garden","kind":"user"}]}`,
1269                 },
1270                 {
1271                         name: "stops startup for sibling delivery",
1272                         file: `{"next":1,"pending":[{"id":1,"source":"Library","target":"Garden","kind":"user"}]}`,
1273                 },
1274                 {
1275                         name: "stops startup for invalid sequence",
1276                         file: `{"next":0,"pending":[{"id":1,"source":"OneChat","target":"Garden","kind":"user"}]}`,
1277                 },
1278         }
1279
1280         for _, tc := range testCases {
1281                 t.Run(tc.name, func(t *testing.T) {
1282                         dir := t.TempDir()
1283                         test := newDaemonMessageTest(t, dir, server)
1284                         path := test.messages.path
1285                         test.Close(t)
1286                         require.NoError(t, os.WriteFile(path, []byte(tc.file), 0o600))
1287
1288                         agents, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
1289                         require.NoError(t, err)
1290
1291                         defer func() {
1292                                 require.NoError(t, agents.Close())
1293                         }()
1294
1295                         var loops []*TurnLoop
1296
1297                         for _, agent := range agents {
1298                                 loop, err := agent.Loop(nil, func(Event) {
1299                                 })
1300                                 require.NoError(t, err)
1301
1302                                 loops = append(loops, loop)
1303                         }
1304
1305                         _, err = openDaemonMessages(loops)
1306                         require.Error(t, err)
1307
1308                         for _, agent := range agents {
1309                                 require.Zero(t, agent.memory.Len())
1310                         }
1311                 })
1312         }
1313 }
1314
1315 func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
1316         server := newModelFake(t)
1317         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
1318                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1319         }
1320
1321         test := newDaemonMessageTest(t, t.TempDir(), server)
1322         child := test.agents[1]
1323         at := child.clock().Add(time.Minute).Format(scheduleAtLayout)
1324         output := test.loops[1].tools.run(t.Context(), "schedule_set", `{"text":"Check the plants.","at":"`+at+`","cron":null}`)
1325         require.Contains(t, output, "Schedule 1 set.")
1326         require.Equal(t, "Garden", child.schedules.sorted()[0].Owner)
1327
1328         child.schedules.clock = func() time.Time {
1329                 return child.clock().Add(time.Minute)
1330         }
1331         require.ErrorContains(t, child.schedules.deliver(t.Context(), test.loops[0]), "does not match")
1332         require.NoError(t, child.schedules.deliver(t.Context(), test.loops[1]))
1333         test.Start(t, 0, 1, 2)
1334         test.events[1].WaitFor(t, "idle", 1)
1335         require.Equal(t, 1, server.Calls())
1336         require.Equal(t, child.agent.CacheKey(), server.Call(0).body["prompt_cache_key"])
1337         require.Zero(t, test.agents[0].memory.Len())
1338         require.Zero(t, test.agents[2].memory.Len())
1339         require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.")
1340         require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source)
1341 }
1342
1343 func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) {
1344         testCases := []struct {
1345                 name string
1346                 chat WhatsAppChat
1347         }{
1348                 {
1349                         name: "routes a private chat only to its child",
1350                         chat: WhatsAppChat{
1351                                 ID:   whatsappTestAna.String(),
1352                                 Name: "Ana",
1353                         },
1354                 },
1355                 {
1356                         name: "routes an unmentioned group only to its child",
1357                         chat: WhatsAppChat{
1358                                 ID:    whatsappTestFamily.String(),
1359                                 Name:  "Garden plans",
1360                                 Group: true,
1361                         },
1362                 },
1363         }
1364
1365         for _, tc := range testCases {
1366                 t.Run(tc.name, func(t *testing.T) {
1367                         server := newModelFake(t)
1368                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
1369                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1370                         }
1371
1372                         dir := t.TempDir()
1373                         test := newDaemonMessageTest(t, dir, server)
1374                         inbox := whatsappRouteTestInbox(t, dir)
1375                         require.NoError(t, test.messages.routes.Set("whatsapp:"+tc.chat.ID, "Garden"))
1376                         whatsappRouteTestAdd(t, inbox, tc.chat, "The gate needs a new latch.")
1377                         test.Start(t, 0, 1, 2)
1378                         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1379                         test.events[1].WaitFor(t, "idle", 1)
1380                         require.Equal(t, 1, server.Calls())
1381                         require.Equal(t, "gpt-test-garden", server.Call(0).body["model"])
1382                         require.Contains(t, string(server.Call(0).raw), "The gate needs a new latch.")
1383                         require.Zero(t, test.agents[0].memory.Len())
1384                         require.Empty(t, test.events[0].Kinds())
1385                         require.Zero(t, test.agents[2].memory.Len())
1386                         require.False(t, test.loops[0].waiting())
1387                         require.Equal(t, whatsappSource, test.events[1].LastEvent("work").Source)
1388                         require.Empty(t, test.events[1].LastEvent("work").Status)
1389                         require.Equal(t, 1, test.agents[1].memory.Len())
1390                         require.Empty(t, inbox.state.Closed())
1391                         require.Empty(t, test.Queue(t).Pending)
1392                         require.Contains(t, test.logs.String(), "batch routed name=Garden kind=whatsapp id=1 to=Garden")
1393                         require.NotContains(t, test.logs.String(), tc.chat.ID)
1394                         require.NotContains(t, test.logs.String(), "new latch")
1395                 })
1396         }
1397 }
1398
1399 func TestWhatsAppUnrouteRestoresRootDelivery(t *testing.T) {
1400         dir := t.TempDir()
1401         test := newDaemonMessageTest(t, dir, newModelFake(t))
1402         inbox := whatsappRouteTestInbox(t, dir)
1403         chat := WhatsAppChat{
1404                 ID:   whatsappTestAna.String(),
1405                 Name: "Ana",
1406         }
1407
1408         source := "whatsapp:" + chat.ID
1409         require.NoError(t, test.messages.routes.Set(source, "Garden"))
1410         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
1411         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1412         require.Zero(t, test.agents[0].memory.Len())
1413         require.True(t, test.loops[1].waiting())
1414         require.NoError(t, test.messages.routes.Set(source, ""))
1415         whatsappRouteTestAdd(t, inbox, chat, "Please check the other plan.")
1416         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1417         require.Equal(t, 1, test.agents[0].memory.Len())
1418         require.Equal(t, 1, test.agents[1].memory.Len())
1419         require.True(t, test.loops[0].waiting())
1420         require.Contains(t, test.events[0].Last("work"), "Please check the other plan.")
1421         require.NotContains(t, test.events[0].Last("work"), "Keep the gate plan.")
1422         require.Empty(t, inbox.state.Closed())
1423 }
1424
1425 func TestMissingWhatsAppRouteChildFallsBackToRoot(t *testing.T) {
1426         dir := t.TempDir()
1427         test := newDaemonMessageTest(t, dir, newModelFake(t))
1428         inbox := whatsappRouteTestInbox(t, dir)
1429         chat := WhatsAppChat{
1430                 ID:   whatsappTestAna.String(),
1431                 Name: "Ana",
1432         }
1433
1434         require.NoError(t, test.messages.routes.Set("whatsapp:"+chat.ID, "Missing"))
1435         whatsappRouteTestAdd(t, inbox, chat, "Please check the plan.")
1436         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1437         require.Equal(t, 1, test.agents[0].memory.Len())
1438         require.True(t, test.loops[0].waiting())
1439         require.Zero(t, test.agents[1].memory.Len())
1440         require.False(t, test.loops[1].waiting())
1441         require.Empty(t, test.events[0].LastEvent("work").Status)
1442         require.Equal(t, whatsappSource, test.events[0].LastEvent("work").Source)
1443         require.Empty(t, inbox.state.Closed())
1444 }
1445
1446 func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) {
1447         testCases := []struct {
1448                 name   string
1449                 finish bool
1450         }{
1451                 {
1452                         name:   "queued turn resumes",
1453                         finish: false,
1454                 },
1455                 {
1456                         name:   "finished turn stays finished",
1457                         finish: true,
1458                 },
1459         }
1460
1461         for _, tc := range testCases {
1462                 t.Run(tc.name, func(t *testing.T) {
1463                         dir := t.TempDir()
1464                         server := newModelFake(t)
1465                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
1466                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1467                         }
1468
1469                         first := newDaemonMessageTest(t, dir, server)
1470                         inbox := whatsappRouteTestInbox(t, dir)
1471                         chat := WhatsAppChat{
1472                                 ID:   whatsappTestAna.String(),
1473                                 Name: "Ana",
1474                         }
1475
1476                         source := "whatsapp:" + chat.ID
1477                         require.NoError(t, first.messages.routes.Set(source, "Garden"))
1478                         whatsappRouteTestAdd(t, inbox, chat, "Remember the gate plan.")
1479                         require.NoError(t, inbox.state.CloseAll())
1480
1481                         batch := inbox.state.Closed()[0]
1482                         routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
1483                         require.NoError(t, err)
1484                         require.True(t, routed)
1485
1486                         if tc.finish {
1487                                 first.Start(t, 1)
1488                                 first.events[1].WaitFor(t, "idle", 1)
1489                                 require.Empty(t, first.Queue(t).Pending)
1490                         }
1491
1492                         first.Close(t)
1493
1494                         second := newDaemonMessageTest(t, dir, server)
1495                         require.Equal(t, source+" -> Garden", second.messages.routes.List())
1496                         require.Equal(t, 1, second.agents[1].memory.Len())
1497                         require.Equal(t, !tc.finish, second.loops[1].waiting())
1498                         require.NoError(t, second.messages.routes.Set(source, ""))
1499
1500                         restored := whatsappRouteTestInbox(t, dir)
1501                         require.Len(t, restored.state.Closed(), 1)
1502                         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
1503                         require.Empty(t, restored.state.Closed())
1504                         require.Zero(t, second.agents[0].memory.Len())
1505                         require.Equal(t, 1, second.agents[1].memory.Len())
1506
1507                         if !tc.finish {
1508                                 second.Start(t, 1)
1509                                 second.events[1].WaitFor(t, "idle", 1)
1510                         }
1511
1512                         require.Equal(t, 1, server.Calls())
1513                         require.Empty(t, second.Queue(t).Pending)
1514                         second.Close(t)
1515
1516                         third := newDaemonMessageTest(t, dir, server)
1517                         require.False(t, third.loops[1].waiting())
1518                         require.Empty(t, third.events[1].Kinds())
1519                         require.Equal(t, 1, third.agents[1].memory.Len())
1520                 })
1521         }
1522 }
1523
1524 func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) {
1525         testCases := []struct {
1526                 name string
1527                 id   *int
1528         }{
1529                 {
1530                         name: "recovers one message before its log ID is stored",
1531                         id:   nil,
1532                 },
1533                 {
1534                         name: "recovers one message after its log ID is stored",
1535                         id:   new(0),
1536                 },
1537         }
1538
1539         for _, tc := range testCases {
1540                 t.Run(tc.name, func(t *testing.T) {
1541                         dir := t.TempDir()
1542                         server := newModelFake(t)
1543                         first := newDaemonMessageTest(t, dir, server)
1544                         path := first.messages.path
1545                         first.Close(t)
1546
1547                         file := DaemonMessageFile{
1548                                 Next:            1,
1549                                 WhatsAppBatches: []int{1},
1550                                 Pending: []DaemonMessage{{
1551                                         ID:            1,
1552                                         Source:        "OneChat",
1553                                         Target:        "Garden",
1554                                         Kind:          "work",
1555                                         Text:          "[WhatsApp] Keep the gate plan.",
1556                                         LogID:         tc.id,
1557                                         WhatsAppBatch: 1,
1558                                 }},
1559                         }
1560
1561                         data, err := json.Marshal(file)
1562                         require.NoError(t, err)
1563                         require.NoError(t, os.WriteFile(path, data, 0o600))
1564
1565                         second := newDaemonMessageTest(t, dir, server)
1566                         require.Equal(t, 1, second.agents[1].memory.Len())
1567                         require.True(t, second.loops[1].waiting())
1568                         require.Equal(t, "[WhatsApp] Keep the gate plan.", second.events[1].Last("work"))
1569                         require.Equal(t, whatsappSource, second.events[1].LastEvent("work").Source)
1570                         second.Close(t)
1571
1572                         third := newDaemonMessageTest(t, dir, server)
1573                         require.Equal(t, 1, third.agents[1].memory.Len())
1574                         require.True(t, third.loops[1].waiting())
1575                         require.Zero(t, third.agents[0].memory.Len())
1576                 })
1577         }
1578 }
1579
1580 func TestQueuedWhatsAppBatchUsesRootIfItsChildIsRemoved(t *testing.T) {
1581         dir := t.TempDir()
1582         server := newModelFake(t)
1583         first := newDaemonMessageTest(t, dir, server)
1584         inbox := whatsappRouteTestInbox(t, dir)
1585         chat := WhatsAppChat{
1586                 ID:   whatsappTestAna.String(),
1587                 Name: "Ana",
1588         }
1589
1590         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
1591         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
1592         require.NoError(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true))
1593         first.Close(t)
1594         require.NoError(t, os.RemoveAll(filepath.Join(dir, daemonAgentDirName, "children", "Garden")))
1595
1596         second := newDaemonMessageTest(t, dir, server)
1597         require.Equal(t, 1, second.agents[0].memory.Len())
1598         require.True(t, second.loops[0].waiting())
1599         require.Equal(t, whatsappSource, second.events[0].LastEvent("work").Source)
1600         require.Empty(t, second.events[0].LastEvent("work").Status)
1601         require.Contains(t, second.events[0].Last("work"), "Keep the gate plan.")
1602         require.False(t, second.loops[1].waiting())
1603         second.Close(t)
1604
1605         third := newDaemonMessageTest(t, dir, server)
1606         require.Equal(t, 1, third.agents[0].memory.Len())
1607         require.True(t, third.loops[0].waiting())
1608 }
1609
1610 func TestRoutedWhatsAppBatchesCanCloseOutOfNumberOrder(t *testing.T) {
1611         dir := t.TempDir()
1612         test := newDaemonMessageTest(t, dir, newModelFake(t))
1613         inbox := whatsappRouteTestInbox(t, dir)
1614         first := WhatsAppChat{
1615                 ID:   whatsappTestAna.String(),
1616                 Name: "Ana",
1617         }
1618
1619         second := WhatsAppChat{
1620                 ID:    whatsappTestFamily.String(),
1621                 Name:  "Garden plans",
1622                 Group: true,
1623         }
1624
1625         require.NoError(t, test.messages.routes.Set("whatsapp:"+first.ID, "Garden"))
1626         require.NoError(t, test.messages.routes.Set("whatsapp:"+second.ID, "Garden"))
1627
1628         now := time.Now()
1629         _, err := inbox.state.Add(first, WhatsAppMessage{
1630                 ID:     "first",
1631                 Text:   "The first chat is still open.",
1632                 Sender: "Ana",
1633                 Time:   now,
1634         }, now)
1635         require.NoError(t, err)
1636
1637         _, err = inbox.state.Add(second, WhatsAppMessage{
1638                 ID:     "second",
1639                 Text:   "The second chat is quiet.",
1640                 Sender: "Bruno",
1641                 Time:   now,
1642         }, now.Add(-time.Minute))
1643         require.NoError(t, err)
1644         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
1645         require.Equal(t, 1, test.agents[1].memory.Len())
1646         require.Contains(t, test.events[1].Last("work"), "The second chat is quiet.")
1647         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1648         require.Equal(t, 2, test.agents[1].memory.Len())
1649         require.Contains(t, test.events[1].Last("work"), "The first chat is still open.")
1650         require.Zero(t, test.agents[0].memory.Len())
1651         require.Empty(t, test.Queue(t).WhatsAppBatches)
1652 }
1653
1654 func TestWhatsAppCatchupSeparatesChildAndRootBatches(t *testing.T) {
1655         dir := t.TempDir()
1656         test := newDaemonMessageTest(t, dir, newModelFake(t))
1657         inbox := whatsappRouteTestInbox(t, dir)
1658         garden := WhatsAppChat{
1659                 ID:    whatsappTestFamily.String(),
1660                 Name:  "Garden plans",
1661                 Group: true,
1662         }
1663
1664         library := WhatsAppChat{
1665                 ID:   whatsappTestBruno.String(),
1666                 Name: "Bruno",
1667         }
1668
1669         root := WhatsAppChat{
1670                 ID:   whatsappTestAna.String(),
1671                 Name: "Ana",
1672         }
1673
1674         require.NoError(t, test.messages.routes.Set("whatsapp:"+garden.ID, "Garden"))
1675         require.NoError(t, test.messages.routes.Set("whatsapp:"+library.ID, "Library"))
1676         inbox.waitForCatchup()
1677         whatsappRouteTestAdd(t, inbox, garden, "Check the gate.")
1678         whatsappRouteTestAdd(t, inbox, library, "Check the book.")
1679         whatsappRouteTestAdd(t, inbox, root, "Check the mixed chat.")
1680         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
1681
1682         for _, agent := range test.agents {
1683                 require.Zero(t, agent.memory.Len())
1684         }
1685
1686         inbox.endCatchup()
1687         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
1688
1689         for _, loop := range test.loops {
1690                 require.Equal(t, 1, loop.memory.Len())
1691                 require.True(t, loop.waiting())
1692         }
1693
1694         require.Contains(t, test.events[0].Last("work"), "Check the mixed chat.")
1695         require.Contains(t, test.events[1].Last("work"), "Check the gate.")
1696         require.Contains(t, test.events[2].Last("work"), "Check the book.")
1697         require.Empty(t, inbox.state.Closed())
1698         require.Empty(t, test.Queue(t).WhatsAppBatches)
1699 }
1700
1701 func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) {
1702         dir := t.TempDir()
1703         server := newModelFake(t)
1704         first := newDaemonMessageTest(t, dir, server)
1705         inbox := whatsappRouteTestInbox(t, dir)
1706         chat := WhatsAppChat{
1707                 ID:   whatsappTestAna.String(),
1708                 Name: "Ana",
1709         }
1710
1711         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
1712         require.NoError(t, os.Mkdir(first.messages.path, 0o700))
1713         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
1714         require.ErrorContains(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true), "save agent messages")
1715         require.Len(t, inbox.state.Closed(), 1)
1716
1717         for _, loop := range first.loops {
1718                 require.ErrorContains(t, loop.failed(), "save agent messages")
1719                 require.Zero(t, loop.memory.Len())
1720         }
1721
1722         path := first.messages.path
1723         first.Close(t)
1724         require.NoError(t, os.Remove(path))
1725
1726         second := newDaemonMessageTest(t, dir, server)
1727         restored := whatsappRouteTestInbox(t, dir)
1728         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
1729         require.Equal(t, 1, second.agents[1].memory.Len())
1730         require.True(t, second.loops[1].waiting())
1731         require.Zero(t, second.agents[0].memory.Len())
1732         require.Empty(t, restored.state.Closed())
1733 }
1734
1735 func whatsappRouteTestInbox(t *testing.T, dir string) *WhatsAppInbox {
1736         t.Helper()
1737
1738         state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName))
1739         require.NoError(t, err)
1740
1741         return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()), &ImageStore{dir: filepath.Join(dir, imageDirName)}, discardLog())
1742 }
1743
1744 func whatsappRouteTestAdd(t *testing.T, inbox *WhatsAppInbox, chat WhatsAppChat, text string) {
1745         t.Helper()
1746
1747         at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
1748         _, err := inbox.state.Add(chat, WhatsAppMessage{
1749                 ID:     text,
1750                 Sender: "Ana",
1751                 Text:   text,
1752                 Time:   at,
1753         }, at)
1754         require.NoError(t, err)
1755 }