Lay out turn requests for the prompt cache

The ChatGPT backend writes a cache entry only at the end of the latest
message of a request, and a later request reads it only at one of its
own message ends. The first request of a turn now puts the stable parts
first and the per-turn state last: the prompt and the tools, then the
view as user messages, then a message with the end of the view and the
date, then each new message.

The view messages only grow between batches. layout.json in the agent's
directory keeps the line count and a hash of each view message sent. The
next request sends the same messages while their lines give the same
text, and the new lines follow as one new message. At the first message
that differs, such as after a batch merge, the rest of the view becomes
one new message.

After the turns, when the builds are done and no message waits, a
warm-up request sends only the prompt, the tools, and the view messages.
It ends at the view, where the next turn has a message end, so that turn
can read the view from the cache. The warm-up stops reading at its first
output item, after the input is read, to save output tokens, and its
output is never logged or used. Each usage record names the request of a
turn: first, next, or warmup.

Issue: 5daf23e

diff --git a/layout.go b/layout.go
new file mode 100644 (file)
index 0000000..a3f9de0
--- /dev/null
+++ b/layout.go
@@ -0,0 +1,123 @@
+package main
+
+import (
+       "crypto/sha256"
+       "encoding/hex"
+       "encoding/json"
+       "errors"
+       "fmt"
+       "os"
+       "strings"
+)
+
+// layoutFileName is the file of the view layout in the directory of an agent.
+const layoutFileName = "layout.json"
+
+// viewLayout splits the view of a turn request into user messages that only
+// grow between batches. The ChatGPT backend writes a cache entry only at the
+// end of the latest message of a request, and a later request reads it only at
+// one of its own message ends. So the view goes out as the same messages as in
+// the request before, while their lines give the same text, and the new lines
+// follow as one new message. Then the end of each earlier view stays a message
+// end, and a warm-up entry there stays readable when the view grows.
+//
+// The first message starts with "<chat>\n". The closing tag goes into the
+// state message after the view, so that no view message changes when lines
+// come. The layout keeps the line count and a hash of the text of each message
+// in layout.json, so it outlives a restart.
+type viewLayout struct {
+       path     string
+       Messages []layoutMessage `json:"messages"`
+}
+
+// layoutMessage is one view message: its number of lines and the hash of its
+// text.
+type layoutMessage struct {
+       Lines int    `json:"lines"`
+       Hash  string `json:"hash"`
+}
+
+// loadLayout reads the layout at path. A missing file gives an empty layout.
+func loadLayout(path string) (*viewLayout, error) {
+       layout := &viewLayout{
+               path: path,
+       }
+
+       data, err := os.ReadFile(path)
+       if errors.Is(err, os.ErrNotExist) {
+               return layout, nil
+       }
+
+       if err != nil {
+               return nil, fmt.Errorf("read view layout: %w", err)
+       }
+
+       err = json.Unmarshal(data, layout)
+       if err != nil {
+               return nil, fmt.Errorf("decode view layout: %w", err)
+       }
+
+       return layout, nil
+}
+
+// Split returns the view messages for lines and saves the layout. It keeps the
+// earlier messages up to the first one whose text differs, and puts the rest
+// of the lines into one new message.
+func (l *viewLayout) Split(lines []string) ([]string, error) {
+       var texts []string
+       var kept []layoutMessage
+       start := 0
+       for _, message := range l.Messages {
+               if start+message.Lines > len(lines) {
+                       break
+               }
+
+               text := viewMessage(len(texts) == 0, lines[start:start+message.Lines])
+               if textHash(text) != message.Hash {
+                       break
+               }
+
+               texts = append(texts, text)
+               kept = append(kept, message)
+               start += message.Lines
+       }
+
+       if start < len(lines) || len(texts) == 0 {
+               text := viewMessage(len(texts) == 0, lines[start:])
+               texts = append(texts, text)
+               kept = append(kept, layoutMessage{
+                       Lines: len(lines) - start,
+                       Hash:  textHash(text),
+               })
+       }
+
+       l.Messages = kept
+       data, err := json.Marshal(l)
+       if err != nil {
+               return nil, fmt.Errorf("encode view layout: %w", err)
+       }
+
+       err = writeFileAtomic(l.path, append(data, '\n'))
+       if err != nil {
+               return nil, fmt.Errorf("save view layout: %w", err)
+       }
+
+       return texts, nil
+}
+
+// viewMessage returns the text of a view message with lines. The first one
+// opens the chat.
+func viewMessage(first bool, lines []string) string {
+       text := strings.Join(lines, "")
+       if first {
+               text = "<chat>\n" + text
+       }
+
+       return text
+}
+
+// textHash returns a short hash of text.
+func textHash(text string) string {
+       sum := sha256.Sum256([]byte(text))
+       return hex.EncodeToString(sum[:16])
+}

