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