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 TestChildNotificationReportsOnlyItsFinalReplyToTheRoot(t *testing.T) {
 784         server := newModelFake(t)
 785         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 786                 reply := ""
 787                 if call.body["model"] == "gpt-test-garden" {
 788                         reply = "A result needs the user."
 789                 }
 790
 791                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(reply)}, 100, 0, 5))
 792         }
 793
 794         test := newDaemonMessageTest(t, t.TempDir(), server)
 795         test.Start(t, 0, 1, 2)
 796
 797         _, err := test.messages.Notify(&NotifyInput{
 798                 AgentID: test.agents[1].agent.config.ID,
 799                 Source:  "check",
 800                 ID:      "report",
 801                 Text:    "Private child event data.",
 802         })
 803         require.NoError(t, err)
 804         test.events[1].WaitFor(t, "idle", 1)
 805         test.events[0].WaitFor(t, "idle", 1)
 806         require.Equal(t, 2, server.Calls())
 807         require.Equal(t, "[Garden] A result needs the user.", test.events[0].Last("work"))
 808         require.Equal(t, 1, test.events[0].Count("work"))
 809         require.Equal(t, notifySource, test.events[1].LastEvent("work").Source)
 810         require.Zero(t, test.agents[2].memory.Len())
 811         require.Empty(t, test.Queue(t).Pending)
 812 }
 813
 814 func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
 815         server := newModelFake(t)
 816         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 817                 text := "I will ask the user."
 818                 if call.body["model"] == "gpt-test-garden" {
 819                         text = "Approve the gate size."
 820                 }
 821
 822                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 823         }
 824
 825         test := newDaemonMessageTest(t, t.TempDir(), server)
 826         test.Start(t, 0, 1, 2)
 827         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
 828         test.events[0].WaitFor(t, "idle", 1)
 829         test.events[1].WaitFor(t, "idle", 1)
 830
 831         require.Equal(t, "OneChat", test.events[1].LastEvent("user").Source)
 832         require.Equal(t, "Check the gate size.", test.events[1].Last("user"))
 833         require.Equal(t, "Garden", test.events[0].LastEvent("work").Source)
 834         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
 835         require.Equal(t, []string{"user", "busy", "reply", "idle"}, test.events[1].Kinds())
 836         require.Equal(t, []string{"work", "busy", "reply", "idle"}, test.events[0].Kinds())
 837         require.Equal(t, 2, server.Calls())
 838         require.Zero(t, test.agents[2].memory.Len())
 839         require.Empty(t, test.events[2].Kinds())
 840
 841         first, err := test.agents[1].memory.Message(0)
 842         require.NoError(t, err)
 843         require.Equal(t, "user", first.Kind)
 844
 845         report, err := test.agents[0].memory.Message(0)
 846         require.NoError(t, err)
 847         require.Equal(t, "work", report.Kind)
 848         require.Equal(t, "[Garden] Approve the gate size.", report.Text)
 849         require.Contains(t, test.logs.String(), "turn started name=Garden")
 850         require.Contains(t, test.logs.String(), "turn done name=Garden")
 851         require.NotContains(t, test.logs.String(), "gate size")
 852         require.Empty(t, test.Queue(t).Pending)
 853
 854         info, err := os.Stat(test.messages.path)
 855         require.NoError(t, err)
 856         require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
 857 }
 858
 859 func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
 860         server := newModelFake(t)
 861         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 862                 text := "I will ask the user."
 863                 if call.body["model"] == "gpt-test-garden" {
 864                         text = "Approve the gate size."
 865                 }
 866
 867                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 868         }
 869
 870         test := newDaemonMessageTest(t, t.TempDir(), server)
 871         telegram := newTelegramFake(t)
 872         bot := newTelegramBot(newTelegramAPI(telegram.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), discardLog())
 873         test.loops[0].notify = func(event Event) {
 874                 test.events[0].Add(event)
 875                 bot.Notify(event)
 876         }
 877
 878         ctx, cancel := context.WithCancel(t.Context())
 879         defer cancel()
 880
 881         done := make(chan struct{})
 882         go func() {
 883                 defer close(done)
 884
 885                 bot.send(ctx)
 886         }()
 887
 888         t.Cleanup(func() {
 889                 cancel()
 890                 <-done
 891         })
 892
 893         test.Start(t, 0, 1)
 894         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
 895         test.events[0].WaitFor(t, "idle", 1)
 896         require.Eventually(t, func() bool {
 897                 return len(telegram.Sent()) >= 2
 898         }, 10*time.Second, time.Millisecond)
 899
 900         require.Equal(t, []string{"[Garden] finished", "I will ask the user."}, telegram.Sent())
 901         require.Equal(t, subagentFinished, test.events[0].LastEvent("work").Status)
 902         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
 903 }
 904
 905 func TestQueuedAgentMessageFailureKeepsItsOperationAndCause(t *testing.T) {
 906         server := newModelFake(t)
 907         test := newDaemonMessageTest(t, t.TempDir(), server)
 908         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 909
 910         failure := errors.New("fake queued message failure")
 911         test.loops[1].fail(failure)
 912
 913         err := test.loops[1].Run(t.Context())
 914         require.ErrorIs(t, err, failure)
 915         require.ErrorContains(t, err, "check turn loop failure")
 916         require.Zero(t, server.Calls())
 917 }
 918
 919 func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.T) {
 920         testCases := []struct {
 921                 name string
 922                 text string
 923         }{
 924                 {
 925                         name: "sends nothing for an empty final reply",
 926                         text: "",
 927                 },
 928                 {
 929                         name: "sends nothing for a whitespace final reply",
 930                         text: " \n\t ",
 931                 },
 932         }
 933
 934         for _, tc := range testCases {
 935                 t.Run(tc.name, func(t *testing.T) {
 936                         server := newModelFake(t)
 937                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 938                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(tc.text)}, 100, 0, 5))
 939                         }
 940
 941                         test := newDaemonMessageTest(t, t.TempDir(), server)
 942                         test.Start(t, 0, 1, 2)
 943                         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 944                         test.events[1].WaitFor(t, "idle", 1)
 945                         require.Equal(t, 1, server.Calls())
 946                         require.Zero(t, test.agents[0].memory.Len())
 947                         require.Empty(t, test.events[0].Kinds())
 948                         require.False(t, test.loops[0].waiting())
 949                         require.Empty(t, test.events[2].Kinds())
 950                         require.Empty(t, test.Queue(t).Pending)
 951                 })
 952         }
 953 }
 954
 955 func TestBusyChildQueuesTheNextParentMessage(t *testing.T) {
 956         server := newModelFake(t)
 957         arrived := make(chan struct{})
 958         release := make(chan struct{})
 959         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 960                 if call.number == 0 {
 961                         close(arrived)
 962                         <-release
 963                 }
 964
 965                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 966         }
 967
 968         test := newDaemonMessageTest(t, t.TempDir(), server)
 969         test.Start(t, 0, 1)
 970         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the first plan."))
 971         <-arrived
 972         require.True(t, test.loops[1].Busy())
 973         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the second plan too."))
 974         require.Equal(t, 1, server.Calls())
 975         require.Len(t, test.Queue(t).Pending, 2)
 976         close(release)
 977         test.events[1].WaitFor(t, "idle", 2)
 978         require.Equal(t, 2, server.Calls())
 979
 980         texts := turnInputTexts(t, server.Call(1))
 981         require.Equal(t, "Check the second plan too.", texts[len(texts)-1])
 982         require.Empty(t, test.Queue(t).Pending)
 983 }
 984
 985 func TestBusyParentQueuesItsChildReport(t *testing.T) {
 986         server := newModelFake(t)
 987         arrived := make(chan struct{})
 988         release := make(chan struct{})
 989         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 990                 text := "The plan is ready."
 991                 if call.number == 0 {
 992                         close(arrived)
 993                         <-release
 994
 995                         text = "I will continue."
 996                 }
 997
 998                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 999         }
1000
1001         test := newDaemonMessageTest(t, t.TempDir(), server)
1002         test.Start(t, 0, 1)
1003
1004         _, err := test.loops[0].Submit("terminal", "Start the work.")
1005         require.NoError(t, err)
1006         <-arrived
1007         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Prepare the plan."))
1008         test.events[1].WaitFor(t, "idle", 1)
1009         require.Equal(t, "[Garden] The plan is ready.", test.events[0].Last("work"))
1010         require.True(t, test.loops[0].Busy())
1011         require.Equal(t, 2, server.Calls())
1012         close(release)
1013         test.events[0].WaitFor(t, "idle", 2)
1014         require.Equal(t, 3, server.Calls())
1015
1016         texts := turnInputTexts(t, server.Call(2))
1017         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
1018         require.Empty(t, test.Queue(t).Pending)
1019 }
1020
1021 func TestChildReportsOnlyTheFinalReplyAndTakesQueuedToolInput(t *testing.T) {
1022         server := newModelFake(t)
1023         arrived := make(chan struct{})
1024         release := make(chan struct{})
1025         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
1026                 if call.number == 0 {
1027                         close(arrived)
1028                         <-release
1029                         modelWriteEvents(w, modelCompletedEvent([]any{
1030                                 modelMessageItem("I will check the plan."),
1031                                 modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`),
1032                         }, 100, 0, 5))
1033
1034                         return
1035                 }
1036
1037                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan needs approval.")}, 100, 0, 5))
1038         }
1039
1040         test := newDaemonMessageTest(t, t.TempDir(), server)
1041         test.Start(t, 1)
1042         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the plan."))
1043         <-arrived
1044         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Also check the date."))
1045         close(release)
1046         test.events[1].WaitFor(t, "idle", 1)
1047         require.Equal(t, 2, server.Calls())
1048         require.Equal(t, 1, test.events[1].Count("busy"))
1049         require.Equal(t, 1, test.events[0].Count("work"))
1050         require.Equal(t, "[Garden] The plan needs approval.", test.events[0].Last("work"))
1051         require.NotContains(t, test.events[0].Last("work"), "I will check")
1052
1053         texts := turnInputTexts(t, server.Call(1))
1054         require.Equal(t, "Also check the date.", texts[len(texts)-1])
1055
1056         queue := test.Queue(t)
1057         require.Len(t, queue.Pending, 1)
1058         require.Equal(t, "work", queue.Pending[0].Kind)
1059 }
1060
1061 func TestQueuedParentMessageSurvivesDaemonRestart(t *testing.T) {
1062         dir := t.TempDir()
1063         server := newModelFake(t)
1064         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
1065                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1066         }
1067
1068         first := newDaemonMessageTest(t, dir, server)
1069         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Remember the garden plan."))
1070         require.Equal(t, 1, first.agents[1].memory.Len())
1071         first.Close(t)
1072
1073         second := newDaemonMessageTest(t, dir, server)
1074         require.Equal(t, 1, second.agents[1].memory.Len())
1075         require.Equal(t, "OneChat", second.events[1].LastEvent("user").Source)
1076         second.Start(t, 0, 1)
1077         second.events[1].WaitFor(t, "idle", 1)
1078         require.Equal(t, 1, server.Calls())
1079         require.Empty(t, second.Queue(t).Pending)
1080         second.Close(t)
1081
1082         third := newDaemonMessageTest(t, dir, server)
1083         require.Equal(t, 1, third.agents[1].memory.Len())
1084         require.False(t, third.loops[1].waiting())
1085         require.Empty(t, third.events[1].Kinds())
1086 }
1087
1088 func TestQueuedChildReportSurvivesDaemonRestart(t *testing.T) {
1089         dir := t.TempDir()
1090         server := newModelFake(t)
1091         first := newDaemonMessageTest(t, dir, server)
1092         first.Start(t, 1)
1093         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
1094         first.events[1].WaitFor(t, "idle", 1)
1095         require.Equal(t, "[Garden] a line", first.events[0].Last("work"))
1096         require.Len(t, first.Queue(t).Pending, 1)
1097         first.Close(t)
1098
1099         second := newDaemonMessageTest(t, dir, server)
1100         require.Equal(t, 1, second.agents[0].memory.Len())
1101         require.False(t, second.loops[1].waiting())
1102         second.Start(t, 0, 1)
1103         second.events[0].WaitFor(t, "idle", 1)
1104         require.Equal(t, 2, server.Calls())
1105         require.Equal(t, 2, second.agents[0].memory.Len())
1106         require.Empty(t, second.Queue(t).Pending)
1107         second.Close(t)
1108
1109         third := newDaemonMessageTest(t, dir, server)
1110         require.Equal(t, 2, third.agents[0].memory.Len())
1111         require.False(t, third.loops[0].waiting())
1112         require.Empty(t, third.events[0].Kinds())
1113 }
1114
1115 func TestInterruptedChildDeliveryResumesAfterDaemonRestart(t *testing.T) {
1116         dir := t.TempDir()
1117         server := newModelFake(t)
1118         arrived := make(chan struct{})
1119         release := make(chan struct{})
1120         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
1121                 if call.number == 0 {
1122                         close(arrived)
1123                         <-release
1124                 }
1125
1126                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1127         }
1128
1129         first := newDaemonMessageTest(t, dir, server)
1130         first.Start(t, 1)
1131         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Finish the garden plan."))
1132         <-arrived
1133         first.Close(t)
1134         close(release)
1135
1136         second := newDaemonMessageTest(t, dir, server)
1137         require.Equal(t, 1, second.agents[1].memory.Len())
1138         second.Start(t, 1)
1139         second.events[1].WaitFor(t, "idle", 1)
1140         require.Equal(t, 2, server.Calls())
1141         require.Equal(t, 1, second.agents[1].memory.Len())
1142         require.Empty(t, second.Queue(t).Pending)
1143 }
1144
1145 func TestInterruptedParentReportResumesAfterDaemonRestart(t *testing.T) {
1146         dir := t.TempDir()
1147         server := newModelFake(t)
1148         arrived := make(chan struct{})
1149         release := make(chan struct{})
1150         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
1151                 if call.number == 1 {
1152                         close(arrived)
1153                         <-release
1154                 }
1155
1156                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan is ready.")}, 100, 0, 5))
1157         }
1158
1159         first := newDaemonMessageTest(t, dir, server)
1160         first.Start(t, 0, 1)
1161         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
1162         <-arrived
1163         first.Close(t)
1164         close(release)
1165
1166         second := newDaemonMessageTest(t, dir, server)
1167         require.Equal(t, 1, second.agents[0].memory.Len())
1168         require.False(t, second.loops[1].waiting())
1169         second.Start(t, 0, 1)
1170         second.events[0].WaitFor(t, "idle", 1)
1171         require.Equal(t, 3, server.Calls())
1172         require.Equal(t, 2, second.agents[0].memory.Len())
1173         require.Empty(t, second.Queue(t).Pending)
1174
1175         texts := turnInputTexts(t, server.Call(2))
1176         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
1177 }
1178
1179 func TestAgentMessageWriteFailureStopsDelivery(t *testing.T) {
1180         server := newModelFake(t)
1181         test := newDaemonMessageTest(t, t.TempDir(), server)
1182         require.NoError(t, os.Mkdir(test.messages.path, 0o700))
1183         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."), "save agent messages")
1184         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Do not send again."), "save agent messages")
1185
1186         for _, loop := range test.loops {
1187                 require.ErrorContains(t, loop.failed(), "save agent messages")
1188                 require.Zero(t, loop.memory.Len())
1189         }
1190 }
1191
1192 func TestAgentMessageSnapshotBlocksDelivery(t *testing.T) {
1193         server := newModelFake(t)
1194         test := newDaemonMessageTest(t, t.TempDir(), server)
1195         entered := make(chan struct{})
1196         release := make(chan struct{})
1197
1198         var once sync.Once
1199
1200         defer once.Do(func() {
1201                 close(release)
1202         })
1203
1204         snapshot := make(chan error, 1)
1205         go func() {
1206                 snapshot <- test.messages.Snapshot(func() error {
1207                         return daemonSnapshot(t.Context(), test.loops, func() error {
1208                                 close(entered)
1209                                 <-release
1210
1211                                 return nil
1212                         })
1213                 })
1214         }()
1215
1216         <-entered
1217
1218         sent := make(chan error, 1)
1219         started := make(chan struct{})
1220         go func() {
1221                 close(started)
1222
1223                 sent <- test.messages.Tell("OneChat", "Garden", "Record the plan after the copy.")
1224         }()
1225
1226         <-started
1227         require.Never(t, func() bool {
1228                 return len(sent) > 0
1229         }, 50*time.Millisecond, time.Millisecond)
1230
1231         once.Do(func() {
1232                 close(release)
1233         })
1234         require.NoError(t, <-snapshot)
1235         require.NoError(t, <-sent)
1236         require.Equal(t, 1, test.agents[1].memory.Len())
1237         require.Len(t, test.Queue(t).Pending, 1)
1238 }
1239
1240 func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) {
1241         server := newModelFake(t)
1242
1243         test := newDaemonMessageTest(t, t.TempDir(), server)
1244         testCases := []struct {
1245                 name   string
1246                 source string
1247                 target string
1248         }{
1249                 {
1250                         name:   "no sibling messages",
1251                         source: "Garden",
1252                         target: "Library",
1253                 },
1254                 {
1255                         name:   "no arbitrary child reports",
1256                         source: "Garden",
1257                         target: "OneChat",
1258                 },
1259                 {
1260                         name:   "no self messages",
1261                         source: "OneChat",
1262                         target: "OneChat",
1263                 },
1264                 {
1265                         name:   "refuses an unknown sender",
1266                         source: "Other",
1267                         target: "Garden",
1268                 },
1269                 {
1270                         name:   "refuses an unknown recipient",
1271                         source: "OneChat",
1272                         target: "Other",
1273                 },
1274         }
1275
1276         for _, tc := range testCases {
1277                 t.Run(tc.name, func(t *testing.T) {
1278                         require.ErrorContains(t, test.messages.Tell(tc.source, tc.target, "Do not deliver."), "not a child")
1279                 })
1280         }
1281
1282         for _, agent := range test.agents {
1283                 require.Zero(t, agent.memory.Len())
1284         }
1285 }
1286
1287 func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) {
1288         server := newModelFake(t)
1289         testCases := []struct {
1290                 name string
1291                 file string
1292         }{
1293                 {
1294                         name: "stops startup for broken JSON",
1295                         file: "{",
1296                 },
1297                 {
1298                         name: "stops startup for unknown sender",
1299                         file: `{"next":1,"pending":[{"id":1,"source":"Other","target":"Garden","kind":"user"}]}`,
1300                 },
1301                 {
1302                         name: "stops startup for sibling delivery",
1303                         file: `{"next":1,"pending":[{"id":1,"source":"Library","target":"Garden","kind":"user"}]}`,
1304                 },
1305                 {
1306                         name: "stops startup for invalid sequence",
1307                         file: `{"next":0,"pending":[{"id":1,"source":"OneChat","target":"Garden","kind":"user"}]}`,
1308                 },
1309         }
1310
1311         for _, tc := range testCases {
1312                 t.Run(tc.name, func(t *testing.T) {
1313                         dir := t.TempDir()
1314                         test := newDaemonMessageTest(t, dir, server)
1315                         path := test.messages.path
1316                         test.Close(t)
1317                         require.NoError(t, os.WriteFile(path, []byte(tc.file), 0o600))
1318
1319                         agents, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
1320                         require.NoError(t, err)
1321
1322                         defer func() {
1323                                 require.NoError(t, agents.Close())
1324                         }()
1325
1326                         var loops []*TurnLoop
1327
1328                         for _, agent := range agents {
1329                                 loop, err := agent.Loop(nil, func(Event) {
1330                                 })
1331                                 require.NoError(t, err)
1332
1333                                 loops = append(loops, loop)
1334                         }
1335
1336                         _, err = openDaemonMessages(loops)
1337                         require.Error(t, err)
1338
1339                         for _, agent := range agents {
1340                                 require.Zero(t, agent.memory.Len())
1341                         }
1342                 })
1343         }
1344 }
1345
1346 func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
1347         server := newModelFake(t)
1348         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
1349                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1350         }
1351
1352         test := newDaemonMessageTest(t, t.TempDir(), server)
1353         child := test.agents[1]
1354         at := child.clock().Add(time.Minute).Format(scheduleAtLayout)
1355         output := test.loops[1].tools.run(t.Context(), "schedule_set", `{"text":"Check the plants.","at":"`+at+`","cron":null}`)
1356         require.Contains(t, output, "Schedule 1 set.")
1357         require.Equal(t, "Garden", child.schedules.sorted()[0].Owner)
1358
1359         child.schedules.clock = func() time.Time {
1360                 return child.clock().Add(time.Minute)
1361         }
1362         require.ErrorContains(t, child.schedules.deliver(t.Context(), test.loops[0]), "does not match")
1363         require.NoError(t, child.schedules.deliver(t.Context(), test.loops[1]))
1364         test.Start(t, 0, 1, 2)
1365         test.events[1].WaitFor(t, "idle", 1)
1366         require.Equal(t, 1, server.Calls())
1367         require.Equal(t, child.agent.CacheKey(), server.Call(0).body["prompt_cache_key"])
1368         require.Zero(t, test.agents[0].memory.Len())
1369         require.Zero(t, test.agents[2].memory.Len())
1370         require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.")
1371         require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source)
1372 }
1373
1374 func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) {
1375         testCases := []struct {
1376                 name string
1377                 chat WhatsAppChat
1378         }{
1379                 {
1380                         name: "routes a private chat only to its child",
1381                         chat: WhatsAppChat{
1382                                 ID:   whatsappTestAna.String(),
1383                                 Name: "Ana",
1384                         },
1385                 },
1386                 {
1387                         name: "routes an unmentioned group only to its child",
1388                         chat: WhatsAppChat{
1389                                 ID:    whatsappTestFamily.String(),
1390                                 Name:  "Garden plans",
1391                                 Group: true,
1392                         },
1393                 },
1394         }
1395
1396         for _, tc := range testCases {
1397                 t.Run(tc.name, func(t *testing.T) {
1398                         server := newModelFake(t)
1399                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
1400                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1401                         }
1402
1403                         dir := t.TempDir()
1404                         test := newDaemonMessageTest(t, dir, server)
1405                         inbox := whatsappRouteTestInbox(t, dir)
1406                         require.NoError(t, test.messages.routes.Set("whatsapp:"+tc.chat.ID, "Garden"))
1407                         whatsappRouteTestAdd(t, inbox, tc.chat, "The gate needs a new latch.")
1408                         test.Start(t, 0, 1, 2)
1409                         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1410                         test.events[1].WaitFor(t, "idle", 1)
1411                         require.Equal(t, 1, server.Calls())
1412                         require.Equal(t, "gpt-test-garden", server.Call(0).body["model"])
1413                         require.Contains(t, string(server.Call(0).raw), "The gate needs a new latch.")
1414                         require.Zero(t, test.agents[0].memory.Len())
1415                         require.Empty(t, test.events[0].Kinds())
1416                         require.Zero(t, test.agents[2].memory.Len())
1417                         require.False(t, test.loops[0].waiting())
1418                         require.Equal(t, whatsappSource, test.events[1].LastEvent("work").Source)
1419                         require.Empty(t, test.events[1].LastEvent("work").Status)
1420                         require.Equal(t, 1, test.agents[1].memory.Len())
1421                         require.Empty(t, inbox.state.Closed())
1422                         require.Empty(t, test.Queue(t).Pending)
1423                         require.Contains(t, test.logs.String(), "batch routed name=Garden kind=whatsapp id=1 to=Garden")
1424                         require.NotContains(t, test.logs.String(), tc.chat.ID)
1425                         require.NotContains(t, test.logs.String(), "new latch")
1426                 })
1427         }
1428 }
1429
1430 func TestWhatsAppUnrouteRestoresRootDelivery(t *testing.T) {
1431         dir := t.TempDir()
1432         test := newDaemonMessageTest(t, dir, newModelFake(t))
1433         inbox := whatsappRouteTestInbox(t, dir)
1434         chat := WhatsAppChat{
1435                 ID:   whatsappTestAna.String(),
1436                 Name: "Ana",
1437         }
1438
1439         source := "whatsapp:" + chat.ID
1440         require.NoError(t, test.messages.routes.Set(source, "Garden"))
1441         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
1442         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1443         require.Zero(t, test.agents[0].memory.Len())
1444         require.True(t, test.loops[1].waiting())
1445         require.NoError(t, test.messages.routes.Set(source, ""))
1446         whatsappRouteTestAdd(t, inbox, chat, "Please check the other plan.")
1447         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1448         require.Equal(t, 1, test.agents[0].memory.Len())
1449         require.Equal(t, 1, test.agents[1].memory.Len())
1450         require.True(t, test.loops[0].waiting())
1451         require.Contains(t, test.events[0].Last("work"), "Please check the other plan.")
1452         require.NotContains(t, test.events[0].Last("work"), "Keep the gate plan.")
1453         require.Empty(t, inbox.state.Closed())
1454 }
1455
1456 func TestMissingWhatsAppRouteChildFallsBackToRoot(t *testing.T) {
1457         dir := t.TempDir()
1458         test := newDaemonMessageTest(t, dir, newModelFake(t))
1459         inbox := whatsappRouteTestInbox(t, dir)
1460         chat := WhatsAppChat{
1461                 ID:   whatsappTestAna.String(),
1462                 Name: "Ana",
1463         }
1464
1465         require.NoError(t, test.messages.routes.Set("whatsapp:"+chat.ID, "Missing"))
1466         whatsappRouteTestAdd(t, inbox, chat, "Please check the plan.")
1467         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1468         require.Equal(t, 1, test.agents[0].memory.Len())
1469         require.True(t, test.loops[0].waiting())
1470         require.Zero(t, test.agents[1].memory.Len())
1471         require.False(t, test.loops[1].waiting())
1472         require.Empty(t, test.events[0].LastEvent("work").Status)
1473         require.Equal(t, whatsappSource, test.events[0].LastEvent("work").Source)
1474         require.Empty(t, inbox.state.Closed())
1475 }
1476
1477 func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) {
1478         testCases := []struct {
1479                 name   string
1480                 finish bool
1481         }{
1482                 {
1483                         name:   "queued turn resumes",
1484                         finish: false,
1485                 },
1486                 {
1487                         name:   "finished turn stays finished",
1488                         finish: true,
1489                 },
1490         }
1491
1492         for _, tc := range testCases {
1493                 t.Run(tc.name, func(t *testing.T) {
1494                         dir := t.TempDir()
1495                         server := newModelFake(t)
1496                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
1497                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
1498                         }
1499
1500                         first := newDaemonMessageTest(t, dir, server)
1501                         inbox := whatsappRouteTestInbox(t, dir)
1502                         chat := WhatsAppChat{
1503                                 ID:   whatsappTestAna.String(),
1504                                 Name: "Ana",
1505                         }
1506
1507                         source := "whatsapp:" + chat.ID
1508                         require.NoError(t, first.messages.routes.Set(source, "Garden"))
1509                         whatsappRouteTestAdd(t, inbox, chat, "Remember the gate plan.")
1510                         require.NoError(t, inbox.state.CloseAll())
1511
1512                         batch := inbox.state.Closed()[0]
1513                         routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
1514                         require.NoError(t, err)
1515                         require.True(t, routed)
1516
1517                         if tc.finish {
1518                                 first.Start(t, 1)
1519                                 first.events[1].WaitFor(t, "idle", 1)
1520                                 require.Empty(t, first.Queue(t).Pending)
1521                         }
1522
1523                         first.Close(t)
1524
1525                         second := newDaemonMessageTest(t, dir, server)
1526                         require.Equal(t, source+" -> Garden", second.messages.routes.List())
1527                         require.Equal(t, 1, second.agents[1].memory.Len())
1528                         require.Equal(t, !tc.finish, second.loops[1].waiting())
1529                         require.NoError(t, second.messages.routes.Set(source, ""))
1530
1531                         restored := whatsappRouteTestInbox(t, dir)
1532                         require.Len(t, restored.state.Closed(), 1)
1533                         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
1534                         require.Empty(t, restored.state.Closed())
1535                         require.Zero(t, second.agents[0].memory.Len())
1536                         require.Equal(t, 1, second.agents[1].memory.Len())
1537
1538                         if !tc.finish {
1539                                 second.Start(t, 1)
1540                                 second.events[1].WaitFor(t, "idle", 1)
1541                         }
1542
1543                         require.Equal(t, 1, server.Calls())
1544                         require.Empty(t, second.Queue(t).Pending)
1545                         second.Close(t)
1546
1547                         third := newDaemonMessageTest(t, dir, server)
1548                         require.False(t, third.loops[1].waiting())
1549                         require.Empty(t, third.events[1].Kinds())
1550                         require.Equal(t, 1, third.agents[1].memory.Len())
1551                 })
1552         }
1553 }
1554
1555 func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) {
1556         testCases := []struct {
1557                 name string
1558                 id   *int
1559         }{
1560                 {
1561                         name: "recovers one message before its log ID is stored",
1562                         id:   nil,
1563                 },
1564                 {
1565                         name: "recovers one message after its log ID is stored",
1566                         id:   new(0),
1567                 },
1568         }
1569
1570         for _, tc := range testCases {
1571                 t.Run(tc.name, func(t *testing.T) {
1572                         dir := t.TempDir()
1573                         server := newModelFake(t)
1574                         first := newDaemonMessageTest(t, dir, server)
1575                         path := first.messages.path
1576                         first.Close(t)
1577
1578                         file := DaemonMessageFile{
1579                                 Next:            1,
1580                                 WhatsAppBatches: []int{1},
1581                                 Pending: []DaemonMessage{{
1582                                         ID:            1,
1583                                         Source:        "OneChat",
1584                                         Target:        "Garden",
1585                                         Kind:          "work",
1586                                         Text:          "[WhatsApp] Keep the gate plan.",
1587                                         LogID:         tc.id,
1588                                         WhatsAppBatch: 1,
1589                                 }},
1590                         }
1591
1592                         data, err := json.Marshal(file)
1593                         require.NoError(t, err)
1594                         require.NoError(t, os.WriteFile(path, data, 0o600))
1595
1596                         second := newDaemonMessageTest(t, dir, server)
1597                         require.Equal(t, 1, second.agents[1].memory.Len())
1598                         require.True(t, second.loops[1].waiting())
1599                         require.Equal(t, "[WhatsApp] Keep the gate plan.", second.events[1].Last("work"))
1600                         require.Equal(t, whatsappSource, second.events[1].LastEvent("work").Source)
1601                         second.Close(t)
1602
1603                         third := newDaemonMessageTest(t, dir, server)
1604                         require.Equal(t, 1, third.agents[1].memory.Len())
1605                         require.True(t, third.loops[1].waiting())
1606                         require.Zero(t, third.agents[0].memory.Len())
1607                 })
1608         }
1609 }
1610
1611 func TestQueuedWhatsAppBatchUsesRootIfItsChildIsRemoved(t *testing.T) {
1612         dir := t.TempDir()
1613         server := newModelFake(t)
1614         first := newDaemonMessageTest(t, dir, server)
1615         inbox := whatsappRouteTestInbox(t, dir)
1616         chat := WhatsAppChat{
1617                 ID:   whatsappTestAna.String(),
1618                 Name: "Ana",
1619         }
1620
1621         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
1622         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
1623         require.NoError(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true))
1624         first.Close(t)
1625         require.NoError(t, os.RemoveAll(filepath.Join(dir, daemonAgentDirName, "children", "Garden")))
1626
1627         second := newDaemonMessageTest(t, dir, server)
1628         require.Equal(t, 1, second.agents[0].memory.Len())
1629         require.True(t, second.loops[0].waiting())
1630         require.Equal(t, whatsappSource, second.events[0].LastEvent("work").Source)
1631         require.Empty(t, second.events[0].LastEvent("work").Status)
1632         require.Contains(t, second.events[0].Last("work"), "Keep the gate plan.")
1633         require.False(t, second.loops[1].waiting())
1634         second.Close(t)
1635
1636         third := newDaemonMessageTest(t, dir, server)
1637         require.Equal(t, 1, third.agents[0].memory.Len())
1638         require.True(t, third.loops[0].waiting())
1639 }
1640
1641 func TestRoutedWhatsAppBatchesCanCloseOutOfNumberOrder(t *testing.T) {
1642         dir := t.TempDir()
1643         test := newDaemonMessageTest(t, dir, newModelFake(t))
1644         inbox := whatsappRouteTestInbox(t, dir)
1645         first := WhatsAppChat{
1646                 ID:   whatsappTestAna.String(),
1647                 Name: "Ana",
1648         }
1649
1650         second := WhatsAppChat{
1651                 ID:    whatsappTestFamily.String(),
1652                 Name:  "Garden plans",
1653                 Group: true,
1654         }
1655
1656         require.NoError(t, test.messages.routes.Set("whatsapp:"+first.ID, "Garden"))
1657         require.NoError(t, test.messages.routes.Set("whatsapp:"+second.ID, "Garden"))
1658
1659         now := time.Now()
1660         _, err := inbox.state.Add(first, WhatsAppMessage{
1661                 ID:     "first",
1662                 Text:   "The first chat is still open.",
1663                 Sender: "Ana",
1664                 Time:   now,
1665         }, now)
1666         require.NoError(t, err)
1667
1668         _, err = inbox.state.Add(second, WhatsAppMessage{
1669                 ID:     "second",
1670                 Text:   "The second chat is quiet.",
1671                 Sender: "Bruno",
1672                 Time:   now,
1673         }, now.Add(-time.Minute))
1674         require.NoError(t, err)
1675         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
1676         require.Equal(t, 1, test.agents[1].memory.Len())
1677         require.Contains(t, test.events[1].Last("work"), "The second chat is quiet.")
1678         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
1679         require.Equal(t, 2, test.agents[1].memory.Len())
1680         require.Contains(t, test.events[1].Last("work"), "The first chat is still open.")
1681         require.Zero(t, test.agents[0].memory.Len())
1682         require.Empty(t, test.Queue(t).WhatsAppBatches)
1683 }
1684
1685 func TestWhatsAppCatchupSeparatesChildAndRootBatches(t *testing.T) {
1686         dir := t.TempDir()
1687         test := newDaemonMessageTest(t, dir, newModelFake(t))
1688         inbox := whatsappRouteTestInbox(t, dir)
1689         garden := WhatsAppChat{
1690                 ID:    whatsappTestFamily.String(),
1691                 Name:  "Garden plans",
1692                 Group: true,
1693         }
1694
1695         library := WhatsAppChat{
1696                 ID:   whatsappTestBruno.String(),
1697                 Name: "Bruno",
1698         }
1699
1700         root := WhatsAppChat{
1701                 ID:   whatsappTestAna.String(),
1702                 Name: "Ana",
1703         }
1704
1705         require.NoError(t, test.messages.routes.Set("whatsapp:"+garden.ID, "Garden"))
1706         require.NoError(t, test.messages.routes.Set("whatsapp:"+library.ID, "Library"))
1707         inbox.waitForCatchup()
1708         whatsappRouteTestAdd(t, inbox, garden, "Check the gate.")
1709         whatsappRouteTestAdd(t, inbox, library, "Check the book.")
1710         whatsappRouteTestAdd(t, inbox, root, "Check the mixed chat.")
1711         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
1712
1713         for _, agent := range test.agents {
1714                 require.Zero(t, agent.memory.Len())
1715         }
1716
1717         inbox.endCatchup()
1718         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
1719
1720         for _, loop := range test.loops {
1721                 require.Equal(t, 1, loop.memory.Len())
1722                 require.True(t, loop.waiting())
1723         }
1724
1725         require.Contains(t, test.events[0].Last("work"), "Check the mixed chat.")
1726         require.Contains(t, test.events[1].Last("work"), "Check the gate.")
1727         require.Contains(t, test.events[2].Last("work"), "Check the book.")
1728         require.Empty(t, inbox.state.Closed())
1729         require.Empty(t, test.Queue(t).WhatsAppBatches)
1730 }
1731
1732 func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) {
1733         dir := t.TempDir()
1734         server := newModelFake(t)
1735         first := newDaemonMessageTest(t, dir, server)
1736         inbox := whatsappRouteTestInbox(t, dir)
1737         chat := WhatsAppChat{
1738                 ID:   whatsappTestAna.String(),
1739                 Name: "Ana",
1740         }
1741
1742         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
1743         require.NoError(t, os.Mkdir(first.messages.path, 0o700))
1744         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
1745         require.ErrorContains(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true), "save agent messages")
1746         require.Len(t, inbox.state.Closed(), 1)
1747
1748         for _, loop := range first.loops {
1749                 require.ErrorContains(t, loop.failed(), "save agent messages")
1750                 require.Zero(t, loop.memory.Len())
1751         }
1752
1753         path := first.messages.path
1754         first.Close(t)
1755         require.NoError(t, os.Remove(path))
1756
1757         second := newDaemonMessageTest(t, dir, server)
1758         restored := whatsappRouteTestInbox(t, dir)
1759         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
1760         require.Equal(t, 1, second.agents[1].memory.Len())
1761         require.True(t, second.loops[1].waiting())
1762         require.Zero(t, second.agents[0].memory.Len())
1763         require.Empty(t, restored.state.Closed())
1764 }
1765
1766 func whatsappRouteTestInbox(t *testing.T, dir string) *WhatsAppInbox {
1767         t.Helper()
1768
1769         state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName))
1770         require.NoError(t, err)
1771
1772         return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()), &ImageStore{dir: filepath.Join(dir, imageDirName)}, discardLog())
1773 }
1774
1775 func whatsappRouteTestAdd(t *testing.T, inbox *WhatsAppInbox, chat WhatsAppChat, text string) {
1776         t.Helper()
1777
1778         at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
1779         _, err := inbox.state.Add(chat, WhatsAppMessage{
1780                 ID:     text,
1781                 Sender: "Ana",
1782                 Text:   text,
1783                 Time:   at,
1784         }, at)
1785         require.NoError(t, err)
1786 }