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