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