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

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