X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/522d7fd117285e4e4ce3ea5ec9b5a1722521b681..8707a09fee94c97e29d668d24fb9713373913e0b:/memory_test.go?ds=inline diff --git a/memory_test.go b/memory_test.go index fe050c4..6f76e5b 100644 --- a/memory_test.go +++ b/memory_test.go @@ -2,11 +2,13 @@ package main import ( "context" + "errors" "fmt" - "log" "maps" + "math/rand/v2" "os" "path/filepath" + "runtime" "slices" "strings" "testing" @@ -15,6 +17,15 @@ import ( "github.com/stretchr/testify/require" ) +const ( + // memoryReplayMessages is the length of the synthetic log. + memoryReplayMessages = 100000 + + // memoryReplayShortMessages is the length of the synthetic log with + // -short. It is long enough for batches of the chat view. + memoryReplayShortMessages = 5000 +) + func TestMemoryRestartLoadsTheSameView(t *testing.T) { root := t.TempDir() clock := newTestClock() @@ -132,6 +143,7 @@ func TestMemoryRunsAtMostEightBuilds(t *testing.T) { slices.SortFunc(running, func(a, b NodeKey) int { return a.index - b.index }) + require.Equal(t, []NodeKey{{0, 0}, {0, 1}, {0, 2}, {0, 3}, {0, 4}, {0, 5}, {0, 6}, {0, 7}}, running) require.Eventually(t, func() bool { @@ -184,7 +196,7 @@ func TestMemorySettleFailsWhileABuildFailed(t *testing.T) { summarizer.SetFailing(true) logs := &SyncBuffer{} - memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, log.New(logs, "", 0)) + memory, err := openMemory(t.Context(), t.TempDir(), newTestClock().Now, summarizer, daemonTestLogger(logs)) require.NoError(t, err) defer func() { @@ -194,7 +206,7 @@ func TestMemorySettleFailsWhileABuildFailed(t *testing.T) { _, err = memory.Append("echo", memoryLongText(0)) require.NoError(t, err) require.ErrorIs(t, memory.Settle(t.Context()), errMemoryUnsettled) - require.Equal(t, "summary failed, retry at next message: node (0, 0): fake failure\n", logs.String()) + require.Equal(t, "<4>summary failed level=0 index=0: fake failure\n", logs.String()) summarizer.SetFailing(false) @@ -390,7 +402,314 @@ 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: "refuses when receipt points to different text", + id: 0, + cause: "does not match", + }, + { + name: "refuses when receipt skips a log position", + id: 2, + cause: "beyond the log", + }, + { + name: "refuses when 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 := memoryReplayMessages + if testing.Short() { + messages = memoryReplayShortMessages + } + + 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 = memoryReplayRequireRestart(t, memory, root, clock) + } + + clock.now = clock.now.Add(time.Minute) + kind, text := memoryReplayMessage(random, id) + long[id] = len(kind)+len(": ")+len(text) > nodeBytes + lines := memoryReplayViewLines(memory) + _, err = memory.Append(kind, text) + require.NoError(t, err) + + size, merged := memoryReplayRequireViews(t, memory, lines, long, id%1000 == 0) + if merged { + batches++ + low = min(low, size) + } + + require.NoError(t, memory.Wait(t.Context())) + } + + memory = memoryReplayRequireRestart(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) } + +// memoryReplayRequireRestart closes memory, opens it again, and checks that the +// views and view.json stay the same byte for byte. +func memoryReplayRequireRestart(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 +} + +// memoryReplayMessage 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 memoryReplayMessage(random *rand.Rand, id int) (string, string) { + roll := random.IntN(100) + switch { + case roll < 5: + return "user", memoryReplayText(id, 10+random.IntN(200)) + case roll < 10: + return "reply", memoryReplayText(id, 100+random.IntN(1500)) + case roll < 12: + return "note", memoryReplayText(id, 50+random.IntN(600)) + case roll < 13: + return "work", memoryReplayText(id, 300+random.IntN(3000)) + case roll < 56: + return "tool", memoryReplayText(id, 60+random.IntN(400)) + case roll < 57: + return "echo", memoryReplayText(id, 30000+random.IntN(30000)) + default: + return "echo", memoryReplayText(id, 20+random.IntN(4000)) + } +} + +// memoryReplayViewLines returns the number of lines in the chat view of memory. +func memoryReplayViewLines(memory *Memory) int { + memory.mutex.Lock() + defer memory.mutex.Unlock() + + return len(memory.chat.lines) +} + +// memoryReplayRequireViews 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 memoryReplayRequireViews(t *testing.T, memory *Memory, lines int, long []bool, full bool) (int, bool) { + t.Helper() + + memory.mutex.Lock() + defer memory.mutex.Unlock() + + size := memoryReplayAppendSize(memory.chat) + require.LessOrEqual(t, size, viewChatHigh) + require.LessOrEqual(t, memoryReplayAppendSize(memory.cview), viewCompactionHigh) + + merged := len(memory.chat.lines) < lines + if merged { + require.LessOrEqual(t, size, viewChatLow) + } + + memoryReplayRequireOnlySummaries(t, memory, long, full) + + return size, merged +} + +// memoryReplayText returns a text of about size bytes that starts with its +// message ID and holds some non-ASCII characters. +func memoryReplayText(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))) +} + +// memoryReplayAppendSize returns the size of view with its last line as the +// pending mark. +func memoryReplayAppendSize(view *View) int { + last := view.lines[len(view.lines)-1] + return view.Size() - view.lineSize(last) + len(viewPending) +} + +// memoryReplayRequireOnlySummaries 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 memoryReplayRequireOnlySummaries(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<