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