diff --git a/layout_test.go b/layout_test.go
new file mode 100644 (file)
index 0000000..3fab197
--- /dev/null
+++ b/layout_test.go
@@ -0,0 +1,56 @@
+package main
+
+import (
+       "path/filepath"
+       "testing"
+
+       "github.com/stretchr/testify/require"
+)
+
+func TestViewLayoutKeepsTheMessagesItSent(t *testing.T) {
+       path := filepath.Join(t.TempDir(), layoutFileName)
+       layout, err := loadLayout(path)
+       require.NoError(t, err)
+
+       texts, err := layout.Split(nil)
+       require.NoError(t, err)
+       require.Equal(t, []string{"<chat>\n"}, texts)
+
+       texts, err = layout.Split([]string{"0+1|a\n", "1+1|b\n"})
+       require.NoError(t, err)
+       require.Equal(t, []string{"<chat>\n", "0+1|a\n1+1|b\n"}, texts)
+
+       // The same lines give the same messages, also after a restart.
+       layout, err = loadLayout(path)
+       require.NoError(t, err)
+
+       texts, err = layout.Split([]string{"0+1|a\n", "1+1|b\n"})
+       require.NoError(t, err)
+       require.Equal(t, []string{"<chat>\n", "0+1|a\n1+1|b\n"}, texts)
+
+       texts, err = layout.Split([]string{"0+1|a\n", "1+1|b\n", "2+1|c\n"})
+       require.NoError(t, err)
+       require.Equal(t, []string{"<chat>\n", "0+1|a\n1+1|b\n", "2+1|c\n"}, texts)
+}
+
+func TestViewLayoutStartsAgainAtAChangedMessage(t *testing.T) {
+       layout, err := loadLayout(filepath.Join(t.TempDir(), layoutFileName))
+       require.NoError(t, err)
+
+       _, err = layout.Split([]string{"0+1|a\n"})
+       require.NoError(t, err)
+
+       _, err = layout.Split([]string{"0+1|a\n", "1+1|b\n", "2+1|c\n"})
+       require.NoError(t, err)
+
+       // A batch merges lines 1 and 2, so the second message changes, and the
+       // rest becomes one new message.
+       texts, err := layout.Split([]string{"0+1|a\n", "1+2|b c\n", "3+1|d\n"})
+       require.NoError(t, err)
+       require.Equal(t, []string{"<chat>\n0+1|a\n", "1+2|b c\n3+1|d\n"}, texts)
+
+       // A view that lost lines starts again with one message.
+       texts, err = layout.Split([]string{"0+4|all\n"})
+       require.NoError(t, err)
+       require.Equal(t, []string{"<chat>\n0+4|all\n"}, texts)
+}

diff --git a/memory.go b/memory.go
index d7fff4ee1d6e6fef5a7c84919dd8605ee5fd6beb..a3665a1d64a57d91e9b4255178b09dba63e66c8c 100644 (file)
--- a/memory.go
+++ b/memory.go
@@ -543,13 +543,13 @@ func (m *Memory) View() string {
        return m.chat.Render()
 }
 
        return m.chat.Render()
 }
 

-// ViewWithout returns the chat view without the leaf lines of the messages in
-// skip.
-func (m *Memory) ViewWithout(skip map[int]bool) string {
+// ViewLines returns the lines of the chat view, each with its newline, without
+// the leaf lines of the messages in skip.
+func (m *Memory) ViewLines(skip map[int]bool) []string {

        m.mutex.Lock()
        defer m.mutex.Unlock()
 
        m.mutex.Lock()
        defer m.mutex.Unlock()
 

-       return m.chat.RenderWithout(skip)
+       return m.chat.LinesWithout(skip)

 }
 
 // Len returns the number of messages in the log.
 }
 
 // Len returns the number of messages in the log.

diff --git a/responses.go b/responses.go
index 3055e706eb985c06f7b0a593877d3cd8f1aa3f24..19ab6a1eda6bb54043f34f09c5eed89485bd8258 100644 (file)
--- a/responses.go
+++ b/responses.go
@@ -73,8 +73,10 @@ type Client struct {
 
 // Request is one model call. Input holds the input items in order, such as
 // userMessage values or output items of an earlier response. Node names the
 
 // Request is one model call. Input holds the input items in order, such as
 // userMessage values or output items of an earlier response. Node names the

-// node that a compaction builds, or is nil. It goes only to the usage file,
-// never to the endpoint.
+// node that a compaction builds, or is nil. Turn names the request of a turn:
+// first, next, or warmup. Node and Turn go only to the usage file, never to the
+// endpoint. A call with StopAtOutput ends at the first output item, after the
+// endpoint has read the whole input, and gives no output and no usage.

 type Request struct {
        Model        string
        Effort       string
 type Request struct {
        Model        string
        Effort       string


@@ -82,6 +84,8 @@ type Request struct {
        Tools        []Tool
        Input        []any
        Node         *nodeKey
        Tools        []Tool
        Input        []any
        Node         *nodeKey

+       Turn         string
+       StopAtOutput bool

 }
 
 // Tool is the definition of a function tool.
 }
 
 // Tool is the definition of a function tool.


@@ -258,7 +262,8 @@ func (c *Client) Create(ctx context.Context, request Request) (Response, error)
                record.Node = &[2]int{request.Node.level, request.Node.index}
        }
 
                record.Node = &[2]int{request.Node.level, request.Node.index}
        }
 

