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