From f59bb67ae1bc2c899ef4f6be9ce2ff6296c7ffcc Mon Sep 17 00:00:00 2001 From: "Anderson A. Giacomolli" Date: Fri, 6 Nov 2026 22:54:20 -0300 Subject: [PATCH] Remove the warm-up request after the turns The first request of a turn reads the view of the turn before from the prompt cache without a warm-up. In two live runs of 6 turns on a view of about 31,000 tokens, turns 2 to 6 read 99.1% to 99.4% of their input from the cache without the warm-up, and 99.3% to 99.6% with it. One call per turn for about 0.1 points is not worth it, so the loop sends no warm-up and the client does not stop a stream early. The view messages that only grow stay. With the whole view in one user message, turn 2 of a third run read nothing from the cache, and turns 3 to 6 read 99.2% to 99.5%. With the grow-only messages, each of the 10 later turns hit. Issue: 5daf23e --- layout.go | 12 ++++---- responses.go | 23 ++++++--------- responses_test.go | 22 -------------- turn.go | 75 +++++++---------------------------------------- turn_test.go | 64 +++++++--------------------------------- 5 files changed, 37 insertions(+), 159 deletions(-) diff --git a/layout.go b/layout.go index a3f9de0..c5e9ac3 100644 --- a/layout.go +++ b/layout.go @@ -14,12 +14,12 @@ import ( 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. +// grow between batches. 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. So the end of the view of each earlier turn stays +// a message end. On the ChatGPT backend, the first request of a turn with this +// layout read the view of the turn before from the prompt cache in each of 10 +// turns, and with the view in one message it missed once in 5. // // The first message starts with "\n". The closing tag goes into the // state message after the view, so that no view message changes when lines diff --git a/responses.go b/responses.go index 19ab6a1..0ba4955 100644 --- a/responses.go +++ b/responses.go @@ -74,9 +74,8 @@ 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 // 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. +// first or next. Node and Turn go only to the usage file, never to the +// endpoint. type Request struct { Model string Effort string @@ -85,7 +84,6 @@ type Request struct { Input []any Node *nodeKey Turn string - StopAtOutput bool } // Tool is the definition of a function tool. @@ -263,7 +261,7 @@ func (c *Client) Create(ctx context.Context, request Request) (Response, error) } record.Turn = request.Turn - response, err := c.send(httpRequest, &record, request.StopAtOutput) + response, err := c.send(httpRequest, &record) if err != nil { record.Error = err.Error() } @@ -279,7 +277,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. -func (c *Client) send(httpRequest *http.Request, record *usageRecord, stopAtOutput bool) (response Response, err error) { +func (c *Client) send(httpRequest *http.Request, record *usageRecord) (response Response, err error) { httpResponse, err := c.http.Do(httpRequest) if err != nil { return Response{}, fmt.Errorf("send model call: %w", err) @@ -298,7 +296,7 @@ func (c *Client) send(httpRequest *http.Request, record *usageRecord, stopAtOutp return Response{}, errors.Join(err, readErr) } - return readStream(httpResponse.Body, stopAtOutput) + return readStream(httpResponse.Body) } // usageLog appends one JSON line for each model call to usage.jsonl. Calls can @@ -406,9 +404,8 @@ 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 -// 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) { +// the completed response when the stream sent none. +func readStream(stream io.Reader) (Response, error) { var response Response reader := bufio.NewReader(stream) var data bytes.Buffer @@ -426,7 +423,7 @@ func readStream(stream io.Reader, stopAtOutput bool) (Response, error) { data.WriteByte('\n') } } else if data.Len() > 0 { - done, eventErr := readEvent(&response, data.Bytes(), stopAtOutput) + done, eventErr := readEvent(&response, data.Bytes()) data.Reset() if eventErr != nil || done { return response, eventErr @@ -441,7 +438,7 @@ func readStream(stream io.Reader, stopAtOutput bool) (Response, error) { // readEvent adds one stream event to response, and tells if it ends the // stream. -func readEvent(response *Response, data []byte, stopAtOutput bool) (bool, error) { +func readEvent(response *Response, data []byte) (bool, error) { var event streamEvent err := json.Unmarshal(data, &event) if err != nil { @@ -449,8 +446,6 @@ func readEvent(response *Response, data []byte, stopAtOutput bool) (bool, error) } 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": diff --git a/responses_test.go b/responses_test.go index 3ff3277..9a1355a 100644 --- a/responses_test.go +++ b/responses_test.go @@ -197,28 +197,6 @@ func TestClientRecordsEachCall(t *testing.T) { 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) diff --git a/turn.go b/turn.go index 91723d2..75cbfb6 100644 --- a/turn.go +++ b/turn.go @@ -31,12 +31,8 @@ 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. // -// 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. +// The view goes out as the messages of a viewLayout, so the first request of +// a turn reads the view of the turn before from the prompt cache. // // 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 @@ -51,7 +47,6 @@ type loop struct { client modelClient tools *toolbox layout *viewLayout - warmUp bool clock func() time.Time notify func(Event) wake chan struct{} @@ -76,9 +71,8 @@ type functionCallOutput struct { Output string `json:"output"` } -// 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. +// 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, error) { layout, err := loadLayout(filepath.Join(agent.dir, layoutFileName)) if err != nil { @@ -90,7 +84,6 @@ func newLoop(agent *Agent, memory *Memory, client modelClient, clock func() time memory: memory, client: client, layout: layout, - warmUp: true, clock: clock, notify: notify, wake: make(chan struct{}, 1), @@ -162,24 +155,15 @@ func (l *loop) Flush(ctx context.Context) error { return l.memory.Flush(ctx) } -// 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. +// 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. func (l *loop) Run(ctx context.Context) error { for { - turned := false for l.waiting() { err := l.turn(ctx) if err != nil { return err } - - turned = true - } - - if turned && l.warmUp { - _ = l.warm(ctx) } select { @@ -365,37 +349,6 @@ func (l *loop) handle(ctx context.Context, input []any, response Response) ([]an return input, nil } -// 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 { @@ -407,7 +360,11 @@ func (l *loop) state() string { // 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 := make([]any, 0, len(view)+1+len(messages)) + for _, text := range view { + input = append(input, userMessage(text)) + } + input = append(input, userMessage(state)) for _, message := range messages { input = append(input, userMessage(message.Text)) @@ -416,16 +373,6 @@ func turnInput(view []string, state string, messages []Message) []any { 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 input -} - // inputItem returns an output item as an input item for the next request. The // calls do not store their items, so the item goes back whole, without its id. func inputItem(raw json.RawMessage) (map[string]any, error) { diff --git a/turn_test.go b/turn_test.go index b8c92dc..f1295e9 100644 --- a/turn_test.go +++ b/turn_test.go @@ -59,6 +59,13 @@ func TestTurnRunsToolsUntilTheFinalAnswer(t *testing.T) { "reply: It said hi.", }, test.Log(t)) require.Equal(t, []string{"user", "busy", "tool", "echo", "reply", "idle"}, test.events.Kinds()) + + var turns []string + for _, record := range readUsage(t, test.usagePath) { + turns = append(turns, record.Turn) + } + + require.Equal(t, []string{"first", "next"}, turns) } func TestFirstRequestHoldsTheViewTheStateAndTheMessage(t *testing.T) { @@ -149,7 +156,7 @@ func TestTurnWaitsForTheSummaries(t *testing.T) { server := newFakeResponses(t) summarizer := newFakeSummarizer() summarizer.gate = make(chan struct{}) - test := newTestLoopWith(t, server, summarizer, false) + test := newTestLoopWith(t, server, summarizer) _, err := test.memory.Append("echo", strings.Repeat("long output ", 100)) require.NoError(t, err) @@ -237,52 +244,6 @@ func TestViewMessagesOnlyGrowBetweenTurns(t *testing.T) { }, 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{"\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, "\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 { agent *Agent @@ -369,13 +330,12 @@ func (r *eventRecorder) WaitFor(t *testing.T, kind string, count int) { func newTestLoop(t *testing.T, server *fakeResponses) *testLoop { t.Helper() - return newTestLoopWith(t, server, newFakeSummarizer(), false) + return newTestLoopWith(t, server, newFakeSummarizer()) } // newTestLoopWith returns a running loop on a new agent whose builds use -// 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 { +// summarizer and whose commands run in a temporary directory. +func newTestLoopWith(t *testing.T, server *fakeResponses, summarizer Summarizer) *testLoop { t.Helper() agent, err := openAgent(t.TempDir()) @@ -396,8 +356,6 @@ func newTestLoopWith(t *testing.T, server *fakeResponses, summarizer Summarizer, 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() { -- 2.53.0