Send the compaction view in block messages

A compaction sends each whole block of four view lines as its own user
message, the first one with the opening tag, and one last user message
with the rest of the lines, the closing tag, and the task. The note of a
retry is one more user message after them. The blocks count from the
first line of the view, so two compactions whose views begin with the
same lines send the same block messages. Joined, the messages are the
same bytes as the one message before.

The backend reads a cache entry only at a message end of a request. With
the view in one message, a compaction read only the shared prompt and
tools, about 1,920 of 10,000 tokens. The first request of a turn sends
its view as messages that only grow and reads 74% to 85% from the cache,
so the compactions get the same kind of split.

Issue: 72c95ad

diff --git a/compactor.go b/compactor.go
index 5e73cfac30197ebf28ccb16dd2e30b140b8e3f24..e7a68664d8b138f0e3cda1dd391afccf1093845c 100644 (file)
--- a/compactor.go
+++ b/compactor.go
@@ -4,6 +4,7 @@ import (
        "context"
        "errors"
        "fmt"
        "context"
        "errors"
        "fmt"

+       "slices"

        "strings"
        "unicode/utf8"
 )
        "strings"
        "unicode/utf8"
 )


@@ -11,6 +12,9 @@ import (
 // compactorTries is the most calls for one node.
 const compactorTries = 5
 
 // compactorTries is the most calls for one node.
 const compactorTries = 5
 

+// compactorBlock is the number of view lines in one block message.
+const compactorBlock = 4
+

 // compactorRuler shows the model the size limit of a line. Models cannot count
 // bytes, and a sample line as the ruler gets its content copied.
 var compactorRuler = strings.Repeat("-", nodeBytes)
 // compactorRuler shows the model the size limit of a line. Models cannot count
 // bytes, and a sample line as the ruler gets its content copied.
 var compactorRuler = strings.Repeat("-", nodeBytes)


@@ -29,16 +33,22 @@ func newCompactor(client *ModelClient, instructions func() (string, error), mode
 
 // Compactor builds the nodes of one agent with a model. A compaction is a call
 // like a turn, with the same system prompt and tools, so it reads them from the
 
 // Compactor builds the nodes of one agent with a model. A compaction is a call
 // like a turn, with the same system prompt and tools, so it reads them from the

-// prompt cache of the turns. The tools are never called. After them comes one
-// user message with the compaction view up to the node and the task.
+// prompt cache of the turns. The tools are never called. After them come the
+// compaction view up to the node and the task, as the user messages that
+// compactorTexts gives. Each whole block of compactorBlock view lines is its
+// own message, and the last message holds the rest of the lines and the task.
+// The backend reads a cache entry only at a message end of a request, so a
+// compaction whose view begins with the same lines as an earlier one begins
+// with the same block messages.

 //
 //

-// When the line is over nodeBytes, the next try sends the same view and task,
-// plus a note with the size of the newest line and its cut at the limit. A try
-// never sends the model's earlier reply or reasoning back, so every try has the
-// same cached prefix and about the same size. After at most compactorTries
-// tries, the shortest line is the node. A few bytes over the limit is fine,
-// because the view measures the real sizes. A call that fails before any line
-// fails the build, and the core tries it again at the next message.
+// When the line is over nodeBytes, the next try sends the same messages, plus
+// one more user message with a note with the size of the newest line and its
+// cut at the limit. A try never sends the model's earlier reply or reasoning
+// back, so every try begins with the whole earlier try and has about the same
+// size. After at most compactorTries tries, the shortest line is the node. A
+// few bytes over the limit is fine, because the view measures the real sizes. A
+// call that fails before any line fails the build, and the core tries it again
+// at the next message.

 type Compactor struct {
        client       *ModelClient
        model        ModelConfig
 type Compactor struct {
        client       *ModelClient
        model        ModelConfig


@@ -48,7 +58,6 @@ type Compactor struct {
 
 // Summarize returns the line of the node of task.
 func (c *Compactor) Summarize(ctx context.Context, task Task) (string, error) {
 
 // Summarize returns the line of the node of task.
 func (c *Compactor) Summarize(ctx context.Context, task Task) (string, error) {

-       view := task.Context + "\n"

        text, err := compactorTask(task)
        if err != nil {
                return "", err
        text, err := compactorTask(task)
        if err != nil {
                return "", err


@@ -59,12 +68,17 @@ func (c *Compactor) Summarize(ctx context.Context, task Task) (string, error) {
                return "", fmt.Errorf("compact node (%d, %d): %w", task.Level, task.Index, err)
        }
 
                return "", fmt.Errorf("compact node (%d, %d): %w", task.Level, task.Index, err)
        }
 

+       var first []any
+       for _, message := range compactorTexts(task.Context, text) {
+               first = append(first, modelUserMessage(message))
+       }
+

        var lines []string
        note := ""
        for len(lines) < compactorTries {
        var lines []string
        note := ""
        for len(lines) < compactorTries {

-               parts := []string{view, text}
+               input := first

                if note != "" {
                if note != "" {

-                       parts = append(parts, note)
+                       input = append(slices.Clip(first), modelUserMessage(note))

                }
 
                var response ModelResponse
                }
 
                var response ModelResponse


@@ -73,7 +87,7 @@ func (c *Compactor) Summarize(ctx context.Context, task Task) (string, error) {
                        Effort:       c.model.Effort,
                        Instructions: instructions,
                        Tools:        c.tools,
                        Effort:       c.model.Effort,
                        Instructions: instructions,
                        Tools:        c.tools,

-                       Input:        []any{modelUserMessage(parts...)},
+                       Input:        input,

                        Node:         &NodeKey{task.Level, task.Index},
                })
                if err != nil {
                        Node:         &NodeKey{task.Level, task.Index},
                })
                if err != nil {


@@ -130,6 +144,26 @@ of them from there too.
        return "", fmt.Errorf("node (%d, %d) has %d inputs", task.Level, task.Index, len(task.Input))
 }
 
        return "", fmt.Errorf("node (%d, %d) has %d inputs", task.Level, task.Index, len(task.Input))
 }
 

+// compactorTexts returns the texts of the user messages of a first try for the
+// compaction view and the task text. Each whole block of compactorBlock lines,
+// counted from the first line of the view, is one message. The last message
+// holds the rest of the lines, the closing tag, a newline, and the task. The
+// first message starts with the opening tag. Joined, the texts are the view, a
+// newline, and the task.
+func compactorTexts(view, task string) []string {
+       body := strings.TrimSuffix(strings.TrimPrefix(view, "<chat>\n"), "</chat>")
+       lines := strings.SplitAfter(body, "\n")
+       lines = lines[:len(lines)-1]
+       whole := len(lines) - len(lines)%compactorBlock
+
+       var texts []string
+       for n := 0; n < whole; n += compactorBlock {
+               texts = append(texts, viewMessage(n == 0, lines[n:n+compactorBlock]))
+       }
+
+       return append(texts, viewMessage(whole == 0, lines[whole:])+"</chat>\n"+task)
+}
+

 // compactorTooLong returns the note for a line over nodeBytes: its size and its
 // first nodeBytes bytes, which show the model where the limit cuts it.
 func compactorTooLong(line string) string {
 // compactorTooLong returns the note for a line over nodeBytes: its size and its
 // first nodeBytes bytes, which show the model where the limit cuts it.
 func compactorTooLong(line string) string {

diff --git a/compactor_test.go b/compactor_test.go
index a128bc7ce7b2caa3ce7967273b7a02beb8a13a89..0d591ee9e5602135ef113712e405bddbe3c11bb8 100644 (file)
--- a/compactor_test.go
+++ b/compactor_test.go
@@ -40,14 +40,98 @@ func TestCompactorSendsThePromptToolsAndTask(t *testing.T) {
        require.NoError(t, err)
        require.JSONEq(t, string(tools), string(sent))
 
        require.NoError(t, err)
        require.JSONEq(t, string(tools), string(sent))
 

-       parts := compactorInputParts(t, server.Call(0))
+       texts := compactorInputTexts(t, server.Call(0))

        require.Equal(t, []string{
        require.Equal(t, []string{

-               "<chat>\n0+4|user: earlier\n</chat>\n",
-               "Compaction: compress message 7 into one line of at most 512 bytes\n" +
+               "<chat>\n0+4|user: earlier\n</chat>\n" +
+                       "Compaction: compress message 7 into one line of at most 512 bytes\n" +

                        "(about 70 words), the length of this ruler:\n" +
                        strings.Repeat("-", 512) + "\n" +
                        "<input>\nuser: " + strings.Repeat("x", 600) + "\n</input>",
                        "(about 70 words), the length of this ruler:\n" +
                        strings.Repeat("-", 512) + "\n" +
                        "<input>\nuser: " + strings.Repeat("x", 600) + "\n</input>",

-       }, parts)
+       }, texts)
+}
+
+func TestCompactorSendsEachWholeBlockOfTheViewAsAMessage(t *testing.T) {
+       testCases := []struct {
+               name    string
+               context string
+               want    []string
+       }{
+               {
+                       name:    "empty view goes with the task",
+                       context: "<chat>\n</chat>",
+                       want:    []string{"<chat>\n</chat>\nTASK"},
+               },
+               {
+                       name:    "lines short of a block go with the task",
+                       context: "<chat>\n0+1|a\n1+1|b\n2+1|c\n</chat>",
+                       want:    []string{"<chat>\n0+1|a\n1+1|b\n2+1|c\n</chat>\nTASK"},
+               },
+               {
+                       name:    "one whole block and the task",
+                       context: "<chat>\n0+1|a\n1+1|b\n2+1|c\n3+1|d\n</chat>",
+                       want:    []string{"<chat>\n0+1|a\n1+1|b\n2+1|c\n3+1|d\n", "</chat>\nTASK"},
+               },
+               {
+                       name:    "two whole blocks and the rest with the task",
+                       context: "<chat>\n0+2|a\n2+1|b\n3+1|c\n4+1|d\n5+1|e\n6+1|f\n7+1|g\n8+1|h\n9+1|i\n</chat>",
+                       want: []string{
+                               "<chat>\n0+2|a\n2+1|b\n3+1|c\n4+1|d\n",
+                               "5+1|e\n6+1|f\n7+1|g\n8+1|h\n",
+                               "9+1|i\n</chat>\nTASK",
+                       },
+               },
+       }
+
+       for _, tc := range testCases {
+               t.Run(tc.name, func(t *testing.T) {
+                       server := newModelFake(t)
+                       task := compactorLeafTask()
+                       task.Context = tc.context
+
+                       _, err := newTestCompactor(t, server).Summarize(t.Context(), task)
+                       require.NoError(t, err)
+
+                       text, err := compactorTask(task)
+                       require.NoError(t, err)
+
+                       want := make([]string, len(tc.want))
+                       for n, message := range tc.want {
+                               want[n] = strings.Replace(message, "TASK", text, 1)
+                       }
+
+                       texts := compactorInputTexts(t, server.Call(0))
+                       require.Equal(t, want, texts)
+                       require.Equal(t, tc.context+"\n"+text, strings.Join(texts, ""))
+               })
+       }
+}
+
+func TestCompactorSendsTheSameBlocksForViewsWithTheSameFirstLines(t *testing.T) {
+       shared := "<chat>\n0+4|a\n4+1|b\n5+1|c\n6+1|d\n7+1|e\n8+1|f\n9+1|g\n10+1|h\n"
+       server := newModelFake(t)
+       compactor := newTestCompactor(t, server)
+       _, err := compactor.Summarize(t.Context(), Task{
+               Level:   0,
+               Index:   11,
+               Context: shared + "11+1|i\n</chat>",
+               Input:   []string{"user: eleven"},
+       })
+       require.NoError(t, err)
+
+       _, err = compactor.Summarize(t.Context(), Task{
+               Level:   1,
+               Index:   5,
+               Context: shared + "11+1|j\n12+1|k\n</chat>",
+               Input:   []string{"left line", "right line"},
+       })
+       require.NoError(t, err)
+
+       first := compactorInputTexts(t, server.Call(0))
+       second := compactorInputTexts(t, server.Call(1))
+       require.Len(t, first, 3)
+       require.Len(t, second, 3)
+       require.Equal(t, first[:2], second[:2])
+       require.NotEqual(t, first[2], second[2])

 }
 
 func TestCompactorWritesTheMergeTask(t *testing.T) {
 }
 
 func TestCompactorWritesTheMergeTask(t *testing.T) {


@@ -72,8 +156,11 @@ func TestCompactorRetriesALongLineWithTheNewestNote(t *testing.T) {
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(long[call.number])}, 100, 0, 5))
        }
 
                modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem(long[call.number])}, 100, 0, 5))
        }
 

+       task := compactorLeafTask()
+       task.Context = "<chat>\n0+1|a\n1+1|b\n2+1|c\n3+1|d\n4+1|e\n</chat>"
+

        compactor := newTestCompactor(t, server)
        compactor := newTestCompactor(t, server)

-       line, err := compactor.Summarize(t.Context(), compactorLeafTask())
+       line, err := compactor.Summarize(t.Context(), task)

        require.NoError(t, err)
        require.Equal(t, long[2], line)
        require.Equal(t, 3, server.Calls())
        require.NoError(t, err)
        require.Equal(t, long[2], line)
        require.Equal(t, 3, server.Calls())


@@ -85,20 +172,17 @@ func TestCompactorRetriesALongLineWithTheNewestNote(t *testing.T) {
                require.Equal(t, len(long[n]), record.TextBytes)
        }
 
                require.Equal(t, len(long[n]), record.TextBytes)
        }
 

-       first := compactorInputParts(t, server.Call(0))
+       first := compactorInputTexts(t, server.Call(0))

        require.Len(t, first, 2)
        for n, want := range []string{long[0], long[1]} {
                call := server.Call(n + 1)
        require.Len(t, first, 2)
        for n, want := range []string{long[0], long[1]} {
                call := server.Call(n + 1)

-               parts := compactorInputParts(t, call)
-               require.Equal(t, first, parts[:2])
-               require.Equal(t, []string{compactorTooLong(want)}, parts[2:])
-               require.Contains(t, parts[2], "was "+strconv.Itoa(len(want))+" bytes,")
-               require.Contains(t, parts[2], "\n"+want[:512]+"| <- LIMIT")
-
-               note, err := json.Marshal(map[string]string{
-                       "type": "input_text",
-                       "text": parts[2],
-               })
+               texts := compactorInputTexts(t, call)
+               require.Equal(t, first, texts[:2])
+               require.Equal(t, []string{compactorTooLong(want)}, texts[2:])
+               require.Contains(t, texts[2], "was "+strconv.Itoa(len(want))+" bytes,")
+               require.Contains(t, texts[2], "\n"+want[:512]+"| <- LIMIT")
+
+               note, err := json.Marshal(modelUserMessage(texts[2]))

                require.NoError(t, err)
                require.LessOrEqual(t, len(call.raw), len(server.Call(0).raw)+len(",")+len(note))
        }
                require.NoError(t, err)
                require.LessOrEqual(t, len(call.raw), len(server.Call(0).raw)+len(",")+len(note))
        }


@@ -204,8 +288,9 @@ func newTestCompactor(t *testing.T, server *ModelFake) *Compactor {
        })
 }
 
        })
 }
 

