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