-       response, err := c.send(httpRequest, &record)
+       record.Turn = request.Turn
+       response, err := c.send(httpRequest, &record, request.StopAtOutput)

        if err != nil {
                record.Error = err.Error()
        }
        if err != nil {
                record.Error = err.Error()
        }


@@ -274,7 +279,7 @@ func (c *Client) Create(ctx context.Context, request Request) (Response, error)
 
 // send makes the HTTP call and reads its stream. It puts the status and the
 // quota headers into record.
 
 // send makes the HTTP call and reads its stream. It puts the status and the
 // quota headers into record.

-func (c *Client) send(httpRequest *http.Request, record *usageRecord) (response Response, err error) {
+func (c *Client) send(httpRequest *http.Request, record *usageRecord, stopAtOutput bool) (response Response, err error) {

        httpResponse, err := c.http.Do(httpRequest)
        if err != nil {
                return Response{}, fmt.Errorf("send model call: %w", err)
        httpResponse, err := c.http.Do(httpRequest)
        if err != nil {
                return Response{}, fmt.Errorf("send model call: %w", err)


@@ -293,7 +298,7 @@ func (c *Client) send(httpRequest *http.Request, record *usageRecord) (response
                return Response{}, errors.Join(err, readErr)
        }
 
                return Response{}, errors.Join(err, readErr)
        }
 

-       return readStream(httpResponse.Body)
+       return readStream(httpResponse.Body, stopAtOutput)

 }
 
 // usageLog appends one JSON line for each model call to usage.jsonl. Calls can
 }
 
 // usageLog appends one JSON line for each model call to usage.jsonl. Calls can


@@ -306,13 +311,14 @@ type usageLog struct {
 // usageRecord is one line of usage.jsonl. Date is the start of the call, and
 // Seconds its length. Node is the node [l, i] of a compaction. Status is the
 // HTTP status, or 0 when no response came. TextBytes is the size of the text of
 // usageRecord is one line of usage.jsonl. Date is the start of the call, and
 // Seconds its length. Node is the node [l, i] of a compaction. Status is the
 // HTTP status, or 0 when no response came. TextBytes is the size of the text of

-// the reply. Quota holds the x-codex-* headers of the response, without the
+// the reply. Turn names the request of a turn. Quota holds the x-codex-* headers of the response, without the

 // turn state.
 type usageRecord struct {
        Date         string            `json:"date"`
        Seconds      float64           `json:"seconds"`
        Model        string            `json:"model"`
        Node         *[2]int           `json:"node,omitempty"`
 // turn state.
 type usageRecord struct {
        Date         string            `json:"date"`
        Seconds      float64           `json:"seconds"`
        Model        string            `json:"model"`
        Node         *[2]int           `json:"node,omitempty"`

+       Turn         string            `json:"turn,omitempty"`

        Status       int               `json:"status"`
        InputTokens  int               `json:"input_tokens"`
        CachedTokens int               `json:"cached_tokens"`
        Status       int               `json:"status"`
        InputTokens  int               `json:"input_tokens"`
        CachedTokens int               `json:"cached_tokens"`


@@ -400,8 +406,9 @@ func quotaHeaders(header http.Header) map[string]string {
 
 // readStream reads the server-sent events of a call until the response
 // completes. The output items come from the output_item.done events, or from
 
 // readStream reads the server-sent events of a call until the response
 // completes. The output items come from the output_item.done events, or from

-// the completed response when the stream sent none.
-func readStream(stream io.Reader) (Response, error) {
+// the completed response when the stream sent none. With stopAtOutput it ends
+// at the first output item.
+func readStream(stream io.Reader, stopAtOutput bool) (Response, error) {

        var response Response
        reader := bufio.NewReader(stream)
        var data bytes.Buffer
        var response Response
        reader := bufio.NewReader(stream)
        var data bytes.Buffer


@@ -419,7 +426,7 @@ func readStream(stream io.Reader) (Response, error) {
                                data.WriteByte('\n')
                        }
                } else if data.Len() > 0 {
                                data.WriteByte('\n')
                        }
                } else if data.Len() > 0 {

-                       done, eventErr := readEvent(&response, data.Bytes())
+                       done, eventErr := readEvent(&response, data.Bytes(), stopAtOutput)

                        data.Reset()
                        if eventErr != nil || done {
                                return response, eventErr
                        data.Reset()
                        if eventErr != nil || done {
                                return response, eventErr


@@ -434,7 +441,7 @@ func readStream(stream io.Reader) (Response, error) {
 
 // readEvent adds one stream event to response, and tells if it ends the
 // stream.
 
 // readEvent adds one stream event to response, and tells if it ends the
 // stream.

-func readEvent(response *Response, data []byte) (bool, error) {
+func readEvent(response *Response, data []byte, stopAtOutput bool) (bool, error) {

        var event streamEvent
        err := json.Unmarshal(data, &event)
        if err != nil {
        var event streamEvent
        err := json.Unmarshal(data, &event)
        if err != nil {


@@ -442,6 +449,8 @@ func readEvent(response *Response, data []byte) (bool, error) {
        }
 
        switch event.Type {
        }
 
        switch event.Type {

+       case "response.output_item.added":
+               return stopAtOutput, nil

        case "response.output_item.done":
                response.Output = append(response.Output, event.Item)
        case "response.completed":
        case "response.output_item.done":
                response.Output = append(response.Output, event.Item)
        case "response.completed":

diff --git a/responses_test.go b/responses_test.go
index 9a1355a8d0098bb2ee18b01a1d0b9e3d0a8d665e..3ff327781adc83451074dec3d028dda86c55c2cd 100644 (file)
--- a/responses_test.go
+++ b/responses_test.go
@@ -197,6 +197,28 @@ func TestClientRecordsEachCall(t *testing.T) {
        require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
 }
 
        require.Equal(t, os.FileMode(0o600), info.Mode().Perm())
 }
 

+func TestClientStopsAtTheFirstOutputItem(t *testing.T) {
+       server := newFakeResponses(t)
+       server.reply = func(w http.ResponseWriter, call fakeCall) {
+               writeEvents(w,
+                       map[string]any{"type": "response.created"},
+                       map[string]any{"type": "response.output_item.added", "item": messageItem("")},
+               )
+       }
+
+       client := newTestClient(t, server)
+       request := testRequest()
+       request.StopAtOutput = true
+       request.Turn = "warmup"
+       response, err := client.Create(t.Context(), request)
+       require.NoError(t, err)
+       require.Empty(t, response.Output)
+
+       request.StopAtOutput = false
+       _, err = client.Create(t.Context(), request)
+       require.ErrorContains(t, err, "event stream ended before the response completed")
+}
+

 func TestClientRefusesCallsOverTheLimit(t *testing.T) {
        server := newFakeResponses(t)
        client := newTestClient(t, server)
 func TestClientRefusesCallsOverTheLimit(t *testing.T) {
        server := newFakeResponses(t)
        client := newTestClient(t, server)

diff --git a/turn.go b/turn.go
index 76f524eefc5fff41012cb391fc144850d9cede2a..91723d2f954c1b61a2ebc0c1fab951798de5bdfb 100644 (file)
--- a/turn.go
+++ b/turn.go
@@ -5,6 +5,7 @@ import (
        "encoding/json"
        "errors"
        "fmt"
        "encoding/json"
        "errors"
        "fmt"

+       "path/filepath"

        "strings"
        "sync"
        "time"
        "strings"
        "sync"
        "time"


@@ -30,6 +31,13 @@ type Event struct {
 // input of turnInput. The loop runs the tool calls of each response and sends
 // their results back, until a response has no tool call: the final answer.
 //
 // input of turnInput. The loop runs the tool calls of each response and sends
 // their results back, until a response has no tool call: the final answer.
 //

+// The view goes out as the messages of a viewLayout. After the turns, when no
+// message waits, a warm-up request sends only the prompt, the tools, and the
+// view messages, and stops at its first output item. It writes a cache entry
+// at the end of the view, where the next turn has a message end, so the first
+// request of that turn reads the view from the cache. The warm-up is not
+// logged, and its output is not used.
+//

 // Each later request of a turn begins with the whole input of the request
 // before it, then its output items, then the tool results. The output items
 // hold the reasoning with its encrypted content, because the calls do not
 // Each later request of a turn begins with the whole input of the request
 // before it, then its output items, then the tool results. The output items
 // hold the reasoning with its encrypted content, because the calls do not


@@ -42,6 +50,8 @@ type loop struct {
        memory *Memory
        client modelClient
        tools  *toolbox
        memory *Memory
        client modelClient
        tools  *toolbox

+       layout *viewLayout
+       warmUp bool

        clock  func() time.Time
        notify func(Event)
        wake   chan struct{}
        clock  func() time.Time
        notify func(Event)
        wake   chan struct{}


@@ -66,13 +76,21 @@ type functionCallOutput struct {
        Output string `json:"output"`
 }
 
        Output string `json:"output"`
 }
 

-// newLoop returns the loop of agent with its memory. Its calls go through
-// client, and notify gets each event. The clock gives the date of a turn.
-func newLoop(agent *Agent, memory *Memory, client modelClient, clock func() time.Time, notify func(Event)) *loop {
+// newLoop returns the loop of agent with its memory, with the warm-up on. Its
+// calls go through client, and notify gets each event. The clock gives the
+// date of a turn.
+func newLoop(agent *Agent, memory *Memory, client modelClient, clock func() time.Time, notify func(Event)) (*loop, error) {
+       layout, err := loadLayout(filepath.Join(agent.dir, layoutFileName))
+       if err != nil {
+               return nil, err
+       }
+

        l := &loop{
                agent:  agent,
                memory: memory,
                client: client,
        l := &loop{
                agent:  agent,
                memory: memory,
                client: client,

+               layout: layout,
+               warmUp: true,

                clock:  clock,
                notify: notify,
                wake:   make(chan struct{}, 1),
                clock:  clock,
                notify: notify,
                wake:   make(chan struct{}, 1),


@@ -84,7 +102,7 @@ func newLoop(agent *Agent, memory *Memory, client modelClient, clock func() time
                workdir: agent.config.Workdir,
        }
 
                workdir: agent.config.Workdir,
        }
 

-       return l
+       return l, nil

 }
 
 // Submit logs a user message and wakes the loop. The message joins the next
 }
 
 // Submit logs a user message and wakes the loop. The message joins the next


@@ -144,15 +162,24 @@ func (l *loop) Flush(ctx context.Context) error {
        return l.memory.Flush(ctx)
 }
 
        return l.memory.Flush(ctx)
 }
 

-// Run runs a turn each time user messages wait, until ctx ends. It returns the
-// error of a failed log write, which stops the agent.
+// Run runs a turn each time user messages wait, and a warm-up after the turns,
+// until ctx ends. It returns the error of a failed log write, which stops the
+// agent. A failed warm-up only costs the cache of the next turn, so its error
+// stays in the usage file.

 func (l *loop) Run(ctx context.Context) error {
        for {
 func (l *loop) Run(ctx context.Context) error {
        for {

+               turned := false

                for l.waiting() {
                        err := l.turn(ctx)
                        if err != nil {
                                return err
                        }
                for l.waiting() {
                        err := l.turn(ctx)
                        if err != nil {
                                return err
                        }

+
+                       turned = true
+               }
+
+               if turned && l.warmUp {
+                       _ = l.warm(ctx)

                }
 
                select {
                }
 
                select {


@@ -246,14 +273,20 @@ func (l *loop) chain(ctx context.Context) error {
                skip[message.ID] = true
        }
 
                skip[message.ID] = true
        }
 

-       input := turnInput(l.memory.ViewWithout(skip), l.state(), messages)
-       for {
+       view, err := l.layout.Split(l.memory.ViewLines(skip))
+       if err != nil {
+               return err
+       }
+
+       input := turnInput(view, l.state(), messages)
+       for turn := "first"; ; turn = "next" {

                response, err := l.client.Create(ctx, Request{
                        Model:        l.agent.config.Model,
                        Effort:       l.agent.config.Effort,
                        Instructions: instructions,
                        Tools:        agentTools(),
                        Input:        input,
                response, err := l.client.Create(ctx, Request{
                        Model:        l.agent.config.Model,
                        Effort:       l.agent.config.Effort,
                        Instructions: instructions,
                        Tools:        agentTools(),
                        Input:        input,

+                       Turn:         turn,

                })
                if err != nil {
                        return err
                })
                if err != nil {
                        return err


@@ -332,21 +365,65 @@ func (l *loop) handle(ctx context.Context, input []any, response Response) ([]an
        return input, nil
 }
 
        return input, nil
 }
 

-// state returns the per-turn state: the date and time of the turn.
+// warm sends the warm-up request, when the builds are done and no message
+// waits.
+func (l *loop) warm(ctx context.Context) error {
+       err := l.memory.Wait(ctx)
+       if err != nil || l.waiting() {
+               return err
+       }
+
+       instructions, err := l.agent.Prompt()
+       if err != nil {
+               return err
+       }
+
+       view, err := l.layout.Split(l.memory.ViewLines(nil))
+       if err != nil {
+               return err
+       }
+
+       _, err = l.client.Create(ctx, Request{
+               Model:        l.agent.config.Model,
+               Effort:       l.agent.config.Effort,
+               Instructions: instructions,
+               Tools:        agentTools(),
+               Input:        warmUpInput(view),
+               Turn:         "warmup",
+               StopAtOutput: true,
+       })
+
+       return err
+}
+
+// state returns the per-turn state: the end of the view, and the date and time
+// of the turn.

 func (l *loop) state() string {
 func (l *loop) state() string {

-       return "Now: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00")
+       return "</chat>\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00")

 }
 
 }
 

-// turnInput returns the input of the first request of a turn: one user message
-// with the view, the per-turn state, and each new message whole. The view
-// leaves out the leaf lines of the new messages, which come whole after it.
-func turnInput(view, state string, messages []Message) []any {
-       parts := []string{view + "\n", state + "\n"}
+// turnInput returns the input of the first request of a turn, stable parts
+// first: a user message for each view message, one with the per-turn state,
+// and one for each new message, whole. The view leaves out the leaf lines of
+// the new messages.
+func turnInput(view []string, state string, messages []Message) []any {
+       input := warmUpInput(view)
+       input = append(input, userMessage(state))

        for _, message := range messages {
        for _, message := range messages {

-               parts = append(parts, message.Text)
+               input = append(input, userMessage(message.Text))
+       }
+
+       return input
+}
+
+// warmUpInput returns a user message for each view message.
+func warmUpInput(view []string) []any {
+       input := make([]any, len(view))
+       for n, text := range view {
+               input[n] = userMessage(text)

        }
 
        }
 

-       return []any{userMessage(parts...)}
+       return input

 }
 
 // inputItem returns an output item as an input item for the next request. The
 }
 
 // inputItem returns an output item as an input item for the next request. The

diff --git a/turn_test.go b/turn_test.go
index 1b179406b72bb087fcfce2db11f39ccac1970006..b8c92dc260f22e95b04ffc0c82c68781cbb941bc 100644 (file)
--- a/turn_test.go
+++ b/turn_test.go
@@ -2,8 +2,10 @@ package main
 
 import (
        "context"
 
 import (
        "context"

+       "encoding/json"

        "errors"
        "net/http"
        "errors"
        "net/http"

+       "path/filepath"

        "strings"
        "sync"
        "testing"
        "strings"
        "sync"
        "testing"


@@ -70,10 +72,10 @@ func TestFirstRequestHoldsTheViewTheStateAndTheMessage(t *testing.T) {
        test.events.WaitFor(t, "idle", 1)
 
        require.Equal(t, []string{
        test.events.WaitFor(t, "idle", 1)
 
        require.Equal(t, []string{

-               "<chat>\n0+1|user: an earlier message\n</chat>\n",
-               "Now: Thursday 2026-01-01 12:00 +00:00\n",
+               "<chat>\n0+1|user: an earlier message\n",
+               "</chat>\n\nNow: Thursday 2026-01-01 12:00 +00:00",

                "The new message.",
                "The new message.",

-       }, inputParts(t, server.Call(0)))
+       }, inputTexts(t, server.Call(0)))

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


@@ -138,16 +140,16 @@ func TestMessageDuringTheFinalRequestStartsANewTurn(t *testing.T) {
        test.events.WaitFor(t, "idle", 2)
        require.Equal(t, 2, server.Calls())
 
        test.events.WaitFor(t, "idle", 2)
        require.Equal(t, 2, server.Calls())
 

-       parts := inputParts(t, server.Call(1))
-       require.Equal(t, "second", parts[len(parts)-1])
-       require.Contains(t, parts[0], "|reply: answer\n")
+       texts := inputTexts(t, server.Call(1))
+       require.Equal(t, "second", texts[len(texts)-1])
+       require.Contains(t, strings.Join(texts, ""), "|reply: answer\n")

 }
 
 func TestTurnWaitsForTheSummaries(t *testing.T) {
        server := newFakeResponses(t)
        summarizer := newFakeSummarizer()
        summarizer.gate = make(chan struct{})
 }
 
 func TestTurnWaitsForTheSummaries(t *testing.T) {
        server := newFakeResponses(t)
        summarizer := newFakeSummarizer()
        summarizer.gate = make(chan struct{})

-       test := newTestLoopWith(t, server, summarizer)
+       test := newTestLoopWith(t, server, summarizer, false)

 
        _, err := test.memory.Append("echo", strings.Repeat("long output ", 100))
        require.NoError(t, err)
 
        _, err := test.memory.Append("echo", strings.Repeat("long output ", 100))
        require.NoError(t, err)


@@ -161,7 +163,7 @@ func TestTurnWaitsForTheSummaries(t *testing.T) {
        close(summarizer.gate)
        test.events.WaitFor(t, "idle", 1)
        require.Equal(t, 1, server.Calls())
        close(summarizer.gate)
        test.events.WaitFor(t, "idle", 1)
        require.Equal(t, 1, server.Calls())

-       require.NotContains(t, inputParts(t, server.Call(0))[0], pending)
+       require.NotContains(t, inputTexts(t, server.Call(0))[0], pending)

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


@@ -212,12 +214,82 @@ func TestFailedCallEndsTheTurnWithoutARetry(t *testing.T) {
        require.Contains(t, test.events.Last("error"), "HTTP 503")
 }
 
        require.Contains(t, test.events.Last("error"), "HTTP 503")
 }
 

+func TestViewMessagesOnlyGrowBetweenTurns(t *testing.T) {
+       server := newFakeResponses(t)
+       test := newTestLoop(t, server)
+       for n, text := range []string{"one", "two", "three"} {
+               _, err := test.loop.Submit(text)
+               require.NoError(t, err)
+               test.events.WaitFor(t, "idle", n+1)
+       }
+
+       first := server.Call(1).body["input"].([]any)
+       second := server.Call(2).body["input"].([]any)
+       require.Len(t, first, 4)
+       require.Len(t, second, 5)
+       require.Equal(t, first[:2], second[:2])
+       require.Equal(t, []string{
+               "<chat>\n",
+               "0+1|user: one\n1+1|reply: a line\n",
+               "2+1|user: two\n3+1|reply: a line\n",
+               "</chat>\n\nNow: Thursday 2026-01-01 12:00 +00:00",
+               "three",
+       }, inputTexts(t, server.Call(2)))
+}
+
+func TestWarmUpEndsWhereTheNextTurnReadsTheCache(t *testing.T) {
+       server := newFakeResponses(t)
+       server.reply = func(w http.ResponseWriter, call fakeCall) {
+               if call.number == 1 {
+                       writeEvents(w, map[string]any{"type": "response.output_item.added", "item": reasoningItem("rs_1", "")})
+                       return
+               }
+
+               writeEvents(w, completedEvent([]any{messageItem("answer")}, 100, 0, 5))
+       }
+
+       test := newTestLoopWith(t, server, newFakeSummarizer(), true)
+       _, err := test.loop.Submit("one")
+       require.NoError(t, err)
+       require.Eventually(t, func() bool {
+               return server.Calls() == 2
+       }, 10*time.Second, time.Millisecond)
+
+       _, err = test.loop.Submit("two")
+       require.NoError(t, err)
+       test.events.WaitFor(t, "idle", 2)
+       require.Eventually(t, func() bool {
+               return server.Calls() == 4
+       }, 10*time.Second, time.Millisecond)
+
+       warmUp := server.Call(1).body
+       require.Equal(t, server.Call(0).body["instructions"], warmUp["instructions"])
+       require.Equal(t, server.Call(0).body["tools"], warmUp["tools"])
+       require.Equal(t, []string{"<chat>\n", "0+1|user: one\n1+1|reply: answer\n"}, inputTexts(t, server.Call(1)))
+
+       // The next turn begins with the whole warm-up, so its message end after
+       // the view is the end of the warm-up entry.
+       warmInput := warmUp["input"].([]any)
+       turnInput := server.Call(2).body["input"].([]any)
+       require.Equal(t, warmInput, turnInput[:len(warmInput)])
+       require.Equal(t, "</chat>\n\nNow: Thursday 2026-01-01 12:00 +00:00", inputTexts(t, server.Call(2))[len(warmInput)])
+
+       turns := []string{}
+       for _, record := range readUsage(t, test.usagePath) {
+               turns = append(turns, record.Turn)
+       }
+
+       require.Equal(t, []string{"first", "warmup", "first", "warmup"}, turns)
+       require.Equal(t, []string{"user: one", "reply: answer", "user: two", "reply: answer"}, test.Log(t))
+}
+

 // testLoop is a loop that runs in the background of a test, with its parts.
 type testLoop struct {
 // testLoop is a loop that runs in the background of a test, with its parts.
 type testLoop struct {

-       agent  *Agent
-       memory *Memory
-       loop   *loop
-       events *eventRecorder
+       agent     *Agent
+       memory    *Memory
+       loop      *loop
+       events    *eventRecorder
+       usagePath string

 }
 
 // Log returns the logged messages as "kind: text".
 }
 
 // Log returns the logged messages as "kind: text".


@@ -297,12 +369,13 @@ func (r *eventRecorder) WaitFor(t *testing.T, kind string, count int) {
 func newTestLoop(t *testing.T, server *fakeResponses) *testLoop {
        t.Helper()
 
 func newTestLoop(t *testing.T, server *fakeResponses) *testLoop {
        t.Helper()
 

-       return newTestLoopWith(t, server, newFakeSummarizer())
+       return newTestLoopWith(t, server, newFakeSummarizer(), false)

 }
 
 // newTestLoopWith returns a running loop on a new agent whose builds use
 }
 
 // newTestLoopWith returns a running loop on a new agent whose builds use

-// summarizer and whose commands run in a temporary directory.
-func newTestLoopWith(t *testing.T, server *fakeResponses, summarizer Summarizer) *testLoop {
+// summarizer and whose commands run in a temporary directory. warmUp turns the
+// warm-up on.
+func newTestLoopWith(t *testing.T, server *fakeResponses, summarizer Summarizer, warmUp bool) *testLoop {

        t.Helper()
 
        agent, err := openAgent(t.TempDir())
        t.Helper()
 
        agent, err := openAgent(t.TempDir())


@@ -314,12 +387,17 @@ func newTestLoopWith(t *testing.T, server *fakeResponses, summarizer Summarizer)
        require.NoError(t, err)
 
        test := &testLoop{
        require.NoError(t, err)
 
        test := &testLoop{

-               agent:  agent,
-               memory: memory,
-               events: &eventRecorder{},
+               agent:     agent,
+               memory:    memory,
+               events:    &eventRecorder{},
+               usagePath: filepath.Join(t.TempDir(), "usage.jsonl"),

        }
 
        }
 

-       test.loop = newLoop(agent, memory, newTestClient(t, server), clock.Now, test.events.Add)
+       client := newClient(server.URL, apiKey(testKey), agent.CacheKey(), test.usagePath)
+       test.loop, err = newLoop(agent, memory, client, clock.Now, test.events.Add)
+       require.NoError(t, err)
+
+       test.loop.warmUp = warmUp

        ctx, cancel := context.WithCancel(t.Context())
        done := make(chan error)
        go func() {
        ctx, cancel := context.WithCancel(t.Context())
        done := make(chan error)
        go func() {


@@ -335,6 +413,38 @@ func newTestLoopWith(t *testing.T, server *fakeResponses, summarizer Summarizer)
        return test
 }
 
        return test
 }
 

+// inputTexts returns the text of each user message of the input of call.
+func inputTexts(t *testing.T, call fakeCall) []string {
+       t.Helper()
+
+       var body struct {
+               Input []struct {
+                       Role    string `json:"role"`
+                       Content []struct {
+                               Text string `json:"text"`
+                       } `json:"content"`
+               } `json:"input"`
+       }
+
+       require.NoError(t, json.Unmarshal(call.raw, &body))
+
+       var texts []string
+       for _, item := range body.Input {
+               if item.Role != "user" {
+                       continue
+               }
+
+               var text strings.Builder
+               for _, part := range item.Content {
+                       text.WriteString(part.Text)
+               }
+
+               texts = append(texts, text.String())
+       }
+
+       return texts
+}
+

 func reasoningItem(id, encrypted string) map[string]any {
        return map[string]any{
                "type":              "reasoning",
 func reasoningItem(id, encrypted string) map[string]any {
        return map[string]any{
                "type":              "reasoning",

diff --git a/view.go b/view.go
index 8e71eeca205a4f83052c5c91a8fabf7486fc321e..a75b1d8418fc9857d6aafbf38a822c989a817d77 100644 (file)
--- a/view.go
+++ b/view.go
@@ -280,25 +280,25 @@ func (v *View) lineSize(line nodeKey) int {
 // Render returns the view as the model reads it: one line "id+n|text" for each
 // node inside <chat> tags, with the newlines of the text as spaces.
 func (v *View) Render() string {
 // Render returns the view as the model reads it: one line "id+n|text" for each
 // node inside <chat> tags, with the newlines of the text as spaces.
 func (v *View) Render() string {

-       return v.RenderWithout(nil)
+       return "<chat>\n" + strings.Join(v.LinesWithout(nil), "") + "</chat>"

 }
 
 }
 

-// RenderWithout returns the view as Render does, without the leaf lines of the
-// messages in skip. A line that merges such a message with others stays.
-func (v *View) RenderWithout(skip map[int]bool) string {
-       var text strings.Builder
-       text.WriteString("<chat>\n")
+// LinesWithout returns the lines of the view as the model reads them, each with
+// its newline, without the leaf lines of the messages in skip. A line that
+// merges such a message with others stays.
+func (v *View) LinesWithout(skip map[int]bool) []string {
+       lines := make([]string, 0, len(v.lines))

        for _, line := range v.lines {
                if line.level == 0 && skip[line.index] {
                        continue
                }
 
        for _, line := range v.lines {
                if line.level == 0 && skip[line.index] {
                        continue
                }
 

+               var text strings.Builder

                v.renderLine(&text, line)
                v.renderLine(&text, line)

+               lines = append(lines, text.String())

        }
 
        }
 

-       text.WriteString("</chat>")
-
-       return text.String()
+       return lines

 }
 
 // renderLine writes one view line, "id+n|text", and a newline.
 }
 
 // renderLine writes one view line, "id+n|text", and a newline.