import (
"context"
+ "encoding/json"
"errors"
"net/http"
"os"
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)
+}
import (
"context"
+ "os"
"path/filepath"
"testing"
"time"
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
+}
+++ /dev/null
-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())
- })
- }
-}
import (
"context"
+ "errors"
"fmt"
"maps"
+ "math/rand/v2"
"os"
"path/filepath"
+ "runtime"
"slices"
"strings"
"testing"
"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()
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<<line.level, 1<<line.level)
+ node, built := memory.chat.nodes[line]
+ if full {
+ require.Equal(t, treeIsBuilt(t, memory.tree, line.level, line.index), built)
+ }
+
+ if !built {
+ require.Equal(t, 0, line.level)
+ require.Equal(t, head+viewPending, rendered[n+1])
+
+ continue
+ }
+
+ require.Equal(t, head+strings.ReplaceAll(node.Text, "\n", " "), rendered[n+1])
+
+ if line.level == 0 && long[line.index] {
+ require.True(t, strings.HasPrefix(node.Text, "summary "), "message %d", line.index)
+ }
+ }
+}
return records
}
-func modelReasoningItem(id, encrypted string) map[string]any {
- return map[string]any{
- "type": "reasoning",
- "id": id,
- "encrypted_content": encrypted,
- "summary": []any{},
- }
-}
-
func modelFunctionCallItem(id, callID, name, arguments string) map[string]any {
return map[string]any{
"type": "function_call",
+++ /dev/null
-package main
-
-import (
- "fmt"
- "math/rand/v2"
- "os"
- "path/filepath"
- "runtime"
- "strings"
- "testing"
- "time"
-
- "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
-)
-
-// 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)
-}
-
-// 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<<line.level, 1<<line.level)
- node, built := memory.chat.nodes[line]
- if full {
- require.Equal(t, treeIsBuilt(t, memory.tree, line.level, line.index), built)
- }
-
- if !built {
- require.Equal(t, 0, line.level)
- require.Equal(t, head+viewPending, rendered[n+1])
-
- continue
- }
-
- require.Equal(t, head+strings.ReplaceAll(node.Text, "\n", " "), rendered[n+1])
-
- if line.level == 0 && long[line.index] {
- require.True(t, strings.HasPrefix(node.Text, "summary "), "message %d", line.index)
- }
- }
-}
+++ /dev/null
-package main
-
-import (
- "os"
- "testing"
-
- "github.com/stretchr/testify/require"
-)
-
-// 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
-}
return test
}
+
+func modelReasoningItem(id, encrypted string) map[string]any {
+ return map[string]any{
+ "type": "reasoning",
+ "id": id,
+ "encrypted_content": encrypted,
+ "summary": []any{},
+ }
+}
import (
"context"
"fmt"
- "log/slog"
"os"
"path/filepath"
"slices"
require.NoError(t, container.Close())
}
-// whatsappDiscardLog is a logger for a WhatsApp whose log a test does not read.
-func whatsappDiscardLog() *slog.Logger {
- return slog.New(slog.DiscardHandler)
-}
-
// whatsappTestMessage returns the event of the text message id of sender in
// chat at the time at. A sender with whatsappTestNumber is the user.
func whatsappTestMessage(chat, sender types.JID, id, text string, at time.Time) *events.Message {
+++ /dev/null
-package main
-
-import (
- "encoding/json"
- "net/http"
- "os"
- "path/filepath"
- "testing"
- "time"
-
- "github.com/stretchr/testify/require"
-)
-
-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)
-}
import (
"errors"
+ "log/slog"
"os"
"path/filepath"
"strings"
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)
+}
projects / onechat / commitdiff