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 "<chat>\n". The closing tag goes into the
// state message after the view, so that no view message changes when lines
// 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
Input []any
Node *nodeKey
Turn string
- StopAtOutput bool
}
// Tool is the definition of a function tool.
}
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()
}
// 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)
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
// 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
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
// 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 {
}
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":
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)
// 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
client modelClient
tools *toolbox
layout *viewLayout
- warmUp bool
clock func() time.Time
notify func(Event)
wake chan 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 {
memory: memory,
client: client,
layout: layout,
- warmUp: true,
clock: clock,
notify: notify,
wake: make(chan struct{}, 1),
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 {
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 {
// 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))
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) {
"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) {
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)
}, 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 {
agent *Agent
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())
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() {
projects / onechat / commitdiff