--- /dev/null
+package main
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "strings"
+ "sync"
+ "time"
+)
+
+// modelClient sends model calls.
+type modelClient interface {
+ Create(ctx context.Context, request Request) (Response, error)
+}
+
+// Event tells the clients what the agent does as it happens. Kind is the kind
+// of a logged message (user, reply, tool, echo), or one of busy, idle, and
+// error. ID is the ID of a logged message.
+type Event struct {
+ Kind string `json:"kind"`
+ ID int `json:"id,omitempty"`
+ Text string `json:"text,omitempty"`
+}
+
+// loop runs the turns of one agent. A user message goes to the log when it
+// comes, and it wakes the loop. A turn waits until every message before it is
+// summarized. Then its first request holds the prompt, the tools, and the
+// 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.
+//
+// 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
+// store it. So each request begins with all of the previous one and reads it
+// from the prompt cache. User messages that come during a turn join the next
+// request after the tool results. The replies, the tool calls, and the results
+// go to the log as they happen. The thoughts never do.
+type loop struct {
+ agent *Agent
+ memory *Memory
+ client modelClient
+ tools *toolbox
+ clock func() time.Time
+ notify func(Event)
+ wake chan struct{}
+
+ mutex sync.Mutex
+ pending []Message
+ cancel context.CancelFunc
+}
+
+// functionCall holds the fields of a function call output item.
+type functionCall struct {
+ Type string `json:"type"`
+ Name string `json:"name"`
+ Arguments string `json:"arguments"`
+ CallID string `json:"call_id"`
+}
+
+// functionCallOutput is the input item with the result of a function call.
+type functionCallOutput struct {
+ Type string `json:"type"`
+ CallID string `json:"call_id"`
+ 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 {
+ l := &loop{
+ agent: agent,
+ memory: memory,
+ client: client,
+ clock: clock,
+ notify: notify,
+ wake: make(chan struct{}, 1),
+ }
+
+ l.tools = &toolbox{
+ log: l,
+ chat: memory,
+ workdir: agent.config.Workdir,
+ }
+
+ return l
+}
+
+// Submit logs a user message and wakes the loop. The message joins the next
+// request: the first one of a new turn, or the next one of a running turn.
+func (l *loop) Submit(text string) (Message, error) {
+ message, err := l.Append("user", text)
+ if err != nil {
+ return Message{}, err
+ }
+
+ l.mutex.Lock()
+ l.pending = append(l.pending, message)
+ l.mutex.Unlock()
+
+ select {
+ case l.wake <- struct{}{}:
+ default:
+ }
+
+ return message, nil
+}
+
+// Interrupt ends the running turn. A running tool is killed, and the turn sends
+// no further request.
+func (l *loop) Interrupt() bool {
+ l.mutex.Lock()
+ defer l.mutex.Unlock()
+
+ if l.cancel == nil {
+ return false
+ }
+
+ l.cancel()
+
+ return true
+}
+
+// Append logs a message and tells the clients. The toolbox logs its calls and
+// results through it.
+func (l *loop) Append(kind, text string) (Message, error) {
+ message, err := l.memory.Append(kind, text)
+ if err != nil {
+ return Message{}, err
+ }
+
+ l.notify(Event{
+ Kind: kind,
+ ID: message.ID,
+ Text: text,
+ })
+
+ return message, nil
+}
+
+// Flush makes the log durable.
+func (l *loop) Flush(ctx context.Context) error {
+ 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.
+func (l *loop) Run(ctx context.Context) error {
+ for {
+ for l.waiting() {
+ err := l.turn(ctx)
+ if err != nil {
+ return err
+ }
+ }
+
+ select {
+ case <-l.wake:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+ }
+}
+
+// waiting tells if user messages wait for a request.
+func (l *loop) waiting() bool {
+ l.mutex.Lock()
+ defer l.mutex.Unlock()
+
+ return len(l.pending) > 0
+}
+
+// take returns the user messages that wait for a request, and removes them.
+func (l *loop) take() []Message {
+ l.mutex.Lock()
+ defer l.mutex.Unlock()
+
+ messages := l.pending
+ l.pending = nil
+
+ return messages
+}
+
+// turn runs one turn. A failed or interrupted model call ends the turn and
+// tells the clients. The messages of the turn stay in the log and the view, and
+// the next turn starts only with a new message, so a lasting failure such as a
+// needed login does not repeat. Only a failed log write is an error of turn.
+func (l *loop) turn(ctx context.Context) error {
+ ctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ l.mutex.Lock()
+ l.cancel = cancel
+ l.mutex.Unlock()
+
+ defer func() {
+ l.mutex.Lock()
+ l.cancel = nil
+ l.mutex.Unlock()
+ }()
+
+ l.notify(Event{
+ Kind: "busy",
+ })
+
+ err := l.chain(ctx)
+ if err != nil && !errors.Is(err, errLogWrite) {
+ l.notify(Event{
+ Kind: "error",
+ Text: turnError(ctx, err),
+ })
+
+ err = nil
+ }
+
+ l.notify(Event{
+ Kind: "idle",
+ })
+
+ return err
+}
+
+// errLogWrite marks an error of a log write, which ends the loop.
+var errLogWrite = errors.New("log write failed")
+
+// chain runs the request chain of a turn until the final answer.
+func (l *loop) chain(ctx context.Context) error {
+ messages := l.take()
+ if len(messages) == 0 {
+ return nil
+ }
+
+ err := l.memory.Wait(ctx)
+ if err != nil {
+ return err
+ }
+
+ instructions, err := l.agent.Prompt()
+ if err != nil {
+ return err
+ }
+
+ skip := make(map[int]bool, len(messages))
+ for _, message := range messages {
+ skip[message.ID] = true
+ }
+
+ input := turnInput(l.memory.ViewWithout(skip), l.state(), messages)
+ for {
+ response, err := l.client.Create(ctx, Request{
+ Model: l.agent.config.Model,
+ Effort: l.agent.config.Effort,
+ Instructions: instructions,
+ Tools: agentTools(),
+ Input: input,
+ })
+ if err != nil {
+ return err
+ }
+
+ input, err = l.handle(ctx, input, response)
+ if err != nil {
+ return err
+ }
+
+ if len(input) == 0 {
+ return nil
+ }
+ }
+}
+
+// handle logs the replies of response and runs its tool calls. It returns the
+// input of the next request: input, the output items, the tool results, and
+// the user messages that came in the meantime. After the final answer it
+// returns no input.
+func (l *loop) handle(ctx context.Context, input []any, response Response) ([]any, error) {
+ var outputs []any
+ for _, raw := range response.Output {
+ item, err := inputItem(raw)
+ if err != nil {
+ return nil, err
+ }
+
+ input = append(input, item)
+
+ var call functionCall
+ err = json.Unmarshal(raw, &call)
+ if err != nil {
+ return nil, fmt.Errorf("decode output item: %w", err)
+ }
+
+ switch call.Type {
+ case "message":
+ text, err := outputText([]json.RawMessage{raw})
+ if err != nil {
+ return nil, err
+ }
+
+ text = strings.TrimSpace(text)
+ if text == "" {
+ continue
+ }
+
+ _, err = l.Append("reply", text)
+ if err != nil {
+ return nil, fmt.Errorf("%w: %w", errLogWrite, err)
+ }
+ case "function_call":
+ output, err := l.tools.Call(ctx, call.Name, call.Arguments)
+ if err != nil {
+ return nil, fmt.Errorf("%w: %w", errLogWrite, err)
+ }
+
+ outputs = append(outputs, functionCallOutput{
+ Type: "function_call_output",
+ CallID: call.CallID,
+ Output: output,
+ })
+ }
+ }
+
+ if len(outputs) == 0 {
+ return nil, nil
+ }
+
+ input = append(input, outputs...)
+ for _, message := range l.take() {
+ input = append(input, userMessage(message.Text))
+ }
+
+ return input, nil
+}
+
+// state returns the per-turn state: the date and time of the turn.
+func (l *loop) state() string {
+ return "Now: " + 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"}
+ for _, message := range messages {
+ parts = append(parts, message.Text)
+ }
+
+ return []any{userMessage(parts...)}
+}
+
+// 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) {
+ var item map[string]any
+ err := json.Unmarshal(raw, &item)
+ if err != nil {
+ return nil, fmt.Errorf("decode output item: %w", err)
+ }
+
+ delete(item, "id")
+
+ return item, nil
+}
+
+// turnError returns the text of the error that ended a turn.
+func turnError(ctx context.Context, err error) string {
+ if ctx.Err() != nil {
+ return "interrupted"
+ }
+
+ return err.Error()
+}
--- /dev/null
+package main
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestTurnRunsToolsUntilTheFinalAnswer(t *testing.T) {
+ server := newFakeResponses(t)
+ server.reply = func(w http.ResponseWriter, call fakeCall) {
+ if call.number == 0 {
+ writeEvents(w, completedEvent([]any{
+ reasoningItem("rs_1", "opaque thought"),
+ functionCallItem("fc_1", "call_1", "shell", `{"command":"echo hi"}`),
+ }, 100, 0, 5))
+
+ return
+ }
+
+ writeEvents(w, completedEvent([]any{messageItem("It said hi.")}, 120, 100, 5))
+ }
+
+ test := newTestLoop(t, server)
+ _, err := test.loop.Submit("Run echo hi.")
+ require.NoError(t, err)
+ test.events.WaitFor(t, "idle", 1)
+ require.Equal(t, 2, server.Calls())
+
+ first := server.Call(0).body
+ second := server.Call(1).body
+ instructions, err := test.agent.Prompt()
+ require.NoError(t, err)
+ require.Equal(t, instructions, first["instructions"])
+ require.Equal(t, test.agent.config.Model, first["model"])
+
+ // The second request begins with the whole first one.
+ firstInput := first["input"].([]any)
+ secondInput := second["input"].([]any)
+ require.Equal(t, firstInput, secondInput[:len(firstInput)])
+ require.Equal(t, []any{
+ map[string]any{"type": "reasoning", "encrypted_content": "opaque thought", "summary": []any{}},
+ map[string]any{"type": "function_call", "call_id": "call_1", "name": "shell", "arguments": `{"command":"echo hi"}`},
+ map[string]any{"type": "function_call_output", "call_id": "call_1", "output": "hi\n"},
+ }, secondInput[len(firstInput):])
+
+ require.Equal(t, []string{
+ "user: Run echo hi.",
+ `tool: shell {"command":"echo hi"}`,
+ "echo: hi\n",
+ "reply: It said hi.",
+ }, test.Log(t))
+ require.Equal(t, []string{"user", "busy", "tool", "echo", "reply", "idle"}, test.events.Kinds())
+}
+
+func TestFirstRequestHoldsTheViewTheStateAndTheMessage(t *testing.T) {
+ server := newFakeResponses(t)
+ test := newTestLoop(t, server)
+ _, err := test.memory.Append("user", "an earlier message")
+ require.NoError(t, err)
+
+ _, err = test.loop.Submit("The new message.")
+ require.NoError(t, err)
+ 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",
+ "The new message.",
+ }, inputParts(t, server.Call(0)))
+}
+
+func TestMessageDuringATurnJoinsTheNextRequest(t *testing.T) {
+ server := newFakeResponses(t)
+ arrived := make(chan struct{})
+ release := make(chan struct{})
+ server.reply = func(w http.ResponseWriter, call fakeCall) {
+ if call.number == 0 {
+ close(arrived)
+ <-release
+ writeEvents(w, completedEvent([]any{functionCallItem("fc_1", "call_1", "date", `{"id":0}`)}, 100, 0, 5))
+
+ return
+ }
+
+ writeEvents(w, completedEvent([]any{messageItem("done")}, 100, 0, 5))
+ }
+
+ test := newTestLoop(t, server)
+ _, err := test.loop.Submit("first")
+ require.NoError(t, err)
+
+ <-arrived
+ _, err = test.loop.Submit("second, during the turn")
+ require.NoError(t, err)
+ close(release)
+
+ test.events.WaitFor(t, "idle", 1)
+ input := server.Call(1).body["input"].([]any)
+ require.Equal(t, map[string]any{
+ "role": "user",
+ "content": []any{map[string]any{"type": "input_text", "text": "second, during the turn"}},
+ }, input[len(input)-1])
+
+ time.Sleep(50 * time.Millisecond)
+ require.Equal(t, 2, server.Calls())
+ require.Equal(t, 1, test.events.Count("busy"))
+}
+
+func TestMessageDuringTheFinalRequestStartsANewTurn(t *testing.T) {
+ server := newFakeResponses(t)
+ arrived := make(chan struct{})
+ release := make(chan struct{})
+ server.reply = func(w http.ResponseWriter, call fakeCall) {
+ if call.number == 0 {
+ close(arrived)
+ <-release
+ }
+
+ writeEvents(w, completedEvent([]any{messageItem("answer")}, 100, 0, 5))
+ }
+
+ test := newTestLoop(t, server)
+ _, err := test.loop.Submit("first")
+ require.NoError(t, err)
+
+ <-arrived
+ _, err = test.loop.Submit("second")
+ require.NoError(t, err)
+ close(release)
+
+ 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")
+}
+
+func TestTurnWaitsForTheSummaries(t *testing.T) {
+ server := newFakeResponses(t)
+ summarizer := newFakeSummarizer()
+ summarizer.gate = make(chan struct{})
+ test := newTestLoopWith(t, server, summarizer)
+
+ _, err := test.memory.Append("echo", strings.Repeat("long output ", 100))
+ require.NoError(t, err)
+
+ _, err = test.loop.Submit("go on")
+ require.NoError(t, err)
+
+ time.Sleep(50 * time.Millisecond)
+ require.Equal(t, 0, 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)
+}
+
+func TestInterruptEndsTheTurn(t *testing.T) {
+ server := newFakeResponses(t)
+ arrived := make(chan struct{})
+ release := make(chan struct{})
+ server.reply = func(w http.ResponseWriter, call fakeCall) {
+ if call.number == 0 {
+ close(arrived)
+ <-release
+ }
+
+ writeEvents(w, completedEvent([]any{messageItem("answer")}, 100, 0, 5))
+ }
+
+ test := newTestLoop(t, server)
+ _, err := test.loop.Submit("a long task")
+ require.NoError(t, err)
+
+ <-arrived
+ require.True(t, test.loop.Interrupt())
+ test.events.WaitFor(t, "idle", 1)
+ close(release)
+
+ require.Equal(t, "interrupted", test.events.Last("error"))
+ require.Equal(t, []string{"user: a long task"}, test.Log(t))
+ require.False(t, test.loop.Interrupt())
+
+ _, err = test.loop.Submit("again")
+ require.NoError(t, err)
+ test.events.WaitFor(t, "idle", 2)
+ require.Equal(t, "reply: answer", test.Log(t)[2])
+}
+
+func TestFailedCallEndsTheTurnWithoutARetry(t *testing.T) {
+ server := newFakeResponses(t)
+ server.reply = func(w http.ResponseWriter, call fakeCall) {
+ http.Error(w, "busy", http.StatusServiceUnavailable)
+ }
+
+ test := newTestLoop(t, server)
+ _, err := test.loop.Submit("hello")
+ require.NoError(t, err)
+ test.events.WaitFor(t, "idle", 1)
+
+ time.Sleep(50 * time.Millisecond)
+ require.Equal(t, 1, server.Calls())
+ require.Contains(t, test.events.Last("error"), "HTTP 503")
+}
+
+// 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
+}
+
+// Log returns the logged messages as "kind: text".
+func (l *testLoop) Log(t *testing.T) []string {
+ t.Helper()
+
+ var lines []string
+ for id := range l.memory.Len() {
+ message, err := l.memory.Message(id)
+ require.NoError(t, err)
+
+ lines = append(lines, message.Kind+": "+message.Text)
+ }
+
+ return lines
+}
+
+// eventRecorder keeps the events of a loop.
+type eventRecorder struct {
+ mutex sync.Mutex
+ events []Event
+}
+
+func (r *eventRecorder) Add(event Event) {
+ r.mutex.Lock()
+ defer r.mutex.Unlock()
+
+ r.events = append(r.events, event)
+}
+
+func (r *eventRecorder) Kinds() []string {
+ r.mutex.Lock()
+ defer r.mutex.Unlock()
+
+ kinds := make([]string, len(r.events))
+ for n, event := range r.events {
+ kinds[n] = event.Kind
+ }
+
+ return kinds
+}
+
+func (r *eventRecorder) Count(kind string) int {
+ count := 0
+ for _, each := range r.Kinds() {
+ if each == kind {
+ count++
+ }
+ }
+
+ return count
+}
+
+// Last returns the text of the last event of kind.
+func (r *eventRecorder) Last(kind string) string {
+ r.mutex.Lock()
+ defer r.mutex.Unlock()
+
+ for n := len(r.events) - 1; n >= 0; n-- {
+ if r.events[n].Kind == kind {
+ return r.events[n].Text
+ }
+ }
+
+ return ""
+}
+
+// WaitFor waits until count events of kind came.
+func (r *eventRecorder) WaitFor(t *testing.T, kind string, count int) {
+ t.Helper()
+
+ require.Eventually(t, func() bool {
+ return r.Count(kind) >= count
+ }, 10*time.Second, time.Millisecond)
+}
+
+func newTestLoop(t *testing.T, server *fakeResponses) *testLoop {
+ t.Helper()
+
+ 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.
+func newTestLoopWith(t *testing.T, server *fakeResponses, summarizer Summarizer) *testLoop {
+ t.Helper()
+
+ agent, err := openAgent(t.TempDir())
+ require.NoError(t, err)
+
+ agent.config.Workdir = t.TempDir()
+ clock := newTestClock()
+ memory, err := openMemory(t.Context(), t.TempDir(), clock.Now, summarizer)
+ require.NoError(t, err)
+
+ test := &testLoop{
+ agent: agent,
+ memory: memory,
+ events: &eventRecorder{},
+ }
+
+ test.loop = newLoop(agent, memory, newTestClient(t, server), clock.Now, test.events.Add)
+ ctx, cancel := context.WithCancel(t.Context())
+ done := make(chan error)
+ go func() {
+ done <- test.loop.Run(ctx)
+ }()
+
+ t.Cleanup(func() {
+ cancel()
+ require.True(t, errors.Is(<-done, context.Canceled))
+ require.NoError(t, memory.Close())
+ })
+
+ return test
+}
+
+func reasoningItem(id, encrypted string) map[string]any {
+ return map[string]any{
+ "type": "reasoning",
+ "id": id,
+ "encrypted_content": encrypted,
+ "summary": []any{},
+ }
+}
+
+func functionCallItem(id, callID, name, arguments string) map[string]any {
+ return map[string]any{
+ "type": "function_call",
+ "id": id,
+ "call_id": callID,
+ "name": name,
+ "arguments": arguments,
+ }
+}
projects / onechat / commitdiff