-// compactorInputParts returns the texts of the one user message of call.
-func compactorInputParts(t *testing.T, call ModelFakeCall) []string {
+// compactorInputTexts returns the text of each user message of call. Each
+// message has one text part.
+func compactorInputTexts(t *testing.T, call ModelFakeCall) []string {

        t.Helper()
 
        var body struct {
        t.Helper()
 
        var body struct {


@@ -219,13 +304,13 @@ func compactorInputParts(t *testing.T, call ModelFakeCall) []string {
        }
 
        require.NoError(t, json.Unmarshal(call.raw, &body))
        }
 
        require.NoError(t, json.Unmarshal(call.raw, &body))

-       require.Len(t, body.Input, 1)
-       require.Equal(t, "user", body.Input[0].Role)

 
 

-       texts := make([]string, len(body.Input[0].Content))
-       for n, part := range body.Input[0].Content {
-               require.Equal(t, "input_text", part.Type)
-               texts[n] = part.Text
+       texts := make([]string, len(body.Input))
+       for n, message := range body.Input {
+               require.Equal(t, "user", message.Role)
+               require.Len(t, message.Content, 1)
+               require.Equal(t, "input_text", message.Content[0].Type)
+               texts[n] = message.Content[0].Text

        }
 
        return texts
        }
 
        return texts