From: Anderson A. Giacomolli Date: Wed, 11 Nov 2026 13:44:30 +0000 (-0300) Subject: Keep scenario tests with their implementation X-Git-Url: https://git.giacomolli.uk/onechat/commitdiff_plain/a5c83e92d7feea2e104b5515043a76c8c4c56c60?ds=inline Keep scenario tests with their implementation Memory tests hold replay and queued append scenarios, and daemon queue tests hold routed WhatsApp scenarios. Their source filenames identify the implementation they exercise without empty production files. Helpers with one consumer stay with that consumer. All function bodies and test behavior stay unchanged. Issue: 947bddb --- diff --git a/daemon_message_test.go b/daemon_message_test.go index 5574172..432a640 100644 --- a/daemon_message_test.go +++ b/daemon_message_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "errors" "net/http" "os" @@ -557,3 +558,401 @@ func TestChildScheduleWakesOnlyItsOwner(t *testing.T) { require.Contains(t, test.events[1].LastEvent("work").Text, "Check the plants.") require.Equal(t, scheduleSource, test.events[1].LastEvent("work").Source) } + +func TestRoutedWhatsAppBatchesWakeOnlyTheirChild(t *testing.T) { + testCases := []struct { + name string + chat WhatsAppChat + }{ + { + name: "private chat", + chat: WhatsAppChat{ + ID: whatsappTestAna.String(), + Name: "Ana", + }, + }, + { + name: "group without a mention", + chat: WhatsAppChat{ + ID: whatsappTestFamily.String(), + Name: "Garden plans", + Group: true, + }, + }, + } + 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 TestWhatsAppUnrouteRestoresRootDelivery(t *testing.T) { + dir := t.TempDir() + test := newDaemonMessageTest(t, dir, newModelFake(t)) + inbox := whatsappRouteTestInbox(t, dir) + chat := WhatsAppChat{ + 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.") + 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()) +} + +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", + } + 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 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", + } + 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()) + + 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()) + }) + } +} + +func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) { + testCases := []struct { + name string + id *int + }{ + { + name: "before log ID is stored", + id: nil, + }, + { + name: "after log ID is stored", + id: new(0), + }, + } + 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, + }}, + } + 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 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", + } + 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", + } + 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")) + + 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) + + _, 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 { + 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", + } + 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()) + } + + 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 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) +} diff --git a/daemon_start_helper_test.go b/daemon_start_helper_test.go index c43ff9c..b09c11b 100644 --- a/daemon_start_helper_test.go +++ b/daemon_start_helper_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "os" "path/filepath" "testing" "time" @@ -65,3 +66,19 @@ func daemonStartTestWithSchedules(t *testing.T, server *ModelFake, bot *Telegram return socket, memory } + +// socketTestDir returns a new short directory under /tmp for data that holds a +// daemon socket. t.TempDir holds the test name and the TMPDIR of the shell, so +// its paths can pass the 107-byte limit of a Unix socket path. +func socketTestDir(t *testing.T) string { + t.Helper() + + dir, err := os.MkdirTemp("/tmp", "onechat-") + require.NoError(t, err) + + t.Cleanup(func() { + require.NoError(t, os.RemoveAll(dir)) + }) + + return dir +} diff --git a/memory_delivery_test.go b/memory_delivery_test.go deleted file mode 100644 index 05b74bc..0000000 --- a/memory_delivery_test.go +++ /dev/null @@ -1,117 +0,0 @@ -package main - -import ( - "errors" - "testing" - - "github.com/stretchr/testify/require" -) - -func TestQueuedAppendResumesWithoutADuplicate(t *testing.T) { - dir := t.TempDir() - memory, err := openMemory(t.Context(), dir, newTestClock().Now, newFakeSummarizer(), discardLog()) - require.NoError(t, err) - - var saved *int - - first, err := memory.AppendOnce("user", "Record the gate size.", nil, func(id int) error { - saved = &id - - require.Equal(t, 0, memory.log.Len()) - - return nil - }) - require.NoError(t, err) - require.NoError(t, memory.Close()) - - memory, err = openMemory(t.Context(), dir, newTestClock().Now, newFakeSummarizer(), discardLog()) - require.NoError(t, err) - - defer func() { - require.NoError(t, memory.Close()) - }() - - second, err := memory.AppendOnce("user", first.Text, saved, func(int) error { - require.FailNow(t, "a recorded message must not be prepared again") - - return nil - }) - require.NoError(t, err) - require.Equal(t, first, second) - require.Equal(t, 1, memory.Len()) -} - -func TestQueuedAppendRetriesAnUnwrittenPreparedMessage(t *testing.T) { - memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog()) - require.NoError(t, err) - - defer func() { - require.NoError(t, memory.Close()) - }() - - failure := errors.New("fake queue write failure") - - var saved *int - - _, err = memory.AppendOnce("work", "[Garden] Check the gate.", nil, func(id int) error { - saved = &id - - return failure - }) - require.ErrorIs(t, err, failure) - require.Equal(t, 0, memory.Len()) - - message, err := memory.AppendOnce("work", "[Garden] Check the gate.", saved, func(id int) error { - require.Equal(t, *saved, id) - - return nil - }) - require.NoError(t, err) - require.Equal(t, 0, message.ID) - require.Equal(t, 1, memory.Len()) -} - -func TestQueuedAppendRefusesAnInvalidReceipt(t *testing.T) { - memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog()) - require.NoError(t, err) - - defer func() { - require.NoError(t, memory.Close()) - }() - - _, err = memory.Append("user", "Keep the original.") - require.NoError(t, err) - - testCases := []struct { - name string - id int - cause string - }{ - { - name: "receipt points to different text", - id: 0, - cause: "does not match", - }, - { - name: "receipt skips a log position", - id: 2, - cause: "beyond the log", - }, - { - name: "receipt has a negative position", - id: -1, - cause: "read queued message", - }, - } - for _, tc := range testCases { - t.Run(tc.name, func(t *testing.T) { - _, err := memory.AppendOnce("user", "Other text.", &tc.id, func(int) error { - require.FailNow(t, "an invalid receipt must not be prepared") - - return nil - }) - require.ErrorContains(t, err, tc.cause) - require.Equal(t, 1, memory.Len()) - }) - } -} diff --git a/memory_test.go b/memory_test.go index 202d4da..6713f6d 100644 --- a/memory_test.go +++ b/memory_test.go @@ -2,10 +2,13 @@ package main import ( "context" + "errors" "fmt" "maps" + "math/rand/v2" "os" "path/filepath" + "runtime" "slices" "strings" "testing" @@ -14,6 +17,15 @@ import ( "github.com/stretchr/testify/require" ) +const ( + // replayMessages is the length of the synthetic log. + replayMessages = 100000 + + // replayShortMessages is the length of the synthetic log with -short. + // It is long enough for batches of the chat view. + replayShortMessages = 5000 +) + func TestMemoryRestartLoadsTheSameView(t *testing.T) { root := t.TempDir() clock := newTestClock() @@ -389,7 +401,313 @@ func TestMemoryRefusesToRebuildAMissingView(t *testing.T) { require.ErrorIs(t, err, errMemoryNoView) } +func TestQueuedAppendResumesWithoutADuplicate(t *testing.T) { + dir := t.TempDir() + memory, err := openMemory(t.Context(), dir, newTestClock().Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + + var saved *int + + first, err := memory.AppendOnce("user", "Record the gate size.", nil, func(id int) error { + saved = &id + + require.Equal(t, 0, memory.log.Len()) + + return nil + }) + require.NoError(t, err) + require.NoError(t, memory.Close()) + + memory, err = openMemory(t.Context(), dir, newTestClock().Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + + defer func() { + require.NoError(t, memory.Close()) + }() + + second, err := memory.AppendOnce("user", first.Text, saved, func(int) error { + require.FailNow(t, "a recorded message must not be prepared again") + + return nil + }) + require.NoError(t, err) + require.Equal(t, first, second) + require.Equal(t, 1, memory.Len()) +} + +func TestQueuedAppendRetriesAnUnwrittenPreparedMessage(t *testing.T) { + memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + + defer func() { + require.NoError(t, memory.Close()) + }() + + failure := errors.New("fake queue write failure") + + var saved *int + + _, err = memory.AppendOnce("work", "[Garden] Check the gate.", nil, func(id int) error { + saved = &id + + return failure + }) + require.ErrorIs(t, err, failure) + require.Equal(t, 0, memory.Len()) + + message, err := memory.AppendOnce("work", "[Garden] Check the gate.", saved, func(id int) error { + require.Equal(t, *saved, id) + + return nil + }) + require.NoError(t, err) + require.Equal(t, 0, message.ID) + require.Equal(t, 1, memory.Len()) +} + +func TestQueuedAppendRefusesAnInvalidReceipt(t *testing.T) { + memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + + defer func() { + require.NoError(t, memory.Close()) + }() + + _, err = memory.Append("user", "Keep the original.") + require.NoError(t, err) + + testCases := []struct { + name string + id int + cause string + }{ + { + name: "receipt points to different text", + id: 0, + cause: "does not match", + }, + { + name: "receipt skips a log position", + id: 2, + cause: "beyond the log", + }, + { + name: "receipt has a negative position", + id: -1, + cause: "read queued message", + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + _, err := memory.AppendOnce("user", "Other text.", &tc.id, func(int) error { + require.FailNow(t, "an invalid receipt must not be prepared") + + return nil + }) + require.ErrorContains(t, err, tc.cause) + require.Equal(t, 1, memory.Len()) + }) + } +} + +// TestReplaySyntheticLog runs a synthetic log through the memory core with the +// fake summarizer. After each message it waits for the builds, as a turn does. +// It checks that each view stays within its sawtooth, that no view line holds a +// whole long message or a part of an unbuilt one, and that a restart in the +// middle and at the end loads the same views byte for byte. With -short, the +// log is shorter, and the test still requires a batch of the chat view. +func TestReplaySyntheticLog(t *testing.T) { + start := time.Now() + messages := replayMessages + if testing.Short() { + messages = replayShortMessages + } + + root := t.TempDir() + clock := newTestClock() + random := rand.New(rand.NewPCG(1, 2)) + memory, err := openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + + defer func() { + require.NoError(t, memory.Close()) + }() + + batches := 0 + + low := viewChatHigh + long := make([]bool, messages) + for id := range messages { + if id == messages/2 { + memory = replayRequireRestart(t, memory, root, clock) + } + + clock.now = clock.now.Add(time.Minute) + kind, text := replayMessage(random, id) + long[id] = len(kind)+len(": ")+len(text) > nodeBytes + lines := replayViewLines(memory) + _, err = memory.Append(kind, text) + require.NoError(t, err) + + size, merged := replayRequireViews(t, memory, lines, long, id%1000 == 0) + if merged { + batches++ + low = min(low, size) + } + + require.NoError(t, memory.Wait(t.Context())) + } + + memory = replayRequireRestart(t, memory, root, clock) + require.Equal(t, messages, memory.log.Len()) + require.Positive(t, batches) + + require.NoError(t, memory.Wait(t.Context())) + t.Logf("%d messages, %d nodes, %d batches in %s", messages, memory.tree.Count(), batches, time.Since(start).Round(time.Second)) + t.Logf("chat view: %d lines, %d bytes, lowest after a batch %d bytes", len(memory.chat.lines), memory.chat.Size(), low) + t.Logf("compaction view: %d lines, %d bytes", len(memory.cview.lines), memory.cview.Size()) + + runtime.GC() + + var stats runtime.MemStats + + runtime.ReadMemStats(&stats) + t.Logf("heap in use: %d KB", stats.HeapInuse/1024) +} + // memoryLongText returns a text that is too long for its own node. func memoryLongText(n int) string { return fmt.Sprintf("long %d ", n) + strings.Repeat("x", nodeBytes) } + +// replayRequireRestart closes memory, opens it again, and checks that the views +// and view.json stay the same byte for byte. +func replayRequireRestart(t *testing.T, memory *Memory, root string, clock *TestClock) *Memory { + t.Helper() + + require.NoError(t, memory.Wait(t.Context())) + + view := memory.View() + context := memory.Context(0, memory.log.Len()) + saved, err := os.ReadFile(filepath.Join(root, "view.json")) + require.NoError(t, err) + require.NoError(t, memory.Close()) + + memory, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog()) + require.NoError(t, err) + require.NoError(t, memory.Wait(t.Context())) + require.Equal(t, view, memory.View()) + require.Equal(t, context, memory.Context(0, memory.log.Len())) + + loaded, err := os.ReadFile(filepath.Join(root, "view.json")) + require.NoError(t, err) + require.Equal(t, saved, loaded) + + return memory +} + +// replayMessage returns the kind and the text of message id. Most messages are +// tool calls and their output. Users write short lines. One output in a hundred +// is over 30 KB. +func replayMessage(random *rand.Rand, id int) (string, string) { + roll := random.IntN(100) + switch { + case roll < 5: + return "user", replayText(id, 10+random.IntN(200)) + case roll < 10: + return "reply", replayText(id, 100+random.IntN(1500)) + case roll < 12: + return "note", replayText(id, 50+random.IntN(600)) + case roll < 13: + return "work", replayText(id, 300+random.IntN(3000)) + case roll < 56: + return "tool", replayText(id, 60+random.IntN(400)) + case roll < 57: + return "echo", replayText(id, 30000+random.IntN(30000)) + default: + return "echo", replayText(id, 20+random.IntN(4000)) + } +} + +// replayViewLines returns the number of lines in the chat view of memory. +func replayViewLines(memory *Memory) int { + memory.mutex.Lock() + defer memory.mutex.Unlock() + + return len(memory.chat.lines) +} + +// replayRequireViews checks the views after a new message, and with full +// compares them with the tree. Each view is at most its high limit, and a batch +// merges the chat view to its low limit. The leaf of the new message can be +// built after the append, so the sizes count it as the pending mark, as the +// append did. It returns the size of the chat view and if a batch merged it, +// from the number of lines before the message. +func replayRequireViews(t *testing.T, memory *Memory, lines int, long []bool, full bool) (int, bool) { + t.Helper() + + memory.mutex.Lock() + defer memory.mutex.Unlock() + + size := replayAppendSize(memory.chat) + require.LessOrEqual(t, size, viewChatHigh) + require.LessOrEqual(t, replayAppendSize(memory.cview), viewCompactionHigh) + + merged := len(memory.chat.lines) < lines + if merged { + require.LessOrEqual(t, size, viewChatLow) + } + + replayRequireOnlySummaries(t, memory, long, full) + + return size, merged +} + +// replayText returns a text of about size bytes that starts with its message ID +// and holds some non-ASCII characters. +func replayText(id, size int) string { + head := fmt.Sprintf("m%d ", id) + words := "ação data line café " + + return head + strings.Repeat(words, max(1, (size-len(head))/len(words))) +} + +// replayAppendSize returns the size of view with its last line as the pending +// mark. +func replayAppendSize(view *View) int { + last := view.lines[len(view.lines)-1] + return view.Size() - view.lineSize(last) + len(viewPending) +} + +// replayRequireOnlySummaries checks each line of the chat view as the model +// reads it. A built line holds its node, a long message has a summary of the +// fake summarizer and not its text, and an unbuilt line holds only the pending +// mark. With full, it also checks that the view keeps the node of each built +// line. The caller holds the lock of memory. +func replayRequireOnlySummaries(t *testing.T, memory *Memory, long []bool, full bool) { + t.Helper() + + rendered := strings.Split(memory.chat.Render(), "\n") + require.Len(t, rendered, len(memory.chat.lines)+2) + + for n, line := range memory.chat.lines { + head := fmt.Sprintf("%d+%d|", line.index< nodeBytes - lines := replayViewLines(memory) - _, err = memory.Append(kind, text) - require.NoError(t, err) - - size, merged := replayRequireViews(t, memory, lines, long, id%1000 == 0) - if merged { - batches++ - low = min(low, size) - } - - require.NoError(t, memory.Wait(t.Context())) - } - - memory = replayRequireRestart(t, memory, root, clock) - require.Equal(t, messages, memory.log.Len()) - require.Positive(t, batches) - - require.NoError(t, memory.Wait(t.Context())) - t.Logf("%d messages, %d nodes, %d batches in %s", messages, memory.tree.Count(), batches, time.Since(start).Round(time.Second)) - t.Logf("chat view: %d lines, %d bytes, lowest after a batch %d bytes", len(memory.chat.lines), memory.chat.Size(), low) - t.Logf("compaction view: %d lines, %d bytes", len(memory.cview.lines), memory.cview.Size()) - - runtime.GC() - - var stats runtime.MemStats - - runtime.ReadMemStats(&stats) - t.Logf("heap in use: %d KB", stats.HeapInuse/1024) -} - -// replayRequireRestart closes memory, opens it again, and checks that the views -// and view.json stay the same byte for byte. -func replayRequireRestart(t *testing.T, memory *Memory, root string, clock *TestClock) *Memory { - t.Helper() - - require.NoError(t, memory.Wait(t.Context())) - - view := memory.View() - context := memory.Context(0, memory.log.Len()) - saved, err := os.ReadFile(filepath.Join(root, "view.json")) - require.NoError(t, err) - require.NoError(t, memory.Close()) - - memory, err = openMemory(t.Context(), root, clock.Now, newFakeSummarizer(), discardLog()) - require.NoError(t, err) - require.NoError(t, memory.Wait(t.Context())) - require.Equal(t, view, memory.View()) - require.Equal(t, context, memory.Context(0, memory.log.Len())) - - loaded, err := os.ReadFile(filepath.Join(root, "view.json")) - require.NoError(t, err) - require.Equal(t, saved, loaded) - - return memory -} - -// replayMessage returns the kind and the text of message id. Most messages are -// tool calls and their output. Users write short lines. One output in a hundred -// is over 30 KB. -func replayMessage(random *rand.Rand, id int) (string, string) { - roll := random.IntN(100) - switch { - case roll < 5: - return "user", replayText(id, 10+random.IntN(200)) - case roll < 10: - return "reply", replayText(id, 100+random.IntN(1500)) - case roll < 12: - return "note", replayText(id, 50+random.IntN(600)) - case roll < 13: - return "work", replayText(id, 300+random.IntN(3000)) - case roll < 56: - return "tool", replayText(id, 60+random.IntN(400)) - case roll < 57: - return "echo", replayText(id, 30000+random.IntN(30000)) - default: - return "echo", replayText(id, 20+random.IntN(4000)) - } -} - -// replayViewLines returns the number of lines in the chat view of memory. -func replayViewLines(memory *Memory) int { - memory.mutex.Lock() - defer memory.mutex.Unlock() - - return len(memory.chat.lines) -} - -// replayRequireViews checks the views after a new message, and with full -// compares them with the tree. Each view is at most its high limit, and a batch -// merges the chat view to its low limit. The leaf of the new message can be -// built after the append, so the sizes count it as the pending mark, as the -// append did. It returns the size of the chat view and if a batch merged it, -// from the number of lines before the message. -func replayRequireViews(t *testing.T, memory *Memory, lines int, long []bool, full bool) (int, bool) { - t.Helper() - - memory.mutex.Lock() - defer memory.mutex.Unlock() - - size := replayAppendSize(memory.chat) - require.LessOrEqual(t, size, viewChatHigh) - require.LessOrEqual(t, replayAppendSize(memory.cview), viewCompactionHigh) - - merged := len(memory.chat.lines) < lines - if merged { - require.LessOrEqual(t, size, viewChatLow) - } - - replayRequireOnlySummaries(t, memory, long, full) - - return size, merged -} - -// replayText returns a text of about size bytes that starts with its message ID -// and holds some non-ASCII characters. -func replayText(id, size int) string { - head := fmt.Sprintf("m%d ", id) - words := "ação data line café " - - return head + strings.Repeat(words, max(1, (size-len(head))/len(words))) -} - -// replayAppendSize returns the size of view with its last line as the pending -// mark. -func replayAppendSize(view *View) int { - last := view.lines[len(view.lines)-1] - return view.Size() - view.lineSize(last) + len(viewPending) -} - -// replayRequireOnlySummaries checks each line of the chat view as the model -// reads it. A built line holds its node, a long message has a summary of the -// fake summarizer and not its text, and an unbuilt line holds only the pending -// mark. With full, it also checks that the view keeps the node of each built -// line. The caller holds the lock of memory. -func replayRequireOnlySummaries(t *testing.T, memory *Memory, long []bool, full bool) { - t.Helper() - - rendered := strings.Split(memory.chat.Render(), "\n") - require.Len(t, rendered, len(memory.chat.lines)+2) - - for n, line := range memory.chat.lines { - head := fmt.Sprintf("%d+%d|", line.index< 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()) - }) - } -} - -func TestQueuedWhatsAppBatchRecoversBeforeItsLogAppend(t *testing.T) { - testCases := []struct { - name string - id *int - }{ - { - name: "before log ID is stored", - id: nil, - }, - { - name: "after log ID is stored", - id: new(0), - }, - } - 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, - }}, - } - 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 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", - } - 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", - } - 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")) - - 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) - - _, 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 { - 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", - } - 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()) - } - - 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 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) -} diff --git a/whatsapp_test.go b/whatsapp_test.go index c1471e9..9ed4105 100644 --- a/whatsapp_test.go +++ b/whatsapp_test.go @@ -2,6 +2,7 @@ package main import ( "errors" + "log/slog" "os" "path/filepath" "strings" @@ -753,3 +754,8 @@ func TestSubagentHasNoWhatsAppTools(t *testing.T) { require.Equal(t, `error: unknown tool "whatsapp_find"`, box.run(t.Context(), "whatsapp_find", `{"name":"ana"}`)) require.Equal(t, `error: unknown tool "whatsapp_send"`, box.run(t.Context(), "whatsapp_send", `{"id":"a","chat":"15550002222@s.whatsapp.net","text":"hi"}`)) } + +// whatsappDiscardLog is a logger for a WhatsApp whose log a test does not read. +func whatsappDiscardLog() *slog.Logger { + return slog.New(slog.DiscardHandler) +}