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

   1 package main
   2
   3 import (
   4         "context"
   5         "errors"
   6         "net/http"
   7         "os"
   8         "path/filepath"
   9         "sync"
  10         "testing"
  11         "time"
  12
  13         "github.com/stretchr/testify/require"
  14 )
  15
  16 func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
  17         server := newModelFake(t)
  18         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
  19                 text := "I will ask the user."
  20                 if call.body["model"] == "gpt-test-garden" {
  21                         text = "Approve the gate size."
  22                 }
  23
  24                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
  25         }
  26         test := newDaemonMessageTest(t, t.TempDir(), server)
  27         test.Start(t, 0, 1, 2)
  28         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
  29         test.events[0].WaitFor(t, "idle", 1)
  30         test.events[1].WaitFor(t, "idle", 1)
  31
  32         require.Equal(t, "OneChat", test.events[1].LastEvent("user").Source)
  33         require.Equal(t, "Check the gate size.", test.events[1].Last("user"))
  34         require.Equal(t, "Garden", test.events[0].LastEvent("work").Source)
  35         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
  36         require.Equal(t, []string{"user", "busy", "reply", "idle"}, test.events[1].Kinds())
  37         require.Equal(t, []string{"work", "busy", "reply", "idle"}, test.events[0].Kinds())
  38         require.Equal(t, 2, server.Calls())
  39         require.Zero(t, test.agents[2].memory.Len())
  40         require.Empty(t, test.events[2].Kinds())
  41
  42         first, err := test.agents[1].memory.Message(0)
  43         require.NoError(t, err)
  44         require.Equal(t, "user", first.Kind)
  45
  46         report, err := test.agents[0].memory.Message(0)
  47         require.NoError(t, err)
  48         require.Equal(t, "work", report.Kind)
  49         require.Equal(t, "[Garden] Approve the gate size.", report.Text)
  50         require.Contains(t, test.logs.String(), "turn started name=Garden")
  51         require.Contains(t, test.logs.String(), "turn done name=Garden")
  52         require.NotContains(t, test.logs.String(), "gate size")
  53         require.Empty(t, test.Queue(t).Pending)
  54
  55         info, err := os.Stat(test.messages.path)
  56         require.NoError(t, err)
  57         require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
  58 }
  59
  60 func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) {
  61         server := newModelFake(t)
  62         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
  63                 text := "I will ask the user."
  64                 if call.body["model"] == "gpt-test-garden" {
  65                         text = "Approve the gate size."
  66                 }
  67
  68                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
  69         }
  70         test := newDaemonMessageTest(t, t.TempDir(), server)
  71         telegram := newTelegramFake(t)
  72         bot := newTelegramBot(newTelegramAPI(telegram.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), discardLog())
  73         test.loops[0].notify = func(event Event) {
  74                 test.events[0].Add(event)
  75                 bot.Notify(event)
  76         }
  77
  78         ctx, cancel := context.WithCancel(t.Context())
  79         defer cancel()
  80
  81         done := make(chan struct{})
  82         go func() {
  83                 defer close(done)
  84
  85                 bot.send(ctx)
  86         }()
  87
  88         t.Cleanup(func() {
  89                 cancel()
  90                 <-done
  91         })
  92
  93         test.Start(t, 0, 1)
  94         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
  95         test.events[0].WaitFor(t, "idle", 1)
  96         require.Eventually(t, func() bool {
  97                 return len(telegram.Sent()) >= 2
  98         }, 10*time.Second, time.Millisecond)
  99         require.Equal(t, []string{"[Garden] finished", "I will ask the user."}, telegram.Sent())
 100         require.Equal(t, subagentFinished, test.events[0].LastEvent("work").Status)
 101         require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
 102 }
 103
 104 func TestQueuedAgentMessageFailureKeepsItsOperationAndCause(t *testing.T) {
 105         server := newModelFake(t)
 106         test := newDaemonMessageTest(t, t.TempDir(), server)
 107         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 108
 109         failure := errors.New("fake queued message failure")
 110         test.loops[1].fail(failure)
 111
 112         err := test.loops[1].Run(t.Context())
 113         require.ErrorIs(t, err, failure)
 114         require.ErrorContains(t, err, "check turn loop failure")
 115         require.Zero(t, server.Calls())
 116 }
 117
 118 func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.T) {
 119         testCases := []struct {
 120                 name string
 121                 text string
 122         }{
 123                 {
 124                         name: "empty final reply",
 125                         text: "",
 126                 },
 127                 {
 128                         name: "whitespace final reply",
 129                         text: " \n\t ",
 130                 },
 131         }
 132         for _, tc := range testCases {
 133                 t.Run(tc.name, func(t *testing.T) {
 134                         server := newModelFake(t)
 135                         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 136                                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(tc.text)}, 100, 0, 5))
 137                         }
 138                         test := newDaemonMessageTest(t, t.TempDir(), server)
 139                         test.Start(t, 0, 1, 2)
 140                         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
 141                         test.events[1].WaitFor(t, "idle", 1)
 142                         require.Equal(t, 1, server.Calls())
 143                         require.Zero(t, test.agents[0].memory.Len())
 144                         require.Empty(t, test.events[0].Kinds())
 145                         require.False(t, test.loops[0].waiting())
 146                         require.Empty(t, test.events[2].Kinds())
 147                         require.Empty(t, test.Queue(t).Pending)
 148                 })
 149         }
 150 }
 151
 152 func TestBusyChildQueuesTheNextParentMessage(t *testing.T) {
 153         server := newModelFake(t)
 154         arrived := make(chan struct{})
 155         release := make(chan struct{})
 156         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 157                 if call.number == 0 {
 158                         close(arrived)
 159                         <-release
 160                 }
 161
 162                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 163         }
 164         test := newDaemonMessageTest(t, t.TempDir(), server)
 165         test.Start(t, 0, 1)
 166         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the first plan."))
 167         <-arrived
 168         require.True(t, test.loops[1].Busy())
 169         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the second plan too."))
 170         require.Equal(t, 1, server.Calls())
 171         require.Len(t, test.Queue(t).Pending, 2)
 172         close(release)
 173         test.events[1].WaitFor(t, "idle", 2)
 174         require.Equal(t, 2, server.Calls())
 175
 176         texts := turnInputTexts(t, server.Call(1))
 177         require.Equal(t, "Check the second plan too.", texts[len(texts)-1])
 178         require.Empty(t, test.Queue(t).Pending)
 179 }
 180
 181 func TestBusyParentQueuesItsChildReport(t *testing.T) {
 182         server := newModelFake(t)
 183         arrived := make(chan struct{})
 184         release := make(chan struct{})
 185         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 186                 text := "The plan is ready."
 187                 if call.number == 0 {
 188                         close(arrived)
 189                         <-release
 190
 191                         text = "I will continue."
 192                 }
 193
 194                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
 195         }
 196         test := newDaemonMessageTest(t, t.TempDir(), server)
 197         test.Start(t, 0, 1)
 198
 199         _, err := test.loops[0].Submit("terminal", "Start the work.")
 200         require.NoError(t, err)
 201         <-arrived
 202         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Prepare the plan."))
 203         test.events[1].WaitFor(t, "idle", 1)
 204         require.Equal(t, "[Garden] The plan is ready.", test.events[0].Last("work"))
 205         require.True(t, test.loops[0].Busy())
 206         require.Equal(t, 2, server.Calls())
 207         close(release)
 208         test.events[0].WaitFor(t, "idle", 2)
 209         require.Equal(t, 3, server.Calls())
 210
 211         texts := turnInputTexts(t, server.Call(2))
 212         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
 213         require.Empty(t, test.Queue(t).Pending)
 214 }
 215
 216 func TestChildReportsOnlyTheFinalReplyAndTakesQueuedToolInput(t *testing.T) {
 217         server := newModelFake(t)
 218         arrived := make(chan struct{})
 219         release := make(chan struct{})
 220         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 221                 if call.number == 0 {
 222                         close(arrived)
 223                         <-release
 224                         modelWriteEvents(w, modelCompletedEvent([]any{
 225                                 modelMessageItem("I will check the plan."),
 226                                 modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`),
 227                         }, 100, 0, 5))
 228
 229                         return
 230                 }
 231
 232                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan needs approval.")}, 100, 0, 5))
 233         }
 234         test := newDaemonMessageTest(t, t.TempDir(), server)
 235         test.Start(t, 1)
 236         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the plan."))
 237         <-arrived
 238         require.NoError(t, test.messages.Tell("OneChat", "Garden", "Also check the date."))
 239         close(release)
 240         test.events[1].WaitFor(t, "idle", 1)
 241         require.Equal(t, 2, server.Calls())
 242         require.Equal(t, 1, test.events[1].Count("busy"))
 243         require.Equal(t, 1, test.events[0].Count("work"))
 244         require.Equal(t, "[Garden] The plan needs approval.", test.events[0].Last("work"))
 245         require.NotContains(t, test.events[0].Last("work"), "I will check")
 246
 247         texts := turnInputTexts(t, server.Call(1))
 248         require.Equal(t, "Also check the date.", texts[len(texts)-1])
 249
 250         queue := test.Queue(t)
 251         require.Len(t, queue.Pending, 1)
 252         require.Equal(t, "work", queue.Pending[0].Kind)
 253 }
 254
 255 func TestQueuedParentMessageSurvivesDaemonRestart(t *testing.T) {
 256         dir := t.TempDir()
 257         server := newModelFake(t)
 258         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 259                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 260         }
 261         first := newDaemonMessageTest(t, dir, server)
 262         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Remember the garden plan."))
 263         require.Equal(t, 1, first.agents[1].memory.Len())
 264         first.Close(t)
 265
 266         second := newDaemonMessageTest(t, dir, server)
 267         require.Equal(t, 1, second.agents[1].memory.Len())
 268         require.Equal(t, "OneChat", second.events[1].LastEvent("user").Source)
 269         second.Start(t, 0, 1)
 270         second.events[1].WaitFor(t, "idle", 1)
 271         require.Equal(t, 1, server.Calls())
 272         require.Empty(t, second.Queue(t).Pending)
 273         second.Close(t)
 274
 275         third := newDaemonMessageTest(t, dir, server)
 276         require.Equal(t, 1, third.agents[1].memory.Len())
 277         require.False(t, third.loops[1].waiting())
 278         require.Empty(t, third.events[1].Kinds())
 279 }
 280
 281 func TestQueuedChildReportSurvivesDaemonRestart(t *testing.T) {
 282         dir := t.TempDir()
 283         server := newModelFake(t)
 284         first := newDaemonMessageTest(t, dir, server)
 285         first.Start(t, 1)
 286         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
 287         first.events[1].WaitFor(t, "idle", 1)
 288         require.Equal(t, "[Garden] a line", first.events[0].Last("work"))
 289         require.Len(t, first.Queue(t).Pending, 1)
 290         first.Close(t)
 291
 292         second := newDaemonMessageTest(t, dir, server)
 293         require.Equal(t, 1, second.agents[0].memory.Len())
 294         require.False(t, second.loops[1].waiting())
 295         second.Start(t, 0, 1)
 296         second.events[0].WaitFor(t, "idle", 1)
 297         require.Equal(t, 2, server.Calls())
 298         require.Equal(t, 2, second.agents[0].memory.Len())
 299         require.Empty(t, second.Queue(t).Pending)
 300         second.Close(t)
 301
 302         third := newDaemonMessageTest(t, dir, server)
 303         require.Equal(t, 2, third.agents[0].memory.Len())
 304         require.False(t, third.loops[0].waiting())
 305         require.Empty(t, third.events[0].Kinds())
 306 }
 307
 308 func TestInterruptedChildDeliveryResumesAfterDaemonRestart(t *testing.T) {
 309         dir := t.TempDir()
 310         server := newModelFake(t)
 311         arrived := make(chan struct{})
 312         release := make(chan struct{})
 313         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 314                 if call.number == 0 {
 315                         close(arrived)
 316                         <-release
 317                 }
 318
 319                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 320         }
 321         first := newDaemonMessageTest(t, dir, server)
 322         first.Start(t, 1)
 323         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Finish the garden plan."))
 324         <-arrived
 325         first.Close(t)
 326         close(release)
 327
 328         second := newDaemonMessageTest(t, dir, server)
 329         require.Equal(t, 1, second.agents[1].memory.Len())
 330         second.Start(t, 1)
 331         second.events[1].WaitFor(t, "idle", 1)
 332         require.Equal(t, 2, server.Calls())
 333         require.Equal(t, 1, second.agents[1].memory.Len())
 334         require.Empty(t, second.Queue(t).Pending)
 335 }
 336
 337 func TestInterruptedParentReportResumesAfterDaemonRestart(t *testing.T) {
 338         dir := t.TempDir()
 339         server := newModelFake(t)
 340         arrived := make(chan struct{})
 341         release := make(chan struct{})
 342         server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
 343                 if call.number == 1 {
 344                         close(arrived)
 345                         <-release
 346                 }
 347
 348                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan is ready.")}, 100, 0, 5))
 349         }
 350         first := newDaemonMessageTest(t, dir, server)
 351         first.Start(t, 0, 1)
 352         require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
 353         <-arrived
 354         first.Close(t)
 355         close(release)
 356
 357         second := newDaemonMessageTest(t, dir, server)
 358         require.Equal(t, 1, second.agents[0].memory.Len())
 359         require.False(t, second.loops[1].waiting())
 360         second.Start(t, 0, 1)
 361         second.events[0].WaitFor(t, "idle", 1)
 362         require.Equal(t, 3, server.Calls())
 363         require.Equal(t, 2, second.agents[0].memory.Len())
 364         require.Empty(t, second.Queue(t).Pending)
 365
 366         texts := turnInputTexts(t, server.Call(2))
 367         require.Equal(t, "[Garden] The plan is ready.", texts[len(texts)-1])
 368 }
 369
 370 func TestAgentMessageWriteFailureStopsDelivery(t *testing.T) {
 371         server := newModelFake(t)
 372         test := newDaemonMessageTest(t, t.TempDir(), server)
 373         require.NoError(t, os.Mkdir(test.messages.path, 0o700))
 374         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."), "save agent messages")
 375         require.ErrorContains(t, test.messages.Tell("OneChat", "Garden", "Do not send again."), "save agent messages")
 376
 377         for _, loop := range test.loops {
 378                 require.ErrorContains(t, loop.failed(), "save agent messages")
 379                 require.Zero(t, loop.memory.Len())
 380         }
 381 }
 382
 383 func TestAgentMessageSnapshotBlocksDelivery(t *testing.T) {
 384         server := newModelFake(t)
 385         test := newDaemonMessageTest(t, t.TempDir(), server)
 386         entered := make(chan struct{})
 387         release := make(chan struct{})
 388
 389         var once sync.Once
 390
 391         defer once.Do(func() {
 392                 close(release)
 393         })
 394
 395         snapshot := make(chan error, 1)
 396         go func() {
 397                 snapshot <- test.messages.Snapshot(func() error {
 398                         return daemonSnapshot(t.Context(), test.loops, func() error {
 399                                 close(entered)
 400                                 <-release
 401
 402                                 return nil
 403                         })
 404                 })
 405         }()
 406
 407         <-entered
 408
 409         sent := make(chan error, 1)
 410         started := make(chan struct{})
 411         go func() {
 412                 close(started)
 413
 414                 sent <- test.messages.Tell("OneChat", "Garden", "Record the plan after the copy.")
 415         }()
 416
 417         <-started
 418         require.Never(t, func() bool {
 419                 return len(sent) > 0
 420         }, 50*time.Millisecond, time.Millisecond)
 421         once.Do(func() {
 422                 close(release)
 423         })
 424         require.NoError(t, <-snapshot)
 425         require.NoError(t, <-sent)
 426         require.Equal(t, 1, test.agents[1].memory.Len())
 427         require.Len(t, test.Queue(t).Pending, 1)
 428 }
 429
 430 func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) {
 431         server := newModelFake(t)
 432
 433         test := newDaemonMessageTest(t, t.TempDir(), server)
 434         testCases := []struct {
 435                 name   string
 436                 source string
 437                 target string
 438         }{
 439                 {
 440                         name:   "no sibling messages",
 441                         source: "Garden",
 442                         target: "Library",
 443                 },
 444                 {
 445                         name:   "no arbitrary child reports",
 446                         source: "Garden",
 447                         target: "OneChat",
 448                 },
 449                 {
 450                         name:   "no self messages",
 451                         source: "OneChat",
 452                         target: "OneChat",
 453                 },
 454                 {
 455                         name:   "unknown sender",
 456                         source: "Other",
 457                         target: "Garden",
 458                 },
 459                 {
 460                         name:   "unknown recipient",
 461                         source: "OneChat",
 462                         target: "Other",
 463                 },
 464         }
 465         for _, tc := range testCases {
 466                 t.Run(tc.name, func(t *testing.T) {
 467                         require.ErrorContains(t, test.messages.Tell(tc.source, tc.target, "Do not deliver."), "not a child")
 468                 })
 469         }
 470
 471         for _, agent := range test.agents {
 472                 require.Zero(t, agent.memory.Len())
 473         }
 474 }
 475
 476 func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) {
 477         server := newModelFake(t)
 478         testCases := []struct {
 479                 name string
 480                 file string
 481         }{
 482                 {
 483                         name: "broken JSON",
 484                         file: "{",
 485                 },
 486                 {
 487                         name: "unknown sender",
 488                         file: `{"next":1,"pending":[{"id":1,"source":"Other","target":"Garden","kind":"user"}]}`,
 489                 },
 490                 {
 491                         name: "sibling delivery",
 492                         file: `{"next":1,"pending":[{"id":1,"source":"Library","target":"Garden","kind":"user"}]}`,
 493                 },
 494                 {
 495                         name: "invalid sequence",
 496                         file: `{"next":0,"pending":[{"id":1,"source":"OneChat","target":"Garden","kind":"user"}]}`,
 497                 },
 498         }
 499         for _, tc := range testCases {
 500                 t.Run(tc.name, func(t *testing.T) {
 501                         dir := t.TempDir()
 502                         test := newDaemonMessageTest(t, dir, server)
 503                         path := test.messages.path
 504                         test.Close(t)
 505                         require.NoError(t, os.WriteFile(path, []byte(tc.file), 0o600))
 506
 507                         agents, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
 508                         require.NoError(t, err)
 509
 510                         defer func() {
 511                                 require.NoError(t, agents.Close())
 512                         }()
 513
 514                         var loops []*TurnLoop
 515
 516                         for _, agent := range agents {
 517                                 loop, err := agent.Loop(nil, func(Event) {
 518                                 })
 519                                 require.NoError(t, err)
 520
 521                                 loops = append(loops, loop)
 522                         }
 523
 524                         _, err = openDaemonMessages(loops)
 525                         require.Error(t, err)
 526
 527                         for _, agent := range agents {
 528                                 require.Zero(t, agent.memory.Len())
 529                         }
 530                 })
 531         }
 532 }
 533
 534 func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
 535         server := newModelFake(t)
 536         server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
 537                 modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
 538         }
 539         test := newDaemonMessageTest(t, t.TempDir(), server)
 540         child := test.agents[1]
 541         at := child.clock().Add(time.Minute).Format(scheduleAtLayout)
 542         output := test.loops[1].tools.run(t.Context(), "schedule_set", `{"text":"Check the plants.","at":"`+at+`","cron":null}`)
 543         require.Contains(t, output, "Schedule 1 set.")
 544         require.Equal(t, "Garden", child.schedules.sorted()[0].Owner)
 545
 546         child.schedules.clock = func() time.Time {
 547                 return child.clock().Add(time.Minute)
 548         }
 549         require.ErrorContains(t, child.schedules.deliver(t.Context(), test.loops[0]), "does not match")
 550         require.NoError(t, child.schedules.deliver(t.Context(), test.loops[1]))
 551         test.Start(t, 0, 1, 2)
 552         test.events[1].WaitFor(t, "idle", 1)
 553         require.Equal(t, 1, server.Calls())
 554         require.Equal(t, child.agent.CacheKey(), server.Call(0).body["prompt_cache_key"])
 555         require.Zero(t, test.agents[0].memory.Len())
 556         require.Zero(t, test.agents[2].memory.Len())
 557         require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.")
 558         require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source)
 559 }