}
}
- return daemonServe(ctx, listener, telegram, func(notify func(Event)) (*loop, error) {
- return newLoop(agent, memory, client, time.Now, notify)
+ return daemonServe(ctx, listener, telegram, func(notify func(Event)) (*TurnLoop, error) {
+ return newTurnLoop(agent, memory, client, time.Now, notify)
})
}
// Accept serves each client of listener until the listener closes. A new
// client gets a ready event with the name of the agent, and a busy event when a
// turn runs.
-func (h *DaemonHub) Accept(ctx context.Context, listener net.Listener, l *loop) {
+func (h *DaemonHub) Accept(ctx context.Context, listener net.Listener, l *TurnLoop) {
for {
conn, err := listener.Accept()
if err != nil {
}
// read runs the requests of client until it closes its connection.
-func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *loop) {
+func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *TurnLoop) {
defer h.workers.Done()
defer h.drop(client)
// daemonServe runs the loop that start makes and serves the clients of listener
// until ctx ends or the loop fails. It closes listener. When bot is not nil, it
// runs the bot too, and a failure of the bot stops the agent.
-func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, start func(notify func(Event)) (*loop, error)) error {
+func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, start func(notify func(Event)) (*TurnLoop, error)) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
}
// daemonRunRequest runs one request of a client.
-func daemonRunRequest(l *loop, request SocketRequest) error {
+func daemonRunRequest(l *TurnLoop, request SocketRequest) error {
switch request.Type {
case "send":
if request.Text == "" {
ctx, cancel := context.WithCancel(t.Context())
done := make(chan error)
go func() {
- done <- daemonServe(ctx, listener, bot, func(notify func(Event)) (*loop, error) {
- return newLoop(agent, memory, client, clock.Now, notify)
+ done <- daemonServe(ctx, listener, bot, func(notify func(Event)) (*TurnLoop, error) {
+ return newTurnLoop(agent, memory, client, clock.Now, notify)
})
}()
// Run polls for the messages of the owner and sends the queued events to the
// chat until ctx ends. It returns an error only when a message or the offset
// cannot be written, because then the bot would lose or repeat messages.
-func (b *TelegramBot) Run(ctx context.Context, l *loop) error {
+func (b *TelegramBot) Run(ctx context.Context, l *TurnLoop) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
// poll gets the updates from the stored offset on and handles each one, until
// ctx ends. Each update is handled to its end also when ctx ends during it, so
// a stop does not repeat its message.
-func (b *TelegramBot) poll(ctx context.Context, l *loop) error {
+func (b *TelegramBot) poll(ctx context.Context, l *TurnLoop) error {
offset, err := telegramLoadOffset(b.offsetPath)
if err != nil {
return err
// /stop interrupts the turn. Any other message of the owner gets the answer
// that only text is supported. Every other update is ignored, and its content
// is not logged.
-func (b *TelegramBot) handle(ctx context.Context, l *loop, update TelegramUpdate) error {
+func (b *TelegramBot) handle(ctx context.Context, l *TurnLoop, update TelegramUpdate) error {
message := update.Message
if message == nil || message.Chat.Type != "private" || message.Chat.ID != b.owner || message.From == nil || message.From.ID != b.owner {
b.logger.Print("update ignored: not a message of the owner in the private chat")
"time"
)
+// errTurnLogWrite marks an error of a log write, which ends the loop.
+var errTurnLogWrite = errors.New("log write failed")
+
+// newTurnLoop 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 newTurnLoop(agent *Agent, memory *Memory, client TurnClient, clock func() time.Time, notify func(Event)) (*TurnLoop, error) {
+ layout, err := loadLayout(filepath.Join(agent.dir, layoutFileName))
+ if err != nil {
+ return nil, err
+ }
+
+ l := &TurnLoop{
+ agent: agent,
+ memory: memory,
+ client: client,
+ layout: layout,
+ clock: clock,
+ notify: notify,
+ wake: make(chan struct{}, 1),
+ }
+
+ l.tools = &toolbox{
+ log: l,
+ chat: memory,
+ workdir: agent.config.Workdir,
+ }
+
+ return l, nil
+}
+
// TurnClient sends model calls.
type TurnClient interface {
Create(ctx context.Context, request ModelRequest) (ModelResponse, error)
Source string `json:"source,omitempty"`
}
-// loop runs the turns of one agent. A user message goes to the log when it
+// TurnLoop 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.
+// 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.
//
// 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.
// 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 {
+type TurnLoop struct {
agent *Agent
memory *Memory
client TurnClient
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 TurnClient, 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,
- layout: layout,
- clock: clock,
- notify: notify,
- wake: make(chan struct{}, 1),
- }
-
- l.tools = &toolbox{
- log: l,
- chat: memory,
- workdir: agent.config.Workdir,
- }
-
- return l, nil
-}
-
// Submit logs a user message from the client source 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(source, text string) (Message, error) {
+func (l *TurnLoop) Submit(source, text string) (Message, error) {
message, err := l.record("user", source, text)
if err != nil {
return Message{}, err
// Interrupt ends the running turn. A running tool is killed, and the turn sends
// no further request.
-func (l *loop) Interrupt() bool {
+func (l *TurnLoop) Interrupt() bool {
l.mutex.Lock()
defer l.mutex.Unlock()
}
// Busy tells if a turn runs.
-func (l *loop) Busy() bool {
+func (l *TurnLoop) Busy() bool {
l.mutex.Lock()
defer l.mutex.Unlock()
// 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) {
+func (l *TurnLoop) Append(kind, text string) (Message, error) {
return l.record(kind, "", text)
}
-// record logs a message and tells the clients, with the client source of a user
-// message.
-func (l *loop) record(kind, source, 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,
- Source: source,
- })
-
- return message, nil
-}
-
// Flush makes the log durable.
-func (l *loop) Flush(ctx context.Context) error {
+func (l *TurnLoop) 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 {
+func (l *TurnLoop) Run(ctx context.Context) error {
for {
for l.waiting() {
err := l.turn(ctx)
}
}
-// waiting tells if user messages wait for a request.
-func (l *loop) waiting() bool {
- l.mutex.Lock()
- defer l.mutex.Unlock()
+// record logs a message and tells the clients, with the client source of a user
+// message.
+func (l *TurnLoop) record(kind, source, text string) (Message, error) {
+ message, err := l.memory.Append(kind, text)
+ if err != nil {
+ return Message{}, err
+ }
- return len(l.pending) > 0
+ l.notify(Event{
+ Kind: kind,
+ ID: message.ID,
+ Text: text,
+ Source: source,
+ })
+
+ return message, nil
}
-// take returns the user messages that wait for a request, and removes them.
-func (l *loop) take() []Message {
+// waiting tells if user messages wait for a request.
+func (l *TurnLoop) waiting() bool {
l.mutex.Lock()
defer l.mutex.Unlock()
- messages := l.pending
- l.pending = nil
-
- return messages
+ return len(l.pending) > 0
}
// 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 {
+func (l *TurnLoop) turn(ctx context.Context) error {
ctx, cancel := context.WithCancel(ctx)
defer cancel()
})
err := l.chain(ctx)
- if err != nil && !errors.Is(err, errLogWrite) {
+ if err != nil && !errors.Is(err, errTurnLogWrite) {
l.notify(Event{
Kind: "error",
Text: turnError(ctx, err),
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 {
+func (l *TurnLoop) chain(ctx context.Context) error {
messages := l.take()
if len(messages) == 0 {
return nil
}
}
+// take returns the user messages that wait for a request, and removes them.
+func (l *TurnLoop) take() []Message {
+ l.mutex.Lock()
+ defer l.mutex.Unlock()
+
+ messages := l.pending
+ l.pending = nil
+
+ return messages
+}
+
+// state returns the per-turn state: the end of the view, and the date and time
+// of the turn.
+func (l *TurnLoop) state() string {
+ return "</chat>\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00")
+}
+
// 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 ModelResponse) ([]any, error) {
+func (l *TurnLoop) handle(ctx context.Context, input []any, response ModelResponse) ([]any, error) {
var outputs []any
for _, raw := range response.Output {
- item, err := inputItem(raw)
+ item, err := turnInputItem(raw)
if err != nil {
return nil, err
}
input = append(input, item)
- var call functionCall
+ var call TurnFunctionCall
err = json.Unmarshal(raw, &call)
if err != nil {
return nil, fmt.Errorf("decode output item: %w", err)
_, err = l.Append("reply", text)
if err != nil {
- return nil, fmt.Errorf("%w: %w", errLogWrite, err)
+ return nil, fmt.Errorf("%w: %w", errTurnLogWrite, 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)
+ return nil, fmt.Errorf("%w: %w", errTurnLogWrite, err)
}
- outputs = append(outputs, functionCallOutput{
+ outputs = append(outputs, TurnFunctionCallOutput{
Type: "function_call_output",
CallID: call.CallID,
Output: output,
return input, nil
}
-// state returns the per-turn state: the end of the view, and the date and time
-// of the turn.
-func (l *loop) state() string {
- return "</chat>\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00")
+// TurnFunctionCall holds the fields of a function call output item.
+type TurnFunctionCall struct {
+ Type string `json:"type"`
+ Name string `json:"name"`
+ Arguments string `json:"arguments"`
+ CallID string `json:"call_id"`
+}
+
+// TurnFunctionCallOutput is the input item with the result of a function call.
+type TurnFunctionCallOutput struct {
+ Type string `json:"type"`
+ CallID string `json:"call_id"`
+ Output string `json:"output"`
+}
+
+// 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()
}
// turnInput returns the input of the first request of a turn, stable parts
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) {
+// turnInputItem 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 turnInputItem(raw json.RawMessage) (map[string]any, error) {
var item map[string]any
err := json.Unmarshal(raw, &item)
if err != nil {
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()
-}
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("It said hi.")}, 120, 100, 5))
}
- test := newTestLoop(t, server)
+ test := newTestTurnLoop(t, server)
_, err := test.loop.Submit(terminalSource, "Run echo hi.")
require.NoError(t, err)
test.events.WaitFor(t, "idle", 1)
func TestFirstRequestHoldsTheViewTheStateAndTheMessage(t *testing.T) {
server := newModelFake(t)
- test := newTestLoop(t, server)
+ test := newTestTurnLoop(t, server)
_, err := test.memory.Append("user", "an earlier message")
require.NoError(t, err)
"<chat>\n0+1|user: an earlier message\n",
"</chat>\n\nNow: Thursday 2026-01-01 12:00 +00:00",
"The new message.",
- }, inputTexts(t, server.Call(0)))
+ }, turnInputTexts(t, server.Call(0)))
}
func TestMessageDuringATurnJoinsTheNextRequest(t *testing.T) {
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("done")}, 100, 0, 5))
}
- test := newTestLoop(t, server)
+ test := newTestTurnLoop(t, server)
_, err := test.loop.Submit(terminalSource, "first")
require.NoError(t, err)
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("answer")}, 100, 0, 5))
}
- test := newTestLoop(t, server)
+ test := newTestTurnLoop(t, server)
_, err := test.loop.Submit(terminalSource, "first")
require.NoError(t, err)
test.events.WaitFor(t, "idle", 2)
require.Equal(t, 2, server.Calls())
- texts := inputTexts(t, server.Call(1))
+ texts := turnInputTexts(t, server.Call(1))
require.Equal(t, "second", texts[len(texts)-1])
require.Contains(t, strings.Join(texts, ""), "|reply: answer\n")
}
server := newModelFake(t)
summarizer := newFakeSummarizer()
summarizer.gate = make(chan struct{})
- test := newTestLoopWith(t, server, summarizer)
+ test := newTestTurnLoopWith(t, server, summarizer)
_, err := test.memory.Append("echo", strings.Repeat("long output ", 100))
require.NoError(t, err)
close(summarizer.gate)
test.events.WaitFor(t, "idle", 1)
require.Equal(t, 1, server.Calls())
- require.NotContains(t, inputTexts(t, server.Call(0))[0], pending)
+ require.NotContains(t, turnInputTexts(t, server.Call(0))[0], pending)
}
func TestInterruptEndsTheTurn(t *testing.T) {
modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("answer")}, 100, 0, 5))
}
- test := newTestLoop(t, server)
+ test := newTestTurnLoop(t, server)
_, err := test.loop.Submit(terminalSource, "a long task")
require.NoError(t, err)
http.Error(w, "busy", http.StatusServiceUnavailable)
}
- test := newTestLoop(t, server)
+ test := newTestTurnLoop(t, server)
_, err := test.loop.Submit(terminalSource, "hello")
require.NoError(t, err)
test.events.WaitFor(t, "idle", 1)
func TestViewMessagesOnlyGrowBetweenTurns(t *testing.T) {
server := newModelFake(t)
- test := newTestLoop(t, server)
+ test := newTestTurnLoop(t, server)
for n, text := range []string{"one", "two", "three"} {
_, err := test.loop.Submit(terminalSource, text)
require.NoError(t, err)
"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)))
+ }, turnInputTexts(t, server.Call(2)))
}
-// testLoop is a loop that runs in the background of a test, with its parts.
-type testLoop struct {
+// TurnTestLoop is a loop that runs in the background of a test, with its parts.
+type TurnTestLoop struct {
agent *Agent
memory *Memory
- loop *loop
- events *eventRecorder
+ loop *TurnLoop
+ events *TurnEventRecorder
usagePath string
}
// Log returns the logged messages as "kind: text".
-func (l *testLoop) Log(t *testing.T) []string {
+func (l *TurnTestLoop) Log(t *testing.T) []string {
t.Helper()
var lines []string
return lines
}
-// eventRecorder keeps the events of a loop.
-type eventRecorder struct {
+// TurnEventRecorder keeps the events of a loop.
+type TurnEventRecorder struct {
mutex sync.Mutex
events []Event
}
-func (r *eventRecorder) Add(event Event) {
- r.mutex.Lock()
- defer r.mutex.Unlock()
+// WaitFor waits until count events of kind came.
+func (r *TurnEventRecorder) WaitFor(t *testing.T, kind string, count int) {
+ t.Helper()
- r.events = append(r.events, event)
+ require.Eventually(t, func() bool {
+ return r.Count(kind) >= count
+ }, 10*time.Second, time.Millisecond)
}
-func (r *eventRecorder) Kinds() []string {
+func (r *TurnEventRecorder) Kinds() []string {
r.mutex.Lock()
defer r.mutex.Unlock()
return kinds
}
-func (r *eventRecorder) Count(kind string) int {
+func (r *TurnEventRecorder) Count(kind string) int {
count := 0
for _, each := range r.Kinds() {
if each == kind {
}
// Last returns the text of the last event of kind.
-func (r *eventRecorder) Last(kind string) string {
+func (r *TurnEventRecorder) Last(kind string) string {
r.mutex.Lock()
defer r.mutex.Unlock()
return ""
}
-// WaitFor waits until count events of kind came.
-func (r *eventRecorder) WaitFor(t *testing.T, kind string, count int) {
+func (r *TurnEventRecorder) Add(event Event) {
+ r.mutex.Lock()
+ defer r.mutex.Unlock()
+
+ r.events = append(r.events, event)
+}
+
+func newTestTurnLoop(t *testing.T, server *ModelFake) *TurnTestLoop {
t.Helper()
- require.Eventually(t, func() bool {
- return r.Count(kind) >= count
- }, 10*time.Second, time.Millisecond)
+ return newTestTurnLoopWith(t, server, newFakeSummarizer())
}
-func newTestLoop(t *testing.T, server *ModelFake) *testLoop {
+// turnInputTexts returns the text of each user message of the input of call.
+func turnInputTexts(t *testing.T, call ModelFakeCall) []string {
t.Helper()
- return newTestLoopWith(t, server, newFakeSummarizer())
+ 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
}
-// newTestLoopWith returns a running loop on a new agent whose builds use
+// newTestTurnLoopWith 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 *ModelFake, summarizer Summarizer) *testLoop {
+func newTestTurnLoopWith(t *testing.T, server *ModelFake, summarizer Summarizer) *TurnTestLoop {
t.Helper()
agent, err := openAgent(t.TempDir())
memory, err := openMemory(t.Context(), t.TempDir(), clock.Now, summarizer)
require.NoError(t, err)
- test := &testLoop{
+ test := &TurnTestLoop{
agent: agent,
memory: memory,
- events: &eventRecorder{},
+ events: &TurnEventRecorder{},
usagePath: filepath.Join(t.TempDir(), "usage.jsonl"),
}
client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), test.usagePath)
- test.loop, err = newLoop(agent, memory, client, clock.Now, test.events.Add)
+ test.loop, err = newTurnLoop(agent, memory, client, clock.Now, test.events.Add)
require.NoError(t, err)
ctx, cancel := context.WithCancel(t.Context())
done := make(chan error)
return test
}
-
-// inputTexts returns the text of each user message of the input of call.
-func inputTexts(t *testing.T, call ModelFakeCall) []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
-}
projects / onechat / commitdiff