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

   1 package main
   2
   3 import (
   4         "context"
   5         "encoding/json"
   6         "errors"
   7         "net/http"
   8         "os"
   9         "path/filepath"
  10         "sync"
  11         "testing"
  12         "time"
  13
  14         "github.com/stretchr/testify/require"
  15 )
  16
  17 func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
  18         server := newModelFake(t)
  19         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
  20                 text := "I will ask the user."
  21                 if call.body["model"] == "gpt-test-garden" {
  22                         text = "Approve the gate size."
  23                 }
  24
  25                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
  26         }
  27         test := newDaemonMessageTest(t, t.TempDir(), server)
  28         test.Start(t, 0, 1, 2)
  29         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
  30         test.events[0].WaitFor(t, "idle", 1)
  31         test.events[1].WaitFor(t, "idle", 1)
  32
  33         require.Equal(t, "OneChat", test.events[1].LastEvent("user").Source)
  34         require.Equal(t, "Check the gate size.", test.events[1].Last("user"))
  35         require.Equal(t, "Garden", test.events[0].LastEvent("work").Source)
  36         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
  37         require.Equal(t, []string{"user", "busy", "reply", "idle"}, test.events[1].Kinds())
  38         require.Equal(t, []string{"work", "busy", "reply", "idle"}, test.events[0].Kinds())
  39         require.Equal(t, 2, server.Calls())
  40         require.Zero(t, test.agents[2].memory.Len())
  41         require.Empty(t, test.events[2].Kinds())
  42
  43         first, err := test.agents[1].memory.Message(0)
  44         require.NoError(t, err)
  45         require.Equal(t, "user", first.Kind)
  46
  47         report, err := test.agents[0].memory.Message(0)
  48         require.NoError(t, err)
  49         require.Equal(t, "work", report.Kind)
  50         require.Equal(t, "[Garden] Approve the gate size.", report.Text)
  51         require.Contains(t, test.logs.String(), "turn started name=Garden")
  52         require.Contains(t, test.logs.String(), "turn done name=Garden")
  53         require.NotContains(t, test.logs.String(), "gate size")
  54         require.Empty(t, test.Queue(t).Pending)
  55
  56         info, err := os.Stat(test.messages.path)
  57         require.NoError(t, err)
  58         require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
  59 }
  60
  61 func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
  62         server := newModelFake(t)
  63         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
  64                 text := "I will ask the user."
  65                 if call.body["model"] == "gpt-test-garden" {
  66                         text = "Approve the gate size."
  67                 }
  68
  69                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
  70         }
  71         test := newDaemonMessageTest(t, t.TempDir(), server)
  72         telegram := newTelegramFake(t)
  73         bot := newTelegramBot(newTelegramAPI(telegram.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), discardLog())
  74         test.loops[0].notify = func(event Event) {
  75                 test.events[0].Add(event)
  76                 bot.Notify(event)
  77         }
  78
  79         ctx, cancel := context.WithCancel(t.Context())
  80         defer cancel()
  81
  82         done := make(chan struct{})
  83         go func() {
  84                 defer close(done)
  85
  86                 bot.send(ctx)
  87         }()
  88
  89         t.Cleanup(func() {
  90                 cancel()
  91                 <-done
  92         })
  93
  94         test.Start(t, 0, 1)
  95         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
  96         test.events[0].WaitFor(t, "idle", 1)
  97         require.Eventually(t, func() bool {
  98                 return len(telegram.Sent()) >= 2
  99         }, 10*time.Second, time.Millisecond)
 100         require.Equal(t, []string{"[Garden] finished", "I will ask the user."}, telegram.Sent())
 101         require.Equal(t, subagentFinished, test.events[0].LastEvent("work").Status)
 102         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
 103 }
 104
 105 func TestQueuedAgentMessageFailureKeepsItsOperationAndCause(t *testing.T) {
 106         server := newModelFake(t)
 107         test := newDaemonMessageTest(t, t.TempDir(), server)
 108         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 109
 110         failure := errors.New("fake queued message failure")
 111         test.loops[1].fail(failure)
 112
 113         err := test.loops[1].Run(t.Context())
 114         require.ErrorIs(t, err, failure)
 115         require.ErrorContains(t, err, "check turn loop failure")
 116         require.Zero(t, server.Calls())
 117 }
 118
 119 func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.T) {
 120         testCases := []struct {
 121                 name string
 122                 text string
 123         }{
 124                 {
 125                         name: "empty final reply",
 126                         text: "",
 127                 },
 128                 {
 129                         name: "whitespace final reply",
 130                         text: " \n\t ",
 131                 },
 132         }
 133         for _, tc := range testCases {
 134                 t.Run(tc.name, func(t *testing.T) {
 135                         server := newModelFake(t)
 136                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 137                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(tc.text)}, 100, 0, 5))
 138                         }
 139                         test := newDaemonMessageTest(t, t.TempDir(), server)
 140                         test.Start(t, 0, 1, 2)
 141                         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 142                         test.events[1].WaitFor(t, "idle", 1)
 143                         require.Equal(t, 1, server.Calls())
 144                         require.Zero(t, test.agents[0].memory.Len())
 145                         require.Empty(t, test.events[0].Kinds())
 146                         require.False(t, test.loops[0].waiting())
 147                         require.Empty(t, test.events[2].Kinds())
 148                         require.Empty(t, test.Queue(t).Pending)
 149                 })
 150         }
 151 }
 152
 153 func TestBusyChildQueuesTheNextParentMessage(t *testing.T) {
 154         server := newModelFake(t)
 155         arrived := make(chan struct{})
 156         release := make(chan struct{})
 157         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 158                 if call.number == 0 {
 159                         close(arrived)
 160                         <-release
 161                 }
 162
 163                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 164         }
 165         test := newDaemonMessageTest(t, t.TempDir(), server)
 166         test.Start(t, 0, 1)
 167         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the first plan."))
 168         <-arrived
 169         require.True(t, test.loops[1].Busy())
 170         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the second plan too."))
 171         require.Equal(t, 1, server.Calls())
 172         require.Len(t, test.Queue(t).Pending, 2)
 173         close(release)
 174         test.events[1].WaitFor(t, "idle", 2)
 175         require.Equal(t, 2, server.Calls())
 176
 177         texts := turnInputTexts(t, server.Call(1))
 178         require.Equal(t, "Check the second plan too.", texts[len(texts)-1])
 179         require.Empty(t, test.Queue(t).Pending)
 180 }
 181
 182 func TestBusyParentQueuesItsChildReport(t *testing.T) {
 183         server := newModelFake(t)
 184         arrived := make(chan struct{})
 185         release := make(chan struct{})
 186         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 187                 text := "The plan is ready."
 188                 if call.number == 0 {
 189                         close(arrived)
 190                         <-release
 191
 192                         text = "I will continue."
 193                 }
 194
 195                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 196         }
 197         test := newDaemonMessageTest(t, t.TempDir(), server)
 198         test.Start(t, 0, 1)
 199
 200         _, err := test.loops[0].Submit("terminal", "Start the work.")
 201         require.NoError(t, err)
 202         <-arrived
 203         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Prepare the plan."))
 204         test.events[1].WaitFor(t, "idle", 1)
 205         require.Equal(t, "[Garden] The plan is ready.", test.events[0].Last("work"))
 206         require.True(t, test.loops[0].Busy())
 207         require.Equal(t, 2, server.Calls())
 208         close(release)
 209         test.events[0].WaitFor(t, "idle", 2)
 210         require.Equal(t, 3, server.Calls())
 211
 212         texts := turnInputTexts(t, server.Call(2))
 213         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
 214         require.Empty(t, test.Queue(t).Pending)
 215 }
 216
 217 func TestChildReportsOnlyTheFinalReplyAndTakesQueuedToolInput(t *testing.T) {
 218         server := newModelFake(t)
 219         arrived := make(chan struct{})
 220         release := make(chan struct{})
 221         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 222                 if call.number == 0 {
 223                         close(arrived)
 224                         <-release
 225                         modelWriteEvents(w, modelCompletedEvent([]any{
 226                                 modelMessageItem("I will check the plan."),
 227                                 modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`),
 228                         }, 100, 0, 5))
 229
 230                         return
 231                 }
 232
 233                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan needs approval.")}, 100, 0, 5))
 234         }
 235         test := newDaemonMessageTest(t, t.TempDir(), server)
 236         test.Start(t, 1)
 237         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the plan."))
 238         <-arrived
 239         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Also check the date."))
 240         close(release)
 241         test.events[1].WaitFor(t, "idle", 1)
 242         require.Equal(t, 2, server.Calls())
 243         require.Equal(t, 1, test.events[1].Count("busy"))
 244         require.Equal(t, 1, test.events[0].Count("work"))
 245         require.Equal(t, "[Garden] The plan needs approval.", test.events[0].Last("work"))
 246         require.NotContains(t, test.events[0].Last("work"), "I will check")
 247
 248         texts := turnInputTexts(t, server.Call(1))
 249         require.Equal(t, "Also check the date.", texts[len(texts)-1])
 250
 251         queue := test.Queue(t)
 252         require.Len(t, queue.Pending, 1)
 253         require.Equal(t, "work", queue.Pending[0].Kind)
 254 }
 255
 256 func TestQueuedParentMessageSurvivesDaemonRestart(t *testing.T) {
 257         dir := t.TempDir()
 258         server := newModelFake(t)
 259         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 260                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 261         }
 262         first := newDaemonMessageTest(t, dir, server)
 263         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Remember the garden plan."))
 264         require.Equal(t, 1, first.agents[1].memory.Len())
 265         first.Close(t)
 266
 267         second := newDaemonMessageTest(t, dir, server)
 268         require.Equal(t, 1, second.agents[1].memory.Len())
 269         require.Equal(t, "OneChat", second.events[1].LastEvent("user").Source)
 270         second.Start(t, 0, 1)
 271         second.events[1].WaitFor(t, "idle", 1)
 272         require.Equal(t, 1, server.Calls())
 273         require.Empty(t, second.Queue(t).Pending)
 274         second.Close(t)
 275
 276         third := newDaemonMessageTest(t, dir, server)
 277         require.Equal(t, 1, third.agents[1].memory.Len())
 278         require.False(t, third.loops[1].waiting())
 279         require.Empty(t, third.events[1].Kinds())
 280 }
 281
 282 func TestQueuedChildReportSurvivesDaemonRestart(t *testing.T) {
 283         dir := t.TempDir()
 284         server := newModelFake(t)
 285         first := newDaemonMessageTest(t, dir, server)
 286         first.Start(t, 1)
 287         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
 288         first.events[1].WaitFor(t, "idle", 1)
 289         require.Equal(t, "[Garden] a line", first.events[0].Last("work"))
 290         require.Len(t, first.Queue(t).Pending, 1)
 291         first.Close(t)
 292
 293         second := newDaemonMessageTest(t, dir, server)
 294         require.Equal(t, 1, second.agents[0].memory.Len())
 295         require.False(t, second.loops[1].waiting())
 296         second.Start(t, 0, 1)
 297         second.events[0].WaitFor(t, "idle", 1)
 298         require.Equal(t, 2, server.Calls())
 299         require.Equal(t, 2, second.agents[0].memory.Len())
 300         require.Empty(t, second.Queue(t).Pending)
 301         second.Close(t)
 302
 303         third := newDaemonMessageTest(t, dir, server)
 304         require.Equal(t, 2, third.agents[0].memory.Len())
 305         require.False(t, third.loops[0].waiting())
 306         require.Empty(t, third.events[0].Kinds())
 307 }
 308
 309 func TestInterruptedChildDeliveryResumesAfterDaemonRestart(t *testing.T) {
 310         dir := t.TempDir()
 311         server := newModelFake(t)
 312         arrived := make(chan struct{})
 313         release := make(chan struct{})
 314         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 315                 if call.number == 0 {
 316                         close(arrived)
 317                         <-release
 318                 }
 319
 320                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 321         }
 322         first := newDaemonMessageTest(t, dir, server)
 323         first.Start(t, 1)
 324         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Finish the garden plan."))
 325         <-arrived
 326         first.Close(t)
 327         close(release)
 328
 329         second := newDaemonMessageTest(t, dir, server)
 330         require.Equal(t, 1, second.agents[1].memory.Len())
 331         second.Start(t, 1)
 332         second.events[1].WaitFor(t, "idle", 1)
 333         require.Equal(t, 2, server.Calls())
 334         require.Equal(t, 1, second.agents[1].memory.Len())
 335         require.Empty(t, second.Queue(t).Pending)
 336 }
 337
 338 func TestInterruptedParentReportResumesAfterDaemonRestart(t *testing.T) {
 339         dir := t.TempDir()
 340         server := newModelFake(t)
 341         arrived := make(chan struct{})
 342         release := make(chan struct{})
 343         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 344                 if call.number == 1 {
 345                         close(arrived)
 346                         <-release
 347                 }
 348
 349                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan is ready.")}, 100, 0, 5))
 350         }
 351         first := newDaemonMessageTest(t, dir, server)
 352         first.Start(t, 0, 1)
 353         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
 354         <-arrived
 355         first.Close(t)
 356         close(release)
 357
 358         second := newDaemonMessageTest(t, dir, server)
 359         require.Equal(t, 1, second.agents[0].memory.Len())
 360         require.False(t, second.loops[1].waiting())
 361         second.Start(t, 0, 1)
 362         second.events[0].WaitFor(t, "idle", 1)
 363         require.Equal(t, 3, server.Calls())
 364         require.Equal(t, 2, second.agents[0].memory.Len())
 365         require.Empty(t, second.Queue(t).Pending)
 366
 367         texts := turnInputTexts(t, server.Call(2))
 368         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
 369 }
 370
 371 func TestAgentMessageWriteFailureStopsDelivery(t *testing.T) {
 372         server := newModelFake(t)
 373         test := newDaemonMessageTest(t, t.TempDir(), server)
 374         require.NoError(t, os.Mkdir(test.messages.path, 0o700))
 375         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."), "save agent messages")
 376         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Do not send again."), "save agent messages")
 377
 378         for _, loop := range test.loops {
 379                 require.ErrorContains(t, loop.failed(), "save agent messages")
 380                 require.Zero(t, loop.memory.Len())
 381         }
 382 }
 383
 384 func TestAgentMessageSnapshotBlocksDelivery(t *testing.T) {
 385         server := newModelFake(t)
 386         test := newDaemonMessageTest(t, t.TempDir(), server)
 387         entered := make(chan struct{})
 388         release := make(chan struct{})
 389
 390         var once sync.Once
 391
 392         defer once.Do(func() {
 393                 close(release)
 394         })
 395
 396         snapshot := make(chan error, 1)
 397         go func() {
 398                 snapshot <- test.messages.Snapshot(func() error {
 399                         return daemonSnapshot(t.Context(), test.loops, func() error {
 400                                 close(entered)
 401                                 <-release
 402
 403                                 return nil
 404                         })
 405                 })
 406         }()
 407
 408         <-entered
 409
 410         sent := make(chan error, 1)
 411         started := make(chan struct{})
 412         go func() {
 413                 close(started)
 414
 415                 sent <- test.messages.Tell("OneChat", "Garden", "Record the plan after the copy.")
 416         }()
 417
 418         <-started
 419         require.Never(t, func() bool {
 420                 return len(sent) > 0
 421         }, 50*time.Millisecond, time.Millisecond)
 422         once.Do(func() {
 423                 close(release)
 424         })
 425         require.NoError(t, <-snapshot)
 426         require.NoError(t, <-sent)
 427         require.Equal(t, 1, test.agents[1].memory.Len())
 428         require.Len(t, test.Queue(t).Pending, 1)
 429 }
 430
 431 func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) {
 432         server := newModelFake(t)
 433
 434         test := newDaemonMessageTest(t, t.TempDir(), server)
 435         testCases := []struct {
 436                 name   string
 437                 source string
 438                 target string
 439         }{
 440                 {
 441                         name:   "no sibling messages",
 442                         source: "Garden",
 443                         target: "Library",
 444                 },
 445                 {
 446                         name:   "no arbitrary child reports",
 447                         source: "Garden",
 448                         target: "OneChat",
 449                 },
 450                 {
 451                         name:   "no self messages",
 452                         source: "OneChat",
 453                         target: "OneChat",
 454                 },
 455                 {
 456                         name:   "unknown sender",
 457                         source: "Other",
 458                         target: "Garden",
 459                 },
 460                 {
 461                         name:   "unknown recipient",
 462                         source: "OneChat",
 463                         target: "Other",
 464                 },
 465         }
 466         for _, tc := range testCases {
 467                 t.Run(tc.name, func(t *testing.T) {
 468                         require.ErrorContains(t, test.messages.Tell(tc.source, tc.target, "Do not deliver."), "not a child")
 469                 })
 470         }
 471
 472         for _, agent := range test.agents {
 473                 require.Zero(t, agent.memory.Len())
 474         }
 475 }
 476
 477 func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) {
 478         server := newModelFake(t)
 479         testCases := []struct {
 480                 name string
 481                 file string
 482         }{
 483                 {
 484                         name: "broken JSON",
 485                         file: "{",
 486                 },
 487                 {
 488                         name: "unknown sender",
 489                         file: `{"next":1,"pending":[{"id":1,"source":"Other","target":"Garden","kind":"user"}]}`,
 490                 },
 491                 {
 492                         name: "sibling delivery",
 493                         file: `{"next":1,"pending":[{"id":1,"source":"Library","target":"Garden","kind":"user"}]}`,
 494                 },
 495                 {
 496                         name: "invalid sequence",
 497                         file: `{"next":0,"pending":[{"id":1,"source":"OneChat","target":"Garden","kind":"user"}]}`,
 498                 },
 499         }
 500         for _, tc := range testCases {
 501                 t.Run(tc.name, func(t *testing.T) {
 502                         dir := t.TempDir()
 503                         test := newDaemonMessageTest(t, dir, server)
 504                         path := test.messages.path
 505                         test.Close(t)
 506                         require.NoError(t, os.WriteFile(path, []byte(tc.file), 0o600))
 507
 508                         agents, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
 509                         require.NoError(t, err)
 510
 511                         defer func() {
 512                                 require.NoError(t, agents.Close())
 513                         }()
 514
 515                         var loops []*TurnLoop
 516
 517                         for _, agent := range agents {
 518                                 loop, err := agent.Loop(nil, func(Event) {
 519                                 })
 520                                 require.NoError(t, err)
 521
 522                                 loops = append(loops, loop)
 523                         }
 524
 525                         _, err = openDaemonMessages(loops)
 526                         require.Error(t, err)
 527
 528                         for _, agent := range agents {
 529                                 require.Zero(t, agent.memory.Len())
 530                         }
 531                 })
 532         }
 533 }
 534
 535 func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
 536         server := newModelFake(t)
 537         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 538                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 539         }
 540         test := newDaemonMessageTest(t, t.TempDir(), server)
 541         child := test.agents[1]
 542         at := child.clock().Add(time.Minute).Format(scheduleAtLayout)
 543         output := test.loops[1].tools.run(t.Context(), "schedule_set", `{"text":"Check the plants.","at":"`+at+`","cron":null}`)
 544         require.Contains(t, output, "Schedule 1 set.")
 545         require.Equal(t, "Garden", child.schedules.sorted()[0].Owner)
 546
 547         child.schedules.clock = func() time.Time {
 548                 return child.clock().Add(time.Minute)
 549         }
 550         require.ErrorContains(t, child.schedules.deliver(t.Context(), test.loops[0]), "does not match")
 551         require.NoError(t, child.schedules.deliver(t.Context(), test.loops[1]))
 552         test.Start(t, 0, 1, 2)
 553         test.events[1].WaitFor(t, "idle", 1)
 554         require.Equal(t, 1, server.Calls())
 555         require.Equal(t, child.agent.CacheKey(), server.Call(0).body["prompt_cache_key"])
 556         require.Zero(t, test.agents[0].memory.Len())
 557         require.Zero(t, test.agents[2].memory.Len())
 558         require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.")
 559         require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source)
 560 }
 561
 562 func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) {
 563         testCases := []struct {
 564                 name string
 565                 chat WhatsAppChat
 566         }{
 567                 {
 568                         name: "private chat",
 569                         chat: WhatsAppChat{
 570                                 ID:   whatsappTestAna.String(),
 571                                 Name: "Ana",
 572                         },
 573                 },
 574                 {
 575                         name: "group without a mention",
 576                         chat: WhatsAppChat{
 577                                 ID:    whatsappTestFamily.String(),
 578                                 Name:  "Garden plans",
 579                                 Group: true,
 580                         },
 581                 },
 582         }
 583         for _, tc := range testCases {
 584                 t.Run(tc.name, func(t *testing.T) {
 585                         server := newModelFake(t)
 586                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 587                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 588                         }
 589                         dir := t.TempDir()
 590                         test := newDaemonMessageTest(t, dir, server)
 591                         inbox := whatsappRouteTestInbox(t, dir)
 592                         require.NoError(t, test.messages.routes.Set("whatsapp:"+tc.chat.ID, "Garden"))
 593                         whatsappRouteTestAdd(t, inbox, tc.chat, "The gate needs a new latch.")
 594                         test.Start(t, 0, 1, 2)
 595                         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 596                         test.events[1].WaitFor(t, "idle", 1)
 597                         require.Equal(t, 1, server.Calls())
 598                         require.Equal(t, "gpt-test-garden", server.Call(0).body["model"])
 599                         require.Contains(t, string(server.Call(0).raw), "The gate needs a new latch.")
 600                         require.Zero(t, test.agents[0].memory.Len())
 601                         require.Empty(t, test.events[0].Kinds())
 602                         require.Zero(t, test.agents[2].memory.Len())
 603                         require.False(t, test.loops[0].waiting())
 604                         require.Equal(t, whatsappSource, test.events[1].LastEvent("work").Source)
 605                         require.Empty(t, test.events[1].LastEvent("work").Status)
 606                         require.Equal(t, 1, test.agents[1].memory.Len())
 607                         require.Empty(t, inbox.state.Closed())
 608                         require.Empty(t, test.Queue(t).Pending)
 609                         require.Contains(t, test.logs.String(), "batch routed name=Garden kind=whatsapp id=1 to=Garden")
 610                         require.NotContains(t, test.logs.String(), tc.chat.ID)
 611                         require.NotContains(t, test.logs.String(), "new latch")
 612                 })
 613         }
 614 }
 615
 616 func TestWhatsAppUnrouteRestoresRootDelivery(t *testing.T) {
 617         dir := t.TempDir()
 618         test := newDaemonMessageTest(t, dir, newModelFake(t))
 619         inbox := whatsappRouteTestInbox(t, dir)
 620         chat := WhatsAppChat{
 621                 ID:   whatsappTestAna.String(),
 622                 Name: "Ana",
 623         }
 624         source := "whatsapp:" + chat.ID
 625         require.NoError(t, test.messages.routes.Set(source, "Garden"))
 626         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
 627         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 628         require.Zero(t, test.agents[0].memory.Len())
 629         require.True(t, test.loops[1].waiting())
 630         require.NoError(t, test.messages.routes.Set(source, ""))
 631         whatsappRouteTestAdd(t, inbox, chat, "Please check the other plan.")
 632         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 633         require.Equal(t, 1, test.agents[0].memory.Len())
 634         require.Equal(t, 1, test.agents[1].memory.Len())
 635         require.True(t, test.loops[0].waiting())
 636         require.Contains(t, test.events[0].Last("work"), "Please check the other plan.")
 637         require.NotContains(t, test.events[0].Last("work"), "Keep the gate plan.")
 638         require.Empty(t, inbox.state.Closed())
 639 }
 640
 641 func TestMissingWhatsAppRouteChildFallsBackToRoot(t *testing.T) {
 642         dir := t.TempDir()
 643         test := newDaemonMessageTest(t, dir, newModelFake(t))
 644         inbox := whatsappRouteTestInbox(t, dir)
 645         chat := WhatsAppChat{
 646                 ID:   whatsappTestAna.String(),
 647                 Name: "Ana",
 648         }
 649         require.NoError(t, test.messages.routes.Set("whatsapp:"+chat.ID, "Missing"))
 650         whatsappRouteTestAdd(t, inbox, chat, "Please check the plan.")
 651         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 652         require.Equal(t, 1, test.agents[0].memory.Len())
 653         require.True(t, test.loops[0].waiting())
 654         require.Zero(t, test.agents[1].memory.Len())
 655         require.False(t, test.loops[1].waiting())
 656         require.Empty(t, test.events[0].LastEvent("work").Status)
 657         require.Equal(t, whatsappSource, test.events[0].LastEvent("work").Source)
 658         require.Empty(t, inbox.state.Closed())
 659 }
 660
 661 func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) {
 662         testCases := []struct {
 663                 name   string
 664                 finish bool
 665         }{
 666                 {
 667                         name:   "queued turn resumes",
 668                         finish: false,
 669                 },
 670                 {
 671                         name:   "finished turn stays finished",
 672                         finish: true,
 673                 },
 674         }
 675         for _, tc := range testCases {
 676                 t.Run(tc.name, func(t *testing.T) {
 677                         dir := t.TempDir()
 678                         server := newModelFake(t)
 679                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 680                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 681                         }
 682                         first := newDaemonMessageTest(t, dir, server)
 683                         inbox := whatsappRouteTestInbox(t, dir)
 684                         chat := WhatsAppChat{
 685                                 ID:   whatsappTestAna.String(),
 686                                 Name: "Ana",
 687                         }
 688                         source := "whatsapp:" + chat.ID
 689                         require.NoError(t, first.messages.routes.Set(source, "Garden"))
 690                         whatsappRouteTestAdd(t, inbox, chat, "Remember the gate plan.")
 691                         require.NoError(t, inbox.state.CloseAll())
 692
 693                         batch := inbox.state.Closed()[0]
 694                         routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
 695                         require.NoError(t, err)
 696                         require.True(t, routed)
 697
 698                         if tc.finish {
 699                                 first.Start(t, 1)
 700                                 first.events[1].WaitFor(t, "idle", 1)
 701                                 require.Empty(t, first.Queue(t).Pending)
 702                         }
 703
 704                         first.Close(t)
 705
 706                         second := newDaemonMessageTest(t, dir, server)
 707                         require.Equal(t, source+" -> Garden", second.messages.routes.List())
 708                         require.Equal(t, 1, second.agents[1].memory.Len())
 709                         require.Equal(t, !tc.finish, second.loops[1].waiting())
 710                         require.NoError(t, second.messages.routes.Set(source, ""))
 711
 712                         restored := whatsappRouteTestInbox(t, dir)
 713                         require.Len(t, restored.state.Closed(), 1)
 714                         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
 715                         require.Empty(t, restored.state.Closed())
 716                         require.Zero(t, second.agents[0].memory.Len())
 717                         require.Equal(t, 1, second.agents[1].memory.Len())
 718
 719                         if !tc.finish {
 720                                 second.Start(t, 1)
 721                                 second.events[1].WaitFor(t, "idle", 1)
 722                         }
 723
 724                         require.Equal(t, 1, server.Calls())
 725                         require.Empty(t, second.Queue(t).Pending)
 726                         second.Close(t)
 727
 728                         third := newDaemonMessageTest(t, dir, server)
 729                         require.False(t, third.loops[1].waiting())
 730                         require.Empty(t, third.events[1].Kinds())
 731                         require.Equal(t, 1, third.agents[1].memory.Len())
 732                 })
 733         }
 734 }
 735
 736 func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) {
 737         testCases := []struct {
 738                 name string
 739                 id   *int
 740         }{
 741                 {
 742                         name: "before log ID is stored",
 743                         id:   nil,
 744                 },
 745                 {
 746                         name: "after log ID is stored",
 747                         id:   new(0),
 748                 },
 749         }
 750         for _, tc := range testCases {
 751                 t.Run(tc.name, func(t *testing.T) {
 752                         dir := t.TempDir()
 753                         server := newModelFake(t)
 754                         first := newDaemonMessageTest(t, dir, server)
 755                         path := first.messages.path
 756                         first.Close(t)
 757
 758                         file := DaemonMessageFile{
 759                                 Next:            1,
 760                                 WhatsAppBatches: []int{1},
 761                                 Pending: []DaemonMessage{{
 762                                         ID:            1,
 763                                         Source:        "OneChat",
 764                                         Target:        "Garden",
 765                                         Kind:          "work",
 766                                         Text:          "[WhatsApp] Keep the gate plan.",
 767                                         LogID:         tc.id,
 768                                         WhatsAppBatch: 1,
 769                                 }},
 770                         }
 771                         data, err := json.Marshal(file)
 772                         require.NoError(t, err)
 773                         require.NoError(t, os.WriteFile(path, data, 0o600))
 774
 775                         second := newDaemonMessageTest(t, dir, server)
 776                         require.Equal(t, 1, second.agents[1].memory.Len())
 777                         require.True(t, second.loops[1].waiting())
 778                         require.Equal(t, "[WhatsApp] Keep the gate plan.", second.events[1].Last("work"))
 779                         require.Equal(t, whatsappSource, second.events[1].LastEvent("work").Source)
 780                         second.Close(t)
 781
 782                         third := newDaemonMessageTest(t, dir, server)
 783                         require.Equal(t, 1, third.agents[1].memory.Len())
 784                         require.True(t, third.loops[1].waiting())
 785                         require.Zero(t, third.agents[0].memory.Len())
 786                 })
 787         }
 788 }
 789
 790 func TestQueuedWhatsAppBatchUsesRootIfItsChildIsRemoved(t *testing.T) {
 791         dir := t.TempDir()
 792         server := newModelFake(t)
 793         first := newDaemonMessageTest(t, dir, server)
 794         inbox := whatsappRouteTestInbox(t, dir)
 795         chat := WhatsAppChat{
 796                 ID:   whatsappTestAna.String(),
 797                 Name: "Ana",
 798         }
 799         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
 800         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
 801         require.NoError(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true))
 802         first.Close(t)
 803         require.NoError(t, os.RemoveAll(filepath.Join(dir, daemonAgentDirName, "children", "Garden")))
 804
 805         second := newDaemonMessageTest(t, dir, server)
 806         require.Equal(t, 1, second.agents[0].memory.Len())
 807         require.True(t, second.loops[0].waiting())
 808         require.Equal(t, whatsappSource, second.events[0].LastEvent("work").Source)
 809         require.Empty(t, second.events[0].LastEvent("work").Status)
 810         require.Contains(t, second.events[0].Last("work"), "Keep the gate plan.")
 811         require.False(t, second.loops[1].waiting())
 812         second.Close(t)
 813
 814         third := newDaemonMessageTest(t, dir, server)
 815         require.Equal(t, 1, third.agents[0].memory.Len())
 816         require.True(t, third.loops[0].waiting())
 817 }
 818
 819 func TestRoutedWhatsAppBatchesCanCloseOutOfNumberOrder(t *testing.T) {
 820         dir := t.TempDir()
 821         test := newDaemonMessageTest(t, dir, newModelFake(t))
 822         inbox := whatsappRouteTestInbox(t, dir)
 823         first := WhatsAppChat{
 824                 ID:   whatsappTestAna.String(),
 825                 Name: "Ana",
 826         }
 827         second := WhatsAppChat{
 828                 ID:    whatsappTestFamily.String(),
 829                 Name:  "Garden plans",
 830                 Group: true,
 831         }
 832         require.NoError(t, test.messages.routes.Set("whatsapp:"+first.ID, "Garden"))
 833         require.NoError(t, test.messages.routes.Set("whatsapp:"+second.ID, "Garden"))
 834
 835         now := time.Now()
 836         _, err := inbox.state.Add(first, WhatsAppMessage{
 837                 ID:     "first",
 838                 Text:   "The first chat is still open.",
 839                 Sender: "Ana",
 840                 Time:   now,
 841         }, now)
 842         require.NoError(t, err)
 843
 844         _, err = inbox.state.Add(second, WhatsAppMessage{
 845                 ID:     "second",
 846                 Text:   "The second chat is quiet.",
 847                 Sender: "Bruno",
 848                 Time:   now,
 849         }, now.Add(-time.Minute))
 850         require.NoError(t, err)
 851         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
 852         require.Equal(t, 1, test.agents[1].memory.Len())
 853         require.Contains(t, test.events[1].Last("work"), "The second chat is quiet.")
 854         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
 855         require.Equal(t, 2, test.agents[1].memory.Len())
 856         require.Contains(t, test.events[1].Last("work"), "The first chat is still open.")
 857         require.Zero(t, test.agents[0].memory.Len())
 858         require.Empty(t, test.Queue(t).WhatsAppBatches)
 859 }
 860
 861 func TestWhatsAppCatchupSeparatesChildAndRootBatches(t *testing.T) {
 862         dir := t.TempDir()
 863         test := newDaemonMessageTest(t, dir, newModelFake(t))
 864         inbox := whatsappRouteTestInbox(t, dir)
 865         garden := WhatsAppChat{
 866                 ID:    whatsappTestFamily.String(),
 867                 Name:  "Garden plans",
 868                 Group: true,
 869         }
 870         library := WhatsAppChat{
 871                 ID:   whatsappTestBruno.String(),
 872                 Name: "Bruno",
 873         }
 874         root := WhatsAppChat{
 875                 ID:   whatsappTestAna.String(),
 876                 Name: "Ana",
 877         }
 878         require.NoError(t, test.messages.routes.Set("whatsapp:"+garden.ID, "Garden"))
 879         require.NoError(t, test.messages.routes.Set("whatsapp:"+library.ID, "Library"))
 880         inbox.waitForCatchup()
 881         whatsappRouteTestAdd(t, inbox, garden, "Check the gate.")
 882         whatsappRouteTestAdd(t, inbox, library, "Check the book.")
 883         whatsappRouteTestAdd(t, inbox, root, "Check the mixed chat.")
 884         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
 885
 886         for _, agent := range test.agents {
 887                 require.Zero(t, agent.memory.Len())
 888         }
 889
 890         inbox.endCatchup()
 891         require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
 892
 893         for _, loop := range test.loops {
 894                 require.Equal(t, 1, loop.memory.Len())
 895                 require.True(t, loop.waiting())
 896         }
 897
 898         require.Contains(t, test.events[0].Last("work"), "Check the mixed chat.")
 899         require.Contains(t, test.events[1].Last("work"), "Check the gate.")
 900         require.Contains(t, test.events[2].Last("work"), "Check the book.")
 901         require.Empty(t, inbox.state.Closed())
 902         require.Empty(t, test.Queue(t).WhatsAppBatches)
 903 }
 904
 905 func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) {
 906         dir := t.TempDir()
 907         server := newModelFake(t)
 908         first := newDaemonMessageTest(t, dir, server)
 909         inbox := whatsappRouteTestInbox(t, dir)
 910         chat := WhatsAppChat{
 911                 ID:   whatsappTestAna.String(),
 912                 Name: "Ana",
 913         }
 914         require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
 915         require.NoError(t, os.Mkdir(first.messages.path, 0o700))
 916         whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
 917         require.ErrorContains(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true), "save agent messages")
 918         require.Len(t, inbox.state.Closed(), 1)
 919
 920         for _, loop := range first.loops {
 921                 require.ErrorContains(t, loop.failed(), "save agent messages")
 922                 require.Zero(t, loop.memory.Len())
 923         }
 924
 925         path := first.messages.path
 926         first.Close(t)
 927         require.NoError(t, os.Remove(path))
 928
 929         second := newDaemonMessageTest(t, dir, server)
 930         restored := whatsappRouteTestInbox(t, dir)
 931         require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
 932         require.Equal(t, 1, second.agents[1].memory.Len())
 933         require.True(t, second.loops[1].waiting())
 934         require.Zero(t, second.agents[0].memory.Len())
 935         require.Empty(t, restored.state.Closed())
 936 }
 937
 938 func whatsappRouteTestInbox(t *testing.T, dir string) *WhatsAppInbox {
 939         t.Helper()
 940
 941         state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName))
 942         require.NoError(t, err)
 943
 944         return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()), &ImageStore{dir: filepath.Join(dir, imageDirName)}, discardLog())
 945 }
 946
 947 func whatsappRouteTestAdd(t *testing.T, inbox *WhatsAppInbox, chat WhatsAppChat, text string) {
 948         t.Helper()
 949
 950         at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
 951         _, err := inbox.state.Add(chat, WhatsAppMessage{
 952                 ID:     text,
 953                 Sender: "Ana",
 954                 Text:   text,
 955                 Time:   at,
 956         }, at)
 957         require.NoError(t, err)
 958 }