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