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 // DaemonMessageTest owns test agents and the loops selected by Start.
 563 type DaemonMessageTest struct {
 564         agents   DaemonAgents
 565         loops    []*TurnLoop
 566         events   []*TurnEventRecorder
 567         messages *DaemonMessages
 568         logs     *SyncBuffer
 569         cancel   context.CancelFunc
 570         done     []chan error
 571 }
 572
 573 func (r *DaemonMessageTest) Start(t *testing.T, indices ...int) {
 574         t.Helper()
 575
 576         ctx, cancel := context.WithCancel(t.Context())
 577         r.cancel = cancel
 578         for _, index := range indices {
 579                 done := make(chan error, 1)
 580                 r.done = append(r.done, done)
 581                 go func() {
 582                         done <- r.loops[index].Run(ctx)
 583                 }()
 584         }
 585 }
 586
 587 func (r *DaemonMessageTest) Queue(t *testing.T) DaemonMessageFile {
 588         t.Helper()
 589
 590         data, err := os.ReadFile(r.messages.path)
 591         require.NoError(t, err)
 592
 593         var file DaemonMessageFile
 594
 595         require.NoError(t, json.Unmarshal(data, &file))
 596
 597         return file
 598 }
 599
 600 func (r *DaemonMessageTest) Close(t *testing.T) {
 601         t.Helper()
 602
 603         if r.cancel != nil {
 604                 r.cancel()
 605
 606                 for _, done := range r.done {
 607                         require.ErrorIs(t, <-done, context.Canceled)
 608                 }
 609
 610                 r.cancel = nil
 611                 r.done = nil
 612         }
 613
 614         require.NoError(t, r.agents.Close())
 615
 616         r.agents = nil
 617 }
 618
 619 func newDaemonMessageTest(t *testing.T, dir string, server *ModelFake) *DaemonMessageTest {
 620         t.Helper()
 621
 622         _, err := os.Stat(filepath.Join(dir, daemonAgentDirName, "children"))
 623         if errors.Is(err, os.ErrNotExist) {
 624                 daemonChildConfig(t, dir, "Garden")
 625                 daemonChildConfig(t, dir, "Library")
 626         } else {
 627                 require.NoError(t, err)
 628         }
 629
 630         test := &DaemonMessageTest{
 631                 logs: &SyncBuffer{},
 632         }
 633         test.agents, err = openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, daemonTestLogger(test.logs))
 634         require.NoError(t, err)
 635         t.Cleanup(func() {
 636                 test.Close(t)
 637         })
 638
 639         for _, agent := range test.agents {
 640                 events := &TurnEventRecorder{}
 641                 loop, err := agent.Loop(nil, events.Add)
 642                 require.NoError(t, err)
 643
 644                 test.events = append(test.events, events)
 645                 test.loops = append(test.loops, loop)
 646         }
 647
 648         test.messages, err = openDaemonMessages(test.loops)
 649         require.NoError(t, err)
 650
 651         return test
 652 }