git.giacomolli.uk - onechat/blob - turn.go

   1 package main
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "path/filepath"
   8         "sync"
   9         "time"
  10 )
  11
  12 // newTurnLoop returns the loop of agent with its memory and its subagents. Its
  13 // calls go through client, and notify gets each event. The clock gives the date
  14 // of a turn.
  15 func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, client ChainClient, clock func() time.Time, notify func(Event)) (*TurnLoop, error) {
  16         layout, err := loadViewLayout(filepath.Join(agent.dir, viewLayoutFileName))
  17         if err != nil {
  18                 return nil, err
  19         }
  20
  21         l := &TurnLoop{
  22                 agent:  agent,
  23                 memory: memory,
  24                 client: client,
  25                 layout: layout,
  26                 clock:  clock,
  27                 notify: notify,
  28                 wake:   make(chan struct{}, 1),
  29         }
  30
  31         l.runs = &SubagentRuns{
  32                 agent:     agent,
  33                 memory:    memory,
  34                 subagents: subagents,
  35                 layout:    layout,
  36                 client:    client,
  37                 state:     l.state,
  38                 report:    l.Report,
  39                 notify:    notify,
  40                 running:   make(map[string]*SubagentRun),
  41         }
  42
  43         l.tools = &Toolbox{
  44                 log:       l,
  45                 chat:      memory,
  46                 subagents: subagents,
  47                 runs:      l.runs,
  48                 workdir:   agent.config.Workdir,
  49         }
  50
  51         return l, nil
  52 }
  53
  54 // Event tells the clients what the agent does as it happens. Kind is the kind
  55 // of a logged message (user, reply, tool, echo, work), or one of ready, busy,
  56 // idle, error, and subagent, which tells that the subagent Text starts a run. ID is the ID of a logged message. Source names the client that
  57 // sent a user message, such as terminal or telegram, so that each client can
  58 // show the messages of the others.
  59 type Event struct {
  60         Kind   string `json:"kind"`
  61         ID     int    `json:"id,omitempty"`
  62         Text   string `json:"text,omitempty"`
  63         Source string `json:"source,omitempty"`
  64 }
  65
  66 // TurnLoop runs the turns of one agent. A user message goes to the log when it
  67 // comes, and it wakes the loop. A turn waits until every message before it is
  68 // summarized. Then the turn runs a Chain whose first request holds the prompt,
  69 // the tools, and the input of turnInput.
  70 //
  71 // The view goes out as the messages of a ViewLayout, so the first request of a
  72 // turn reads the view of the turn before from the prompt cache. User messages
  73 // that come during a turn join the next request of its chain after the tool
  74 // results, and so do the reports of the subagents. The subagents run until the
  75 // loop ends.
  76 type TurnLoop struct {
  77         agent  *Agent
  78         memory *Memory
  79         client ChainClient
  80         tools  *Toolbox
  81         runs   *SubagentRuns
  82         layout *ViewLayout
  83         clock  func() time.Time
  84         notify func(Event)
  85         wake   chan struct{}
  86
  87         mutex   sync.Mutex
  88         pending []Message
  89         cancel  context.CancelFunc
  90         failure error
  91 }
  92
  93 // Submit logs a user message from the client source and wakes the loop. The
  94 // message joins the next request: the first one of a new turn, or the next one
  95 // of a running turn.
  96 func (l *TurnLoop) Submit(source, text string) (Message, error) {
  97         message, err := l.record("user", source, text)
  98         if err != nil {
  99                 return Message{}, err
 100         }
 101
 102         l.mutex.Lock()
 103         l.pending = append(l.pending, message)
 104         l.mutex.Unlock()
 105
 106         select {
 107         case l.wake <- struct{}{}:
 108         default:
 109         }
 110
 111         return message, nil
 112 }
 113
 114 // Report logs the report text of a subagent as a work message and tells the
 115 // clients. Unless the subagent was stopped, the report joins the waiting
 116 // messages and wakes the loop, as a user message of Submit does. A failed write
 117 // stops the loop, because the log refuses all writes after it.
 118 func (l *TurnLoop) Report(text string, stopped bool) {
 119         message, err := l.record("work", "", text)
 120
 121         l.mutex.Lock()
 122         switch {
 123         case err != nil:
 124                 l.failure = fmt.Errorf("%w: %w", errChainLogWrite, err)
 125         case !stopped:
 126                 l.pending = append(l.pending, message)
 127         }
 128         l.mutex.Unlock()
 129
 130         select {
 131         case l.wake <- struct{}{}:
 132         default:
 133         }
 134 }
 135
 136 // Interrupt ends the running turn. A running tool is killed, and the turn sends
 137 // no further request.
 138 func (l *TurnLoop) Interrupt() bool {
 139         l.mutex.Lock()
 140         defer l.mutex.Unlock()
 141
 142         if l.cancel == nil {
 143                 return false
 144         }
 145
 146         l.cancel()
 147
 148         return true
 149 }
 150
 151 // Busy tells if a turn runs.
 152 func (l *TurnLoop) Busy() bool {
 153         l.mutex.Lock()
 154         defer l.mutex.Unlock()
 155
 156         return l.cancel != nil
 157 }
 158
 159 // Append logs a message and tells the clients. The toolbox logs its calls and
 160 // results through it.
 161 func (l *TurnLoop) Append(kind, text string) (Message, error) {
 162         return l.record(kind, "", text)
 163 }
 164
 165 // Flush makes the log durable.
 166 func (l *TurnLoop) Flush(ctx context.Context) error {
 167         return l.memory.Flush(ctx)
 168 }
 169
 170 // Run runs a turn each time user messages or reports wait, until ctx ends. It
 171 // returns the error of a failed log write, which stops the agent. At its end
 172 // it ends the runs of the subagents and waits for them.
 173 func (l *TurnLoop) Run(ctx context.Context) error {
 174         defer l.runs.Close()
 175
 176         for {
 177                 for l.waiting() {
 178                         err := l.turn(ctx)
 179                         if err != nil {
 180                                 return err
 181                         }
 182                 }
 183
 184                 select {
 185                 case <-l.wake:
 186                 case <-ctx.Done():
 187                         return ctx.Err()
 188                 }
 189
 190                 err := l.failed()
 191                 if err != nil {
 192                         return err
 193                 }
 194         }
 195 }
 196
 197 // record logs a message and tells the clients, with the client source of a user
 198 // message.
 199 func (l *TurnLoop) record(kind, source, text string) (Message, error) {
 200         message, err := l.memory.Append(kind, text)
 201         if err != nil {
 202                 return Message{}, err
 203         }
 204
 205         l.notify(Event{
 206                 Kind:   kind,
 207                 ID:     message.ID,
 208                 Text:   text,
 209                 Source: source,
 210         })
 211
 212         return message, nil
 213 }
 214
 215 // waiting tells if user messages or reports wait for a request.
 216 func (l *TurnLoop) waiting() bool {
 217         l.mutex.Lock()
 218         defer l.mutex.Unlock()
 219
 220         return len(l.pending) > 0
 221 }
 222
 223 // failed returns the error of a failed write of a report.
 224 func (l *TurnLoop) failed() error {
 225         l.mutex.Lock()
 226         defer l.mutex.Unlock()
 227
 228         return l.failure
 229 }
 230
 231 // turn runs one turn. A failed or interrupted model call ends the turn and
 232 // tells the clients. The messages of the turn stay in the log and the view, and
 233 // the next turn starts only with a new message, so a lasting failure such as a
 234 // needed login does not repeat. Only a failed log write is an error of turn.
 235 func (l *TurnLoop) turn(ctx context.Context) error {
 236         ctx, cancel := context.WithCancel(ctx)
 237         defer cancel()
 238
 239         l.mutex.Lock()
 240         l.cancel = cancel
 241         l.mutex.Unlock()
 242
 243         defer func() {
 244                 l.mutex.Lock()
 245                 l.cancel = nil
 246                 l.mutex.Unlock()
 247         }()
 248
 249         l.notify(Event{
 250                 Kind: "busy",
 251         })
 252
 253         err := l.chain(ctx)
 254         if err != nil && !errors.Is(err, errChainLogWrite) {
 255                 l.notify(Event{
 256                         Kind: "error",
 257                         Text: turnError(ctx, err),
 258                 })
 259
 260                 err = nil
 261         }
 262
 263         l.notify(Event{
 264                 Kind: "idle",
 265         })
 266
 267         return err
 268 }
 269
 270 // chain runs the request chain of a turn until the final answer.
 271 func (l *TurnLoop) chain(ctx context.Context) error {
 272         messages := l.take()
 273         if len(messages) == 0 {
 274                 return nil
 275         }
 276
 277         err := l.memory.Wait(ctx)
 278         if err != nil {
 279                 return err
 280         }
 281
 282         instructions, err := l.agent.Prompt()
 283         if err != nil {
 284                 return err
 285         }
 286
 287         skip := make(map[int]bool, len(messages))
 288         for _, message := range messages {
 289                 skip[message.ID] = true
 290         }
 291
 292         view, err := l.layout.Split(l.memory.ViewLines(skip))
 293         if err != nil {
 294                 return err
 295         }
 296
 297         chain := &Chain{
 298                 client: l.client,
 299                 tools:  l.tools,
 300                 request: ModelRequest{
 301                         Model:        l.agent.config.Model,
 302                         Effort:       l.agent.config.Effort,
 303                         Instructions: instructions,
 304                         Tools:        promptTools(),
 305                 },
 306                 take: l.take,
 307         }
 308
 309         _, err = chain.Run(ctx, turnInput(view, l.state(), messages))
 310
 311         return err
 312 }
 313
 314 // take returns the user messages that wait for a request, and removes them.
 315 func (l *TurnLoop) take() []Message {
 316         l.mutex.Lock()
 317         defer l.mutex.Unlock()
 318
 319         messages := l.pending
 320         l.pending = nil
 321
 322         return messages
 323 }
 324
 325 // state returns the per-turn state: the end of the view, and the date and time
 326 // of the turn.
 327 func (l *TurnLoop) state() string {
 328         return "</chat>\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00")
 329 }
 330
 331 // turnError returns the text of the error that ended a turn.
 332 func turnError(ctx context.Context, err error) string {
 333         if ctx.Err() != nil {
 334                 return "interrupted"
 335         }
 336
 337         return err.Error()
 338 }
 339
 340 // turnInput returns the input of the first request of a turn, stable parts
 341 // first: a user message for each view message, one with the per-turn state,
 342 // and one for each new message, whole. The view leaves out the leaf lines of
 343 // the new messages.
 344 func turnInput(view []string, state string, messages []Message) []any {
 345         input := make([]any, 0, len(view)+1+len(messages))
 346         for _, text := range view {
 347                 input = append(input, modelUserMessage(text))
 348         }
 349
 350         input = append(input, modelUserMessage(state))
 351         for _, message := range messages {
 352                 input = append(input, modelUserMessage(message.Text))
 353         }
 354
 355         return input
 356 }