X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/a5c83e92d7feea2e104b5515043a76c8c4c56c60..f4509310b5abe5012cb83bed3960404e8fa92217:/daemon_message_test.go?ds=inline diff --git a/daemon_message_test.go b/daemon_message_test.go index 432a640..0ef67fc 100644 --- a/daemon_message_test.go +++ b/daemon_message_test.go @@ -4,9 +4,11 @@ import ( "context" "encoding/json" "errors" + "fmt" "net/http" "os" "path/filepath" + "strings" "sync" "testing" "time" @@ -14,6 +16,801 @@ import ( "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) { @@ -24,6 +821,7 @@ func TestParentMessageWakesChildAndFinalReplyWakesParent(t *testing.T) { 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.")) @@ -68,6 +866,7 @@ func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) { 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()) @@ -97,6 +896,7 @@ func TestChildReportSendsATelegramCompletionBeforeTheParentReply(t *testing.T) { 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")) @@ -122,20 +922,22 @@ func TestEmptyChildReplySendsNothingAndWakesNobody(t *testing.T) { 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.")) @@ -162,6 +964,7 @@ func TestBusyChildQueuesTheNextParentMessage(t *testing.T) { 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.")) @@ -194,6 +997,7 @@ func TestBusyParentQueuesItsChildReport(t *testing.T) { modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(text)}, 100, 0, 5)) } + test := newDaemonMessageTest(t, t.TempDir(), server) test.Start(t, 0, 1) @@ -232,6 +1036,7 @@ func TestChildReportsOnlyTheFinalReplyAndTakesQueuedToolInput(t *testing.T) { 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.")) @@ -259,6 +1064,7 @@ func TestQueuedParentMessageSurvivesDaemonRestart(t *testing.T) { 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()) @@ -319,6 +1125,7 @@ func TestInterruptedChildDeliveryResumesAfterDaemonRestart(t *testing.T) { 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.")) @@ -348,6 +1155,7 @@ func TestInterruptedParentReportResumesAfterDaemonRestart(t *testing.T) { 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.")) @@ -419,6 +1227,7 @@ func TestAgentMessageSnapshotBlocksDelivery(t *testing.T) { require.Never(t, func() bool { return len(sent) > 0 }, 50*time.Millisecond, time.Millisecond) + once.Do(func() { close(release) }) @@ -453,16 +1262,17 @@ func TestParentCanMessageOnlyItsOwnChildren(t *testing.T) { 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") @@ -481,22 +1291,23 @@ func TestInvalidAgentMessageQueueStopsStartup(t *testing.T) { 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() @@ -537,6 +1348,7 @@ func TestChildScheduleWakesOnlyItsOwner(t *testing.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) @@ -565,14 +1377,14 @@ func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) { chat WhatsAppChat }{ { - name: "private chat", + name: "routes a private chat only to its child", chat: WhatsAppChat{ ID: whatsappTestAna.String(), Name: "Ana", }, }, { - name: "group without a mention", + name: "routes an unmentioned group only to its child", chat: WhatsAppChat{ ID: whatsappTestFamily.String(), Name: "Garden plans", @@ -580,12 +1392,14 @@ func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.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("")}, 100, 0, 5)) } + dir := t.TempDir() test := newDaemonMessageTest(t, dir, server) inbox := whatsappRouteTestInbox(t, dir) @@ -621,6 +1435,7 @@ func TestWhatsAppUnrouteRestoresRootDelivery(t *testing.T) { ID: whatsappTestAna.String(), Name: "Ana", } + source := "whatsapp:" + chat.ID require.NoError(t, test.messages.routes.Set(source, "Garden")) whatsappRouteTestAdd(t, inbox, chat, "Keep the gate plan.") @@ -646,6 +1461,7 @@ func TestMissingWhatsAppRouteChildFallsBackToRoot(t *testing.T) { ID: whatsappTestAna.String(), Name: "Ana", } + 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)) @@ -672,6 +1488,7 @@ func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.T) { finish: true, }, } + for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { dir := t.TempDir() @@ -679,12 +1496,14 @@ func TestRoutedWhatsAppRestartDoesNotDeliverTheBatchTwice(t *testing.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", } + source := "whatsapp:" + chat.ID require.NoError(t, first.messages.routes.Set(source, "Garden")) whatsappRouteTestAdd(t, inbox, chat, "Remember the gate plan.") @@ -739,14 +1558,15 @@ func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) { id *int }{ { - name: "before log ID is stored", + name: "recovers one message before its log ID is stored", id: nil, }, { - name: "after log ID is stored", + name: "recovers one message after its log ID is stored", id: new(0), }, } + for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { dir := t.TempDir() @@ -768,6 +1588,7 @@ func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) { WhatsAppBatch: 1, }}, } + data, err := json.Marshal(file) require.NoError(t, err) require.NoError(t, os.WriteFile(path, data, 0o600)) @@ -796,6 +1617,7 @@ func TestQueuedWhatsAppBatchUsesRootIfItsChildIsRemoved(t *testing.T) { ID: whatsappTestAna.String(), Name: "Ana", } + 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)) @@ -824,11 +1646,13 @@ func TestRoutedWhatsAppBatchesCanCloseOutOfNumberOrder(t *testing.T) { ID: whatsappTestAna.String(), Name: "Ana", } + second := WhatsAppChat{ ID: whatsappTestFamily.String(), Name: "Garden plans", Group: true, } + require.NoError(t, test.messages.routes.Set("whatsapp:"+first.ID, "Garden")) require.NoError(t, test.messages.routes.Set("whatsapp:"+second.ID, "Garden")) @@ -867,14 +1691,17 @@ func TestWhatsAppCatchupSeparatesChildAndRootBatches(t *testing.T) { 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() @@ -911,6 +1738,7 @@ func TestRoutedWhatsAppQueueFailureKeepsTheInboxBatch(t *testing.T) { ID: whatsappTestAna.String(), Name: "Ana", } + 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.")