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