"context"
"encoding/json"
"errors"
+ "fmt"
"net/http"
"os"
"path/filepath"
+ "strings"
"sync"
"testing"
"time"
"github.com/stretchr/testify/require"
)
+func TestNotificationWaitsForTheNextTurnAfterEveryActiveToolRequest(t *testing.T) {
+ server := newModelFake(t)
+ arrived := make(chan struct{})
+ release := make(chan struct{})
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ if call.number == 0 {
+ close(arrived)
+ <-release
+ modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`)}, 100, 0, 5))
+
+ return
+ }
+
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.Start(t, 0, 1, 2)
+
+ _, err := test.loops[0].Submit(terminalSource, "Start a tool chain.")
+ require.NoError(t, err)
+ <-arrived
+
+ before := test.agents[0].memory.Len()
+ input := &NotifyInput{
+ Source: "test-check",
+ ID: "event-1",
+ Text: "Unique queued external data.",
+ }
+
+ receipt, err := test.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, "accepted", receipt.Status)
+ require.Equal(t, before, test.agents[0].memory.Len())
+ require.True(t, test.loops[0].Busy())
+ close(release)
+ test.events[0].WaitFor(t, "idle", 2)
+ require.Equal(t, 3, server.Calls())
+ require.NotContains(t, strings.Join(turnInputTexts(t, server.Call(1)), "\n"), input.Text)
+ require.Contains(t, strings.Join(turnInputTexts(t, server.Call(2)), "\n"), input.Text)
+ require.Equal(t, notifySource, test.events[0].LastEvent("work").Source)
+ require.Empty(t, test.Queue(t).Pending)
+ require.Len(t, test.Queue(t).Notifications, 1)
+ require.Zero(t, test.agents[1].memory.Len())
+ require.Zero(t, test.agents[2].memory.Len())
+ require.NotContains(t, test.logs.String(), input.Source)
+ require.NotContains(t, test.logs.String(), input.Text)
+ require.NotContains(t, test.logs.String(), input.ID)
+}
+
+func TestNotificationRetriesSurviveRestartBeforeAndAfterDelivery(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ input := &NotifyInput{
+ Source: "check",
+ ID: "result-1",
+ Text: "A new result.",
+ }
+
+ first := newDaemonMessageTest(t, directory, server)
+ receipt, err := first.messages.Notify(input)
+ require.NoError(t, err)
+ require.Zero(t, first.agents[0].memory.Len())
+ first.Close(t)
+
+ second := newDaemonMessageTest(t, directory, server)
+ duplicate, err := second.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.Equal(t, "duplicate", duplicate.Status)
+ require.Zero(t, second.agents[0].memory.Len())
+ second.Start(t, 0)
+ second.events[0].WaitFor(t, "idle", 1)
+ require.Empty(t, second.Queue(t).Pending)
+ second.Close(t)
+
+ third := newDaemonMessageTest(t, directory, server)
+ duplicate, err = third.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, "duplicate", duplicate.Status)
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.False(t, third.loops[0].waiting())
+ require.Equal(t, 1, server.Calls())
+
+ changed := *input
+ changed.Text = "A different result."
+ _, err = third.messages.Notify(&changed)
+ require.ErrorContains(t, err, "conflicts")
+ require.Len(t, third.Queue(t).Notifications, 1)
+}
+
+func TestNotificationAcceptanceIsSerializedForConcurrentRetries(t *testing.T) {
+ server := newModelFake(t)
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ input := &NotifyInput{
+ Source: "check",
+ ID: "1",
+ Text: "A result.",
+ }
+
+ results := make(chan Event, 20)
+ failures := make(chan error, 20)
+
+ var workers sync.WaitGroup
+
+ for range 20 {
+ workers.Add(1)
+
+ go func() {
+ defer workers.Done()
+
+ receipt, err := test.messages.Notify(input)
+ results <- receipt
+
+ failures <- err
+ }()
+ }
+
+ workers.Wait()
+ close(results)
+ close(failures)
+
+ accepted := 0
+ for receipt := range results {
+ require.Equal(t, 1, receipt.ID)
+
+ if receipt.Status == "accepted" {
+ accepted++
+ } else {
+ require.Equal(t, "duplicate", receipt.Status)
+ }
+ }
+
+ for err := range failures {
+ require.NoError(t, err)
+ }
+
+ require.Equal(t, 1, accepted)
+ require.Len(t, test.Queue(t).Pending, 1)
+ require.Len(t, test.Queue(t).Notifications, 1)
+}
+
+func TestNotificationStorageFailureReturnsNoAcceptedReceipt(t *testing.T) {
+ server := newModelFake(t)
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.messages.path = filepath.Join(t.TempDir(), "missing", "messages.json")
+ receipt, err := test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: "1",
+ Text: "A result.",
+ })
+ require.ErrorContains(t, err, "accept external event")
+ require.Empty(t, receipt.Kind)
+ require.False(t, test.loops[0].waiting())
+ require.Zero(t, test.agents[0].memory.Len())
+
+ _, err = test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: "2",
+ Text: "Another result.",
+ })
+ require.ErrorContains(t, err, "queue stopped")
+}
+
+func TestNotificationFullStoreRejectsNewKeysWithoutRemovingReceipts(t *testing.T) {
+ server := newModelFake(t)
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+
+ for number := range notifyReceipts {
+ test.messages.state.Notifications = append(test.messages.state.Notifications, NotifyReceipt{
+ Source: "check",
+ ID: fmt.Sprint(number),
+ Digest: notifyDigest("data"),
+ Number: number + 1,
+ })
+ }
+
+ test.messages.state.Next = notifyReceipts
+ receipt, err := test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: "new-key",
+ Text: "data",
+ })
+ require.ErrorContains(t, err, "store is full")
+ require.Empty(t, receipt.Kind)
+ require.Len(t, test.messages.state.Notifications, notifyReceipts)
+ require.Empty(t, test.messages.state.Pending)
+
+ receipt, err = test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: "0",
+ Text: "data",
+ })
+ require.NoError(t, err)
+ require.Equal(t, "duplicate", receipt.Status)
+}
+
+func TestNotificationBatchesKeepAcceptanceOrder(t *testing.T) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.agents[0].memory.summarizer = newFakeSummarizer()
+ for number := range 21 {
+ _, err := test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: fmt.Sprint(number),
+ Text: fmt.Sprintf("result-%02d", number),
+ })
+ require.NoError(t, err)
+ }
+
+ test.Start(t, 0)
+ test.events[0].WaitFor(t, "idle", 2)
+ require.Equal(t, 2, server.Calls())
+
+ first := turnInputTexts(t, server.Call(0))
+ require.Len(t, first, 22)
+
+ for number := range 20 {
+ require.Contains(t, first[number+2], fmt.Sprintf("result-%02d", number))
+ }
+
+ second := turnInputTexts(t, server.Call(1))
+ require.Contains(t, second[len(second)-1], "result-20")
+ require.Empty(t, test.Queue(t).Pending)
+}
+
+func TestNotificationRecoversItsPreparedLogIDWithoutAnotherAppend(t *testing.T) {
+ testCases := []struct {
+ name string
+ appended bool
+ }{
+ {
+ name: "recovers a saved ID before the log append",
+ },
+ {
+ name: "reuses the log entry after the append",
+ appended: true,
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ first := newDaemonMessageTest(t, directory, server)
+ _, err := first.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: "1",
+ Text: "A result.",
+ })
+ require.NoError(t, err)
+
+ if tc.appended {
+ require.NoError(t, first.messages.Notifications(first.loops[0]))
+ } else {
+ id := 0
+ first.messages.state.Pending[0].LogID = &id
+ require.NoError(t, first.messages.save())
+ }
+
+ first.Close(t)
+
+ second := newDaemonMessageTest(t, directory, server)
+ second.Start(t, 0)
+ second.events[0].WaitFor(t, "idle", 1)
+ require.Equal(t, 1, server.Calls())
+ require.Equal(t, 1, second.agents[0].memory.Len())
+ require.Empty(t, second.Queue(t).Pending)
+ })
+ }
+}
+
+func TestPreparedNotificationKeepsItsLogIDBeforeStartupWriters(t *testing.T) {
+ testCases := []struct {
+ name string
+ resume bool
+ child bool
+ }{
+ {
+ name: "keeps its reservation before a startup user message",
+ },
+ {
+ name: "keeps its reservation before a worker recovery note",
+ resume: true,
+ },
+ {
+ name: "keeps a child reservation before startup input",
+ child: true,
+ },
+ {
+ name: "keeps a child reservation before a worker recovery note",
+ child: true,
+ resume: true,
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ index := 0
+ if tc.child {
+ index = 1
+ }
+
+ directory := t.TempDir()
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ first := newDaemonMessageTest(t, directory, server)
+ input := &NotifyInput{
+ AgentID: first.agents[index].agent.config.ID,
+ Source: "check",
+ ID: "reserved",
+ Text: "A result with a prepared log ID.",
+ }
+
+ receipt, err := first.messages.Notify(input)
+ require.NoError(t, err)
+
+ id := 0
+ first.messages.state.Pending[0].LogID = &id
+ require.NoError(t, first.messages.save())
+
+ expected := first.Queue(t).Pending[0].Text
+ first.Close(t)
+
+ second := newDaemonMessageTest(t, directory, server)
+ require.False(t, second.loops[index].waiting())
+ require.Empty(t, second.events[index].Kinds())
+
+ if tc.resume {
+ path := filepath.Join(second.agents[index].agent.dir, subagentWorkFileName)
+ require.NoError(t, os.WriteFile(path, []byte(`{"running":["BlueFox"]}`), 0o600))
+ require.NoError(t, second.loops[index].runs.Resume(t.Context()))
+
+ note, err := second.agents[index].memory.Message(1)
+ require.NoError(t, err)
+ require.Equal(t, "note", note.Kind)
+ require.Contains(t, note.Text, "The daemon restarted, cutting BlueFox")
+ } else {
+ submitted, err := second.loops[index].Submit(terminalSource, "A user message during startup.")
+ require.NoError(t, err)
+ require.Equal(t, 1, submitted.ID)
+ }
+
+ restored, err := second.agents[index].memory.Message(id)
+ require.NoError(t, err)
+ require.Equal(t, "work", restored.Kind)
+ require.Equal(t, expected, restored.Text)
+ require.Len(t, second.Queue(t).Pending, 1)
+
+ second.Start(t, index)
+ second.events[index].WaitFor(t, "idle", 1)
+ require.Empty(t, second.Queue(t).Pending)
+ require.Equal(t, 1, server.Calls())
+ require.Equal(t, id, second.events[index].LastEvent("work").ID)
+
+ duplicate, err := second.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, "duplicate", duplicate.Status)
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.Len(t, second.Queue(t).Notifications, 1)
+ })
+ }
+}
+
+func TestNotificationBatchMakesProgressWithEscapedControlCharacters(t *testing.T) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ _, err := test.messages.Notify(&NotifyInput{
+ Source: "check",
+ ID: "escaped",
+ Text: "data" + strings.Repeat("\x00", notifyTextBytes-4),
+ })
+ require.NoError(t, err)
+ require.Greater(t, len(test.Queue(t).Pending[0].Text), 64<<10)
+ test.Start(t, 0)
+ test.events[0].WaitFor(t, "idle", 1)
+ require.Empty(t, test.Queue(t).Pending)
+}
+
+func TestNotificationSelectorsBindOnePermanentIdentity(t *testing.T) {
+ server := newModelFake(t)
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ input := &NotifyInput{
+ AgentID: test.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "targeted",
+ Text: "A result for one child.",
+ }
+
+ accepted, err := test.messages.Notify(input)
+ require.NoError(t, err)
+
+ queue := test.Queue(t)
+ require.Equal(t, test.agents[1].agent.config.ID, queue.Pending[0].TargetID)
+ require.Equal(t, queue.Pending[0].TargetID, queue.Notifications[0].TargetID)
+ require.Equal(t, "Garden", queue.Pending[0].Target)
+ require.Zero(t, test.agents[0].memory.Len())
+ require.Zero(t, test.agents[1].memory.Len())
+ require.Zero(t, test.agents[2].memory.Len())
+
+ input.AgentID = strings.ToUpper(test.agents[1].agent.config.ID)
+ duplicate, err := test.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, accepted.ID, duplicate.ID)
+ require.Equal(t, "duplicate", duplicate.Status)
+
+ input.AgentID = test.agents[2].agent.config.ID
+ _, err = test.messages.Notify(input)
+ require.ErrorContains(t, err, "conflicts")
+
+ input.AgentID = ""
+ _, err = test.messages.Notify(input)
+ require.ErrorContains(t, err, "conflicts")
+ require.Equal(t, queue, test.Queue(t))
+}
+
+func TestNotificationRejectsUnknownWorkerAndMalformedIdentities(t *testing.T) {
+ server := newModelFake(t)
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.loops[0].runs.subagents.draw = func() string {
+ return "BlueFox"
+ }
+
+ worker, err := test.loops[0].runs.subagents.Create(t.Context())
+ require.NoError(t, err)
+
+ testCases := []NotifyInput{
+ {AgentID: worker.name},
+ {AgentID: "Missing"},
+ {AgentID: "garden"},
+ {AgentID: "../Garden"},
+ {AgentID: strings.Repeat("f", 2*agentIDBytes)},
+ {AgentID: "invalid"},
+ }
+
+ for _, tc := range testCases {
+ tc.Source = "check"
+ tc.ID = "1"
+ tc.Text = "Unchanged data."
+ receipt, err := test.messages.Notify(&tc)
+ require.Error(t, err)
+ require.Empty(t, receipt.Kind)
+ }
+
+ require.Empty(t, test.messages.state.Pending)
+ require.Empty(t, test.messages.state.Notifications)
+ require.Zero(t, server.Calls())
+}
+
+func TestChildNotificationWaitsForTheWholeActiveChain(t *testing.T) {
+ server := newModelFake(t)
+ arrived := make(chan struct{})
+ release := make(chan struct{})
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ if call.number == 0 {
+ close(arrived)
+ <-release
+ modelWriteEvents(w, modelCompletedEvent([]any{modelFunctionCallItem("fc_1", "call_1", "date", `{"id":0}`)}, 100, 0, 5))
+
+ return
+ }
+
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.Start(t, 0, 1, 2)
+ require.NoError(t, test.messages.Tell("OneChat", "Garden", "Start the first chain."))
+ <-arrived
+
+ _, err := test.messages.Notify(&NotifyInput{
+ AgentID: test.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "next-turn",
+ Text: "Only the next child turn sees this.",
+ })
+ require.NoError(t, err)
+ require.Equal(t, 1, test.agents[1].memory.Len())
+ close(release)
+ test.events[1].WaitFor(t, "idle", 2)
+ require.Equal(t, 3, server.Calls())
+ require.NotContains(t, strings.Join(turnInputTexts(t, server.Call(1)), "\n"), "Only the next child turn")
+ require.Contains(t, strings.Join(turnInputTexts(t, server.Call(2)), "\n"), "Only the next child turn")
+ require.Equal(t, notifySource, test.events[1].LastEvent("work").Source)
+ require.Zero(t, test.agents[0].memory.Len())
+ require.Zero(t, test.agents[2].memory.Len())
+ require.Empty(t, test.Queue(t).Pending)
+}
+
+func TestNotificationStableIdentitySurvivesOfflineNameChange(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ first := newDaemonMessageTest(t, directory, server)
+ input := &NotifyInput{
+ AgentID: first.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "rename-proof",
+ Text: "A saved child result.",
+ }
+
+ receipt, err := first.messages.Notify(input)
+ require.NoError(t, err)
+
+ id := 0
+ first.messages.state.Pending[0].LogID = &id
+ require.NoError(t, first.messages.save())
+
+ envelope := first.Queue(t).Pending[0].Text
+ config := first.agents[1].agent.config
+ oldPath := first.agents[1].agent.dir
+ first.Close(t)
+
+ // Change only test fixtures to exercise recovery under the same
+ // identity.
+ newPath := filepath.Join(filepath.Dir(oldPath), "Orchard")
+ require.NoError(t, os.Rename(oldPath, newPath))
+
+ config.Name = "Orchard"
+ agentWriteTestConfig(t, newPath, config)
+
+ second := newDaemonMessageTest(t, directory, server)
+ child := second.messages.Child("OneChat", "Orchard")
+ require.NotNil(t, child)
+ require.False(t, child.waiting())
+
+ restored, err := child.memory.Message(id)
+ require.NoError(t, err)
+ require.Equal(t, envelope, restored.Text)
+
+ duplicate, err := second.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.Equal(t, "duplicate", duplicate.Status)
+ require.NoError(t, second.messages.Notifications(child))
+ require.NoError(t, child.turn(t.Context()))
+ require.Empty(t, second.Queue(t).Pending)
+ require.Zero(t, second.agents[0].memory.Len())
+
+ second.Close(t)
+
+ third := newDaemonMessageTest(t, directory, server)
+ duplicate, err = third.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, "duplicate", duplicate.Status)
+ require.Empty(t, third.Queue(t).Pending)
+ require.Equal(t, 1, server.Calls())
+}
+
+func TestMissingNotificationIdentityStopsStartupWithoutFallback(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ first := newDaemonMessageTest(t, directory, server)
+ _, err := first.messages.Notify(&NotifyInput{
+ AgentID: first.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "missing",
+ Text: "Do not redirect this result.",
+ })
+ require.NoError(t, err)
+
+ path := first.messages.path
+ before, err := os.ReadFile(path)
+ require.NoError(t, err)
+
+ childPath := first.agents[1].agent.dir
+ first.Close(t)
+
+ require.NoError(t, os.RemoveAll(childPath))
+ // A new agent with the same name must not receive the old identity's
+ // work.
+ daemonChildConfig(t, directory, "Garden")
+
+ agents, err := openDaemonAgents(t.Context(), directory, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, discardLog())
+ require.NoError(t, err)
+
+ var loops []*TurnLoop
+
+ for _, agent := range agents {
+ loop, err := agent.Loop(nil, func(Event) {
+ })
+ require.NoError(t, err)
+
+ loops = append(loops, loop)
+ }
+
+ _, err = openDaemonMessages(loops)
+ require.ErrorContains(t, err, "notification target is missing")
+
+ after, err := os.ReadFile(path)
+ require.NoError(t, err)
+ require.Equal(t, before, after)
+ require.Zero(t, agents[0].memory.Len())
+ require.Zero(t, agents[1].memory.Len())
+ require.NoError(t, agents.Close())
+}
+
+func TestCompletedNotificationReceiptDoesNotRequireItsTargetAtStartup(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ first := newDaemonMessageTest(t, directory, server)
+ _, err := first.messages.Notify(&NotifyInput{
+ AgentID: first.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "completed-child",
+ Text: "A completed result.",
+ })
+ require.NoError(t, err)
+ require.NoError(t, first.messages.Notifications(first.loops[1]))
+ require.NoError(t, first.loops[1].turn(t.Context()))
+ require.Empty(t, first.Queue(t).Pending)
+
+ childPath := first.agents[1].agent.dir
+ first.Close(t)
+ require.NoError(t, os.RemoveAll(childPath))
+
+ second := newDaemonMessageTest(t, directory, server)
+ require.Empty(t, second.Queue(t).Pending)
+ require.Len(t, second.Queue(t).Notifications, 1)
+ require.Zero(t, second.agents[0].memory.Len())
+}
+
+func TestLegacyNotificationReceiptsStillBindTheRoot(t *testing.T) {
+ directory := t.TempDir()
+ server := newModelFake(t)
+ first := newDaemonMessageTest(t, directory, server)
+ input := &NotifyInput{
+ Source: "check",
+ ID: "legacy",
+ Text: "A root result.",
+ }
+
+ receipt, err := first.messages.Notify(input)
+ require.NoError(t, err)
+
+ first.messages.state.Pending[0].TargetID = ""
+ first.messages.state.Notifications[0].TargetID = ""
+ require.NoError(t, first.messages.save())
+
+ before, err := os.ReadFile(first.messages.path)
+ require.NoError(t, err)
+ first.Close(t)
+
+ second := newDaemonMessageTest(t, directory, server)
+ after, err := os.ReadFile(second.messages.path)
+ require.NoError(t, err)
+ require.Equal(t, before, after)
+
+ duplicate, err := second.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, receipt.ID, duplicate.ID)
+ require.Equal(t, "duplicate", duplicate.Status)
+
+ input.AgentID = second.agents[0].agent.config.ID
+ duplicate, err = second.messages.Notify(input)
+ require.NoError(t, err)
+ require.Equal(t, "duplicate", duplicate.Status)
+
+ input.AgentID = second.agents[1].agent.config.ID
+ _, err = second.messages.Notify(input)
+ require.ErrorContains(t, err, "conflicts")
+}
+
+func TestChildNotificationRunsWhileTheRootIsBusy(t *testing.T) {
+ server := newModelFake(t)
+ arrived := make(chan struct{})
+ release := make(chan struct{})
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ if call.number == 0 {
+ close(arrived)
+ <-release
+ }
+
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.Start(t, 0, 1)
+
+ _, err := test.loops[0].Submit(terminalSource, "Keep the root busy.")
+ require.NoError(t, err)
+ <-arrived
+
+ _, err = test.messages.Notify(&NotifyInput{
+ AgentID: test.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "independent",
+ Text: "A result for the idle child.",
+ })
+ require.NoError(t, err)
+ test.events[1].WaitFor(t, "idle", 1)
+ require.True(t, test.loops[0].Busy())
+ require.Equal(t, 1, test.agents[0].memory.Len())
+ require.Empty(t, test.events[0].Last("work"))
+ close(release)
+ test.events[0].WaitFor(t, "idle", 1)
+ require.Empty(t, test.Queue(t).Pending)
+ require.Equal(t, 2, server.Calls())
+}
+
+func TestNotificationBatchesStayOrderedForEachChild(t *testing.T) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.agents[1].memory.summarizer = newFakeSummarizer()
+ test.agents[2].memory.summarizer = newFakeSummarizer()
+
+ for number := range 21 {
+ for index := 1; index <= 2; index++ {
+ _, err := test.messages.Notify(&NotifyInput{
+ AgentID: test.agents[index].agent.config.ID,
+ Source: "check",
+ ID: fmt.Sprintf("%d-%d", index, number),
+ Text: fmt.Sprintf("target-%d-result-%02d", index, number),
+ })
+ require.NoError(t, err)
+ }
+ }
+
+ test.Start(t, 0, 1, 2)
+ test.events[1].WaitFor(t, "idle", 2)
+ test.events[2].WaitFor(t, "idle", 2)
+ require.Equal(t, 4, server.Calls())
+ require.Zero(t, test.agents[0].memory.Len())
+ require.Empty(t, test.Queue(t).Pending)
+
+ for index := 1; index <= 2; index++ {
+ for number := range 21 {
+ logged, err := test.agents[index].memory.Message(number)
+ require.NoError(t, err)
+ require.Contains(t, logged.Text, fmt.Sprintf("target-%d-result-%02d", index, number))
+ }
+ }
+}
+
+func TestChildNotificationReportsOnlyItsFinalReplyToTheRoot(t *testing.T) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
+ reply := ""
+ if call.body["model"] == "gpt-test-garden" {
+ reply = "A result needs the user."
+ }
+
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(reply)}, 100, 0, 5))
+ }
+
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ test.Start(t, 0, 1, 2)
+
+ _, err := test.messages.Notify(&NotifyInput{
+ AgentID: test.agents[1].agent.config.ID,
+ Source: "check",
+ ID: "report",
+ Text: "Private child event data.",
+ })
+ require.NoError(t, err)
+ test.events[1].WaitFor(t, "idle", 1)
+ test.events[0].WaitFor(t, "idle", 1)
+ require.Equal(t, 2, server.Calls())
+ require.Equal(t, "[Garden] A result needs the user.", test.events[0].Last("work"))
+ require.Equal(t, 1, test.events[0].Count("work"))
+ require.Equal(t, notifySource, test.events[1].LastEvent("work").Source)
+ require.Zero(t, test.agents[2].memory.Len())
+ require.Empty(t, test.Queue(t).Pending)
+}
+
func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) {
server := newModelFake(t)
server.reply = func(w http.ResponseWriter, call ModelFakeCall) {
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
}
+
test := newDaemonMessageTest(t, t.TempDir(), server)
test.Start(t, 0, 1, 2)
require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the gate size."))
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
}
+
test := newDaemonMessageTest(t, t.TempDir(), server)
telegram := newTelegramFake(t)
bot := newTelegramBot(newTelegramAPI(telegram.URL, telegramTestToken), telegramTestOwner, filepath.Join(t.TempDir(), telegramOffsetFileName), discardLog())
require.Eventually(t, func() bool {
return len(telegram.Sent()) >= 2
}, 10*time.Second, time.Millisecond)
+
require.Equal(t, []string{"[Garden] finished", "I will ask the user."}, telegram.Sent())
require.Equal(t, subagentFinished, test.events[0].LastEvent("work").Status)
require.Equal(t, "[Garden] Approve the gate size.", test.events[0].Last("work"))
text string
}{
{
- name: "empty final reply",
+ name: "sends nothing for an empty final reply",
text: "",
},
{
- name: "whitespace final reply",
+ name: "sends nothing for a whitespace final reply",
text: " \n\t ",
},
}
+
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
server := newModelFake(t)
server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(tc.text)}, 100, 0, 5))
}
+
test := newDaemonMessageTest(t, t.TempDir(), server)
test.Start(t, 0, 1, 2)
require.NoError(t, test.messages.Tell("OneChat", "Garden", "Keep the plan."))
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
}
+
test := newDaemonMessageTest(t, t.TempDir(), server)
test.Start(t, 0, 1)
require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the first plan."))
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5))
}
+
test := newDaemonMessageTest(t, t.TempDir(), server)
test.Start(t, 0, 1)
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan needs approval.")}, 100, 0, 5))
}
+
test := newDaemonMessageTest(t, t.TempDir(), server)
test.Start(t, 1)
require.NoError(t, test.messages.Tell("OneChat", "Garden", "Check the plan."))
server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
}
+
first := newDaemonMessageTest(t, dir, server)
require.NoError(t, first.messages.Tell("OneChat", "Garden", "Remember the garden plan."))
require.Equal(t, 1, first.agents[1].memory.Len())
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
}
+
first := newDaemonMessageTest(t, dir, server)
first.Start(t, 1)
require.NoError(t, first.messages.Tell("OneChat", "Garden", "Finish the garden plan."))
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("The plan is ready.")}, 100, 0, 5))
}
+
first := newDaemonMessageTest(t, dir, server)
first.Start(t, 0, 1)
require.NoError(t, first.messages.Tell("OneChat", "Garden", "Review the plan."))
require.Never(t, func() bool {
return len(sent) > 0
}, 50*time.Millisecond, time.Millisecond)
+
once.Do(func() {
close(release)
})
target: "OneChat",
},
{
- name: "unknown sender",
+ name: "refuses an unknown sender",
source: "Other",
target: "Garden",
},
{
- name: "unknown recipient",
+ name: "refuses an unknown recipient",
source: "OneChat",
target: "Other",
},
}
+
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
require.ErrorContains(t, test.messages.Tell(tc.source, tc.target, "Do not deliver."), "not a child")
file string
}{
{
- name: "broken JSON",
+ name: "stops startup for broken JSON",
file: "{",
},
{
- name: "unknown sender",
+ name: "stops startup for unknown sender",
file: `{"next":1,"pending":[{"id":1,"source":"Other","target":"Garden","kind":"user"}]}`,
},
{
- name: "sibling delivery",
+ name: "stops startup for sibling delivery",
file: `{"next":1,"pending":[{"id":1,"source":"Library","target":"Garden","kind":"user"}]}`,
},
{
- name: "invalid sequence",
+ name: "stops startup for invalid sequence",
file: `{"next":0,"pending":[{"id":1,"source":"OneChat","target":"Garden","kind":"user"}]}`,
},
}
+
for _, tc := range testCases {
t.Run(tc.name, func(t *testing.T) {
dir := t.TempDir()
}
}
-// DaemonMessageTest owns test agents and the loops selected by Start.
-type DaemonMessageTest struct {
- agents DaemonAgents
- loops []*TurnLoop
- events []*TurnEventRecorder
- messages *DaemonMessages
- logs *SyncBuffer
- cancel context.CancelFunc
- done []chan error
+func TestChildScheduleWakesOnlyItsOwner(t *testing.T) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ test := newDaemonMessageTest(t, t.TempDir(), server)
+ child := test.agents[1]
+ at := child.clock().Add(time.Minute).Format(scheduleAtLayout)
+ output := test.loops[1].tools.run(t.Context(), "schedule_set", `{"text":"Check the plants.","at":"`+at+`","cron":null}`)
+ require.Contains(t, output, "Schedule 1 set.")
+ require.Equal(t, "Garden", child.schedules.sorted()[0].Owner)
+
+ child.schedules.clock = func() time.Time {
+ return child.clock().Add(time.Minute)
+ }
+ require.ErrorContains(t, child.schedules.deliver(t.Context(), test.loops[0]), "does not match")
+ require.NoError(t, child.schedules.deliver(t.Context(), test.loops[1]))
+ test.Start(t, 0, 1, 2)
+ test.events[1].WaitFor(t, "idle", 1)
+ require.Equal(t, 1, server.Calls())
+ require.Equal(t, child.agent.CacheKey(), server.Call(0).body["prompt_cache_key"])
+ require.Zero(t, test.agents[0].memory.Len())
+ require.Zero(t, test.agents[2].memory.Len())
+ require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.")
+ require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source)
}
-func (r *DaemonMessageTest) Start(t *testing.T, indices ...int) {
- t.Helper()
+func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) {
+ testCases := []struct {
+ name string
+ chat WhatsAppChat
+ }{
+ {
+ name: "routes a private chat only to its child",
+ chat: WhatsAppChat{
+ ID: whatsappTestAna.String(),
+ Name: "Ana",
+ },
+ },
+ {
+ name: "routes an unmentioned group only to its child",
+ chat: WhatsAppChat{
+ ID: whatsappTestFamily.String(),
+ Name: "Garden plans",
+ Group: true,
+ },
+ },
+ }
- ctx, cancel := context.WithCancel(t.Context())
- r.cancel = cancel
- for _, index := range indices {
- done := make(chan error, 1)
- r.done = append(r.done, done)
- go func() {
- done <- r.loops[index].Run(ctx)
- }()
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ dir := t.TempDir()
+ test := newDaemonMessageTest(t, dir, server)
+ inbox := whatsappRouteTestInbox(t, dir)
+ require.NoError(t, test.messages.routes.Set("whatsapp:"+tc.chat.ID, "Garden"))
+ whatsappRouteTestAdd(t, inbox, tc.chat, "The gate needs a new latch.")
+ test.Start(t, 0, 1, 2)
+ require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
+ test.events[1].WaitFor(t, "idle", 1)
+ require.Equal(t, 1, server.Calls())
+ require.Equal(t, "gpt-test-garden", server.Call(0).body["model"])
+ require.Contains(t, string(server.Call(0).raw), "The gate needs a new latch.")
+ require.Zero(t, test.agents[0].memory.Len())
+ require.Empty(t, test.events[0].Kinds())
+ require.Zero(t, test.agents[2].memory.Len())
+ require.False(t, test.loops[0].waiting())
+ require.Equal(t, whatsappSource, test.events[1].LastEvent("work").Source)
+ require.Empty(t, test.events[1].LastEvent("work").Status)
+ require.Equal(t, 1, test.agents[1].memory.Len())
+ require.Empty(t, inbox.state.Closed())
+ require.Empty(t, test.Queue(t).Pending)
+ require.Contains(t, test.logs.String(), "batch routed name=Garden kind=whatsapp id=1 to=Garden")
+ require.NotContains(t, test.logs.String(), tc.chat.ID)
+ require.NotContains(t, test.logs.String(), "new latch")
+ })
}
}
-func (r *DaemonMessageTest) Queue(t *testing.T) DaemonMessageFile {
- t.Helper()
-
- data, err := os.ReadFile(r.messages.path)
- require.NoError(t, err)
+func TestWhatsAppUnrouteRestoresRootDelivery(t *testing.T) {
+ dir := t.TempDir()
+ test := newDaemonMessageTest(t, dir, newModelFake(t))
+ inbox := whatsappRouteTestInbox(t, dir)
+ chat := WhatsAppChat{
+ ID: whatsappTestAna.String(),
+ Name: "Ana",
+ }
- var file DaemonMessageFile
+ source := "whatsapp:" + chat.ID
+ require.NoError(t, test.messages.routes.Set(source, "Garden"))
+ whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
+ require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
+ require.Zero(t, test.agents[0].memory.Len())
+ require.True(t, test.loops[1].waiting())
+ require.NoError(t, test.messages.routes.Set(source, ""))
+ whatsappRouteTestAdd(t, inbox, chat, "Please check the other plan.")
+ require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
+ require.Equal(t, 1, test.agents[0].memory.Len())
+ require.Equal(t, 1, test.agents[1].memory.Len())
+ require.True(t, test.loops[0].waiting())
+ require.Contains(t, test.events[0].Last("work"), "Please check the other plan.")
+ require.NotContains(t, test.events[0].Last("work"), "Keep the gate plan.")
+ require.Empty(t, inbox.state.Closed())
+}
- require.NoError(t, json.Unmarshal(data, &file))
+func TestMissingWhatsAppRouteChildFallsBackToRoot(t *testing.T) {
+ dir := t.TempDir()
+ test := newDaemonMessageTest(t, dir, newModelFake(t))
+ inbox := whatsappRouteTestInbox(t, dir)
+ chat := WhatsAppChat{
+ ID: whatsappTestAna.String(),
+ Name: "Ana",
+ }
- return file
+ require.NoError(t, test.messages.routes.Set("whatsapp:"+chat.ID, "Missing"))
+ whatsappRouteTestAdd(t, inbox, chat, "Please check the plan.")
+ require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
+ require.Equal(t, 1, test.agents[0].memory.Len())
+ require.True(t, test.loops[0].waiting())
+ require.Zero(t, test.agents[1].memory.Len())
+ require.False(t, test.loops[1].waiting())
+ require.Empty(t, test.events[0].LastEvent("work").Status)
+ require.Equal(t, whatsappSource, test.events[0].LastEvent("work").Source)
+ require.Empty(t, inbox.state.Closed())
}
-func (r *DaemonMessageTest) Close(t *testing.T) {
- t.Helper()
+func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) {
+ testCases := []struct {
+ name string
+ finish bool
+ }{
+ {
+ name: "queued turn resumes",
+ finish: false,
+ },
+ {
+ name: "finished turn stays finished",
+ finish: true,
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ dir := t.TempDir()
+ server := newModelFake(t)
+ server.reply = func(w http.ResponseWriter, _ ModelFakeCall) {
+ modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("")}, 100, 0, 5))
+ }
+
+ first := newDaemonMessageTest(t, dir, server)
+ inbox := whatsappRouteTestInbox(t, dir)
+ chat := WhatsAppChat{
+ ID: whatsappTestAna.String(),
+ Name: "Ana",
+ }
- if r.cancel != nil {
- r.cancel()
+ source := "whatsapp:" + chat.ID
+ require.NoError(t, first.messages.routes.Set(source, "Garden"))
+ whatsappRouteTestAdd(t, inbox, chat, "Remember the gate plan.")
+ require.NoError(t, inbox.state.CloseAll())
- for _, done := range r.done {
- require.ErrorIs(t, <-done, context.Canceled)
- }
+ batch := inbox.state.Closed()[0]
+ routed, err := first.messages.RouteWhatsApp(batch, batch.Text(time.UTC))
+ require.NoError(t, err)
+ require.True(t, routed)
+
+ if tc.finish {
+ first.Start(t, 1)
+ first.events[1].WaitFor(t, "idle", 1)
+ require.Empty(t, first.Queue(t).Pending)
+ }
+
+ first.Close(t)
+
+ second := newDaemonMessageTest(t, dir, server)
+ require.Equal(t, source+" -> Garden", second.messages.routes.List())
+ require.Equal(t, 1, second.agents[1].memory.Len())
+ require.Equal(t, !tc.finish, second.loops[1].waiting())
+ require.NoError(t, second.messages.routes.Set(source, ""))
+
+ restored := whatsappRouteTestInbox(t, dir)
+ require.Len(t, restored.state.Closed(), 1)
+ require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
+ require.Empty(t, restored.state.Closed())
+ require.Zero(t, second.agents[0].memory.Len())
+ require.Equal(t, 1, second.agents[1].memory.Len())
+
+ if !tc.finish {
+ second.Start(t, 1)
+ second.events[1].WaitFor(t, "idle", 1)
+ }
+
+ require.Equal(t, 1, server.Calls())
+ require.Empty(t, second.Queue(t).Pending)
+ second.Close(t)
+
+ third := newDaemonMessageTest(t, dir, server)
+ require.False(t, third.loops[1].waiting())
+ require.Empty(t, third.events[1].Kinds())
+ require.Equal(t, 1, third.agents[1].memory.Len())
+ })
+ }
+}
- r.cancel = nil
- r.done = nil
+func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) {
+ testCases := []struct {
+ name string
+ id *int
+ }{
+ {
+ name: "recovers one message before its log ID is stored",
+ id: nil,
+ },
+ {
+ name: "recovers one message after its log ID is stored",
+ id: new(0),
+ },
}
- require.NoError(t, r.agents.Close())
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ dir := t.TempDir()
+ server := newModelFake(t)
+ first := newDaemonMessageTest(t, dir, server)
+ path := first.messages.path
+ first.Close(t)
+
+ file := DaemonMessageFile{
+ Next: 1,
+ WhatsAppBatches: []int{1},
+ Pending: []DaemonMessage{{
+ ID: 1,
+ Source: "OneChat",
+ Target: "Garden",
+ Kind: "work",
+ Text: "[WhatsApp] Keep the gate plan.",
+ LogID: tc.id,
+ WhatsAppBatch: 1,
+ }},
+ }
- r.agents = nil
+ data, err := json.Marshal(file)
+ require.NoError(t, err)
+ require.NoError(t, os.WriteFile(path, data, 0o600))
+
+ second := newDaemonMessageTest(t, dir, server)
+ require.Equal(t, 1, second.agents[1].memory.Len())
+ require.True(t, second.loops[1].waiting())
+ require.Equal(t, "[WhatsApp] Keep the gate plan.", second.events[1].Last("work"))
+ require.Equal(t, whatsappSource, second.events[1].LastEvent("work").Source)
+ second.Close(t)
+
+ third := newDaemonMessageTest(t, dir, server)
+ require.Equal(t, 1, third.agents[1].memory.Len())
+ require.True(t, third.loops[1].waiting())
+ require.Zero(t, third.agents[0].memory.Len())
+ })
+ }
}
-func newDaemonMessageTest(t *testing.T, dir string, server *ModelFake) *DaemonMessageTest {
- t.Helper()
+func TestQueuedWhatsAppBatchUsesRootIfItsChildIsRemoved(t *testing.T) {
+ dir := t.TempDir()
+ server := newModelFake(t)
+ first := newDaemonMessageTest(t, dir, server)
+ inbox := whatsappRouteTestInbox(t, dir)
+ chat := WhatsAppChat{
+ ID: whatsappTestAna.String(),
+ Name: "Ana",
+ }
- _, err := os.Stat(filepath.Join(dir, daemonAgentDirName, "children"))
- if errors.Is(err, os.ErrNotExist) {
- daemonChildConfig(t, dir, "Garden")
- daemonChildConfig(t, dir, "Library")
- } else {
- require.NoError(t, err)
+ require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
+ whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
+ require.NoError(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true))
+ first.Close(t)
+ require.NoError(t, os.RemoveAll(filepath.Join(dir, daemonAgentDirName, "children", "Garden")))
+
+ second := newDaemonMessageTest(t, dir, server)
+ require.Equal(t, 1, second.agents[0].memory.Len())
+ require.True(t, second.loops[0].waiting())
+ require.Equal(t, whatsappSource, second.events[0].LastEvent("work").Source)
+ require.Empty(t, second.events[0].LastEvent("work").Status)
+ require.Contains(t, second.events[0].Last("work"), "Keep the gate plan.")
+ require.False(t, second.loops[1].waiting())
+ second.Close(t)
+
+ third := newDaemonMessageTest(t, dir, server)
+ require.Equal(t, 1, third.agents[0].memory.Len())
+ require.True(t, third.loops[0].waiting())
+}
+
+func TestRoutedWhatsAppBatchesCanCloseOutOfNumberOrder(t *testing.T) {
+ dir := t.TempDir()
+ test := newDaemonMessageTest(t, dir, newModelFake(t))
+ inbox := whatsappRouteTestInbox(t, dir)
+ first := WhatsAppChat{
+ ID: whatsappTestAna.String(),
+ Name: "Ana",
}
- test := &DaemonMessageTest{
- logs: &SyncBuffer{},
+ second := WhatsAppChat{
+ ID: whatsappTestFamily.String(),
+ Name: "Garden plans",
+ Group: true,
}
- test.agents, err = openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, newTestClock().Now, daemonTestLogger(test.logs))
+
+ require.NoError(t, test.messages.routes.Set("whatsapp:"+first.ID, "Garden"))
+ require.NoError(t, test.messages.routes.Set("whatsapp:"+second.ID, "Garden"))
+
+ now := time.Now()
+ _, err := inbox.state.Add(first, WhatsAppMessage{
+ ID: "first",
+ Text: "The first chat is still open.",
+ Sender: "Ana",
+ Time: now,
+ }, now)
require.NoError(t, err)
- t.Cleanup(func() {
- test.Close(t)
- })
+
+ _, err = inbox.state.Add(second, WhatsAppMessage{
+ ID: "second",
+ Text: "The second chat is quiet.",
+ Sender: "Bruno",
+ Time: now,
+ }, now.Add(-time.Minute))
+ require.NoError(t, err)
+ require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
+ require.Equal(t, 1, test.agents[1].memory.Len())
+ require.Contains(t, test.events[1].Last("work"), "The second chat is quiet.")
+ require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, true))
+ require.Equal(t, 2, test.agents[1].memory.Len())
+ require.Contains(t, test.events[1].Last("work"), "The first chat is still open.")
+ require.Zero(t, test.agents[0].memory.Len())
+ require.Empty(t, test.Queue(t).WhatsAppBatches)
+}
+
+func TestWhatsAppCatchupSeparatesChildAndRootBatches(t *testing.T) {
+ dir := t.TempDir()
+ test := newDaemonMessageTest(t, dir, newModelFake(t))
+ inbox := whatsappRouteTestInbox(t, dir)
+ garden := WhatsAppChat{
+ ID: whatsappTestFamily.String(),
+ Name: "Garden plans",
+ Group: true,
+ }
+
+ library := WhatsAppChat{
+ ID: whatsappTestBruno.String(),
+ Name: "Bruno",
+ }
+
+ root := WhatsAppChat{
+ ID: whatsappTestAna.String(),
+ Name: "Ana",
+ }
+
+ require.NoError(t, test.messages.routes.Set("whatsapp:"+garden.ID, "Garden"))
+ require.NoError(t, test.messages.routes.Set("whatsapp:"+library.ID, "Library"))
+ inbox.waitForCatchup()
+ whatsappRouteTestAdd(t, inbox, garden, "Check the gate.")
+ whatsappRouteTestAdd(t, inbox, library, "Check the book.")
+ whatsappRouteTestAdd(t, inbox, root, "Check the mixed chat.")
+ require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
for _, agent := range test.agents {
- events := &TurnEventRecorder{}
- loop, err := agent.Loop(nil, events.Add)
- require.NoError(t, err)
+ require.Zero(t, agent.memory.Len())
+ }
+
+ inbox.endCatchup()
+ require.NoError(t, inbox.deliver(t.Context(), test.loops[0], WhatsAppLink{}, false))
+
+ for _, loop := range test.loops {
+ require.Equal(t, 1, loop.memory.Len())
+ require.True(t, loop.waiting())
+ }
+
+ require.Contains(t, test.events[0].Last("work"), "Check the mixed chat.")
+ require.Contains(t, test.events[1].Last("work"), "Check the gate.")
+ require.Contains(t, test.events[2].Last("work"), "Check the book.")
+ require.Empty(t, inbox.state.Closed())
+ require.Empty(t, test.Queue(t).WhatsAppBatches)
+}
+
+func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) {
+ dir := t.TempDir()
+ server := newModelFake(t)
+ first := newDaemonMessageTest(t, dir, server)
+ inbox := whatsappRouteTestInbox(t, dir)
+ chat := WhatsAppChat{
+ ID: whatsappTestAna.String(),
+ Name: "Ana",
+ }
- test.events = append(test.events, events)
- test.loops = append(test.loops, loop)
+ require.NoError(t, first.messages.routes.Set("whatsapp:"+chat.ID, "Garden"))
+ require.NoError(t, os.Mkdir(first.messages.path, 0o700))
+ whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.")
+ require.ErrorContains(t, inbox.deliver(t.Context(), first.loops[0], WhatsAppLink{}, true), "save agent messages")
+ require.Len(t, inbox.state.Closed(), 1)
+
+ for _, loop := range first.loops {
+ require.ErrorContains(t, loop.failed(), "save agent messages")
+ require.Zero(t, loop.memory.Len())
}
- test.messages, err = openDaemonMessages(test.loops)
+ path := first.messages.path
+ first.Close(t)
+ require.NoError(t, os.Remove(path))
+
+ second := newDaemonMessageTest(t, dir, server)
+ restored := whatsappRouteTestInbox(t, dir)
+ require.NoError(t, restored.deliver(t.Context(), second.loops[0], WhatsAppLink{}, true))
+ require.Equal(t, 1, second.agents[1].memory.Len())
+ require.True(t, second.loops[1].waiting())
+ require.Zero(t, second.agents[0].memory.Len())
+ require.Empty(t, restored.state.Closed())
+}
+
+func whatsappRouteTestInbox(t *testing.T, dir string) *WhatsAppInbox {
+ t.Helper()
+
+ state, err := openWhatsAppState(filepath.Join(dir, whatsappDirName, whatsappStateFileName))
require.NoError(t, err)
- return test
+ return newWhatsAppInbox(state, newTestClock().Now, newWhatsAppGroups(discardLog()), &ImageStore{dir: filepath.Join(dir, imageDirName)}, discardLog())
+}
+
+func whatsappRouteTestAdd(t *testing.T, inbox *WhatsAppInbox, chat WhatsAppChat, text string) {
+ t.Helper()
+
+ at := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
+ _, err := inbox.state.Add(chat, WhatsAppMessage{
+ ID: text,
+ Sender: "Ana",
+ Text: text,
+ Time: at,
+ }, at)
+ require.NoError(t, err)
}
projects / onechat / blobdiff