git.giacomolli.uk - onechat/blob - turn.go

   1 package main
   2
   3 import (
   4         "context"
   5         "errors"
   6         "fmt"
   7         "log"
   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 *log.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 *log.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         l.pending = append(l.pending, message)
 116         l.mutex.Unlock()
 117
 118         select {
 119         case l.wake <- struct{}{}:
 120         default:
 121         }
 122
 123         return message, nil
 124 }
 125
 126 // Report logs the report text of the subagent name as a work message and tells
 127 // the clients, with the status of the run. Unless the subagent was stopped, the
 128 // report joins the waiting messages and wakes the loop, as a user message of
 129 // Submit does. A failed write stops the loop, because the log refuses all
 130 // writes after it.
 131 func (l *TurnLoop) Report(name, text, status string) {
 132         message, err := l.Work(name, text, status)
 133         switch {
 134         case err != nil:
 135                 l.fail(fmt.Errorf("%w: %w", errChainLogWrite, err))
 136         case status != subagentStopped:
 137                 l.Wake(message)
 138         }
 139 }
 140
 141 // Work logs text as a work message of source, such as the name of a subagent,
 142 // and tells the clients, with status. The message starts no turn until it goes
 143 // to Wake.
 144 func (l *TurnLoop) Work(source, text, status string) (Message, error) {
 145         message, err := l.memory.Append("work", text)
 146         if err != nil {
 147                 return Message{}, err
 148         }
 149
 150         l.notify(Event{
 151                 Kind:   "work",
 152                 ID:     message.ID,
 153                 Text:   text,
 154                 Source: source,
 155                 Status: status,
 156         })
 157
 158         return message, nil
 159 }
 160
 161 // Wake adds the logged messages to the waiting messages and wakes the loop
 162 // once, as a user message of Submit does. They join the first request of a new
 163 // turn together, or the next request of a running turn after the tool results.
 164 func (l *TurnLoop) Wake(messages ...Message) {
 165         if len(messages) == 0 {
 166                 return
 167         }
 168
 169         l.mutex.Lock()
 170         l.pending = append(l.pending, messages...)
 171         l.mutex.Unlock()
 172
 173         select {
 174         case l.wake <- struct{}{}:
 175         default:
 176         }
 177 }
 178
 179 // StopSubagent ends the run of the subagent name and tells if it ran. Its
 180 // report is logged, but it starts no turn.
 181 func (l *TurnLoop) StopSubagent(name string) (bool, error) {
 182         return l.runs.Stop(name)
 183 }
 184
 185 // Interrupt ends the running turn. A running tool is killed, and the turn sends
 186 // no further request.
 187 func (l *TurnLoop) Interrupt() bool {
 188         l.mutex.Lock()
 189         defer l.mutex.Unlock()
 190
 191         if l.cancel == nil {
 192                 return false
 193         }
 194
 195         l.cancel()
 196
 197         return true
 198 }
 199
 200 // Busy tells if a turn runs.
 201 func (l *TurnLoop) Busy() bool {
 202         l.mutex.Lock()
 203         defer l.mutex.Unlock()
 204
 205         return l.cancel != nil
 206 }
 207
 208 // Append logs a message and tells the clients. The toolbox logs its calls and
 209 // results through it.
 210 func (l *TurnLoop) Append(kind, text string) (Message, error) {
 211         return l.record(kind, "", text)
 212 }
 213
 214 // Flush makes the log durable.
 215 func (l *TurnLoop) Flush(ctx context.Context) error {
 216         return l.memory.Flush(ctx)
 217 }
 218
 219 // Run runs a turn each time user messages or reports wait, until ctx ends. It
 220 // returns the error of a failed log write, which stops the agent. At its start
 221 // it resumes the subagents that a stop of the daemon cut, and at its end it
 222 // ends the runs of the subagents and waits for them.
 223 func (l *TurnLoop) Run(ctx context.Context) error {
 224         defer l.runs.Close()
 225
 226         err := l.runs.Resume(ctx)
 227         if err != nil {
 228                 return fmt.Errorf("resume subagents: %w", err)
 229         }
 230
 231         for {
 232                 for l.waiting() {
 233                         err := l.turn(ctx)
 234                         if err != nil {
 235                                 return err
 236                         }
 237                 }
 238
 239                 select {
 240                 case <-l.wake:
 241                 case <-ctx.Done():
 242                         return ctx.Err()
 243                 }
 244
 245                 err := l.failed()
 246                 if err != nil {
 247                         return err
 248                 }
 249         }
 250 }
 251
 252 // note logs a note and tells the clients. A note starts no turn.
 253 func (l *TurnLoop) note(text string) error {
 254         _, err := l.record("note", "", text)
 255
 256         return err
 257 }
 258
 259 // record logs a message and tells the clients, with the client source of a user
 260 // message.
 261 func (l *TurnLoop) record(kind, source, text string) (Message, error) {
 262         message, err := l.memory.Append(kind, text)
 263         if err != nil {
 264                 return Message{}, err
 265         }
 266
 267         l.notify(Event{
 268                 Kind:   kind,
 269                 ID:     message.ID,
 270                 Text:   text,
 271                 Source: source,
 272         })
 273
 274         return message, nil
 275 }
 276
 277 // waiting tells if user messages or reports wait for a request.
 278 func (l *TurnLoop) waiting() bool {
 279         l.mutex.Lock()
 280         defer l.mutex.Unlock()
 281
 282         return len(l.pending) > 0
 283 }
 284
 285 // fail stops the loop with err.
 286 func (l *TurnLoop) fail(err error) {
 287         l.mutex.Lock()
 288         l.failure = errors.Join(l.failure, err)
 289         l.mutex.Unlock()
 290
 291         select {
 292         case l.wake <- struct{}{}:
 293         default:
 294         }
 295 }
 296
 297 // failed returns the error of a failed write of a report.
 298 func (l *TurnLoop) failed() error {
 299         l.mutex.Lock()
 300         defer l.mutex.Unlock()
 301
 302         return l.failure
 303 }
 304
 305 // turn runs one turn. A failed or interrupted model call ends the turn and
 306 // tells the clients. A failure also goes to the logger, so that it is not lost
 307 // when no client is connected, but an interrupt does not. The messages of the
 308 // turn stay in the log and the view, and the next turn starts only with a new
 309 // message, so a lasting failure such as a needed login does not repeat. Only a
 310 // failed log write is an error of turn.
 311 func (l *TurnLoop) turn(ctx context.Context) error {
 312         ctx, cancel := context.WithCancel(ctx)
 313         defer cancel()
 314
 315         l.mutex.Lock()
 316         l.cancel = cancel
 317         l.mutex.Unlock()
 318
 319         defer func() {
 320                 l.mutex.Lock()
 321                 l.cancel = nil
 322                 l.mutex.Unlock()
 323         }()
 324
 325         l.notify(Event{
 326                 Kind: "busy",
 327         })
 328
 329         err := l.chain(ctx)
 330         if err != nil && !errors.Is(err, errChainLogWrite) {
 331                 if ctx.Err() == nil {
 332                         l.logger.Printf("turn failed: %v", err)
 333                 }
 334
 335                 l.notify(Event{
 336                         Kind: "error",
 337                         Text: turnError(ctx, err),
 338                 })
 339
 340                 err = nil
 341         }
 342
 343         l.notify(Event{
 344                 Kind: "idle",
 345         })
 346
 347         return err
 348 }
 349
 350 // chain runs the request chain of a turn until the final answer.
 351 func (l *TurnLoop) chain(ctx context.Context) error {
 352         messages := l.take()
 353         if len(messages) == 0 {
 354                 return nil
 355         }
 356
 357         err := l.memory.Wait(ctx)
 358         if err != nil {
 359                 return err
 360         }
 361
 362         instructions, err := l.agent.Prompt()
 363         if err != nil {
 364                 return err
 365         }
 366
 367         skip := make(map[int]bool, len(messages))
 368         for _, message := range messages {
 369                 skip[message.ID] = true
 370         }
 371
 372         view, err := l.layout.Split(l.memory.ViewLines(skip))
 373         if err != nil {
 374                 return err
 375         }
 376
 377         chain := &Chain{
 378                 client: l.client,
 379                 tools:  l.tools,
 380                 request: ModelRequest{
 381                         Model:        l.agent.config.Model,
 382                         Effort:       l.agent.config.Effort,
 383                         Instructions: instructions,
 384                         Tools:        promptTools(),
 385                 },
 386                 take: l.take,
 387         }
 388
 389         _, err = chain.Run(ctx, turnInput(view, l.state(), messages))
 390
 391         return err
 392 }
 393
 394 // take returns the user messages that wait for a request, and removes them.
 395 func (l *TurnLoop) take() []Message {
 396         l.mutex.Lock()
 397         defer l.mutex.Unlock()
 398
 399         messages := l.pending
 400         l.pending = nil
 401
 402         return messages
 403 }
 404
 405 // state returns the per-turn state: the end of the view, and the date and time
 406 // of the turn.
 407 func (l *TurnLoop) state() string {
 408         return "</chat>\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00")
 409 }
 410
 411 // turnError returns the text of the error that ended a turn.
 412 func turnError(ctx context.Context, err error) string {
 413         if ctx.Err() != nil {
 414                 return "interrupted"
 415         }
 416
 417         return err.Error()
 418 }
 419
 420 // turnInput returns the input of the first request of a turn, stable parts
 421 // first: a user message for each view message, one with the per-turn state,
 422 // and one for each new message, whole. The view leaves out the leaf lines of
 423 // the new messages.
 424 func turnInput(view []string, state string, messages []Message) []any {
 425         input := make([]any, 0, len(view)+1+len(messages))
 426         for _, text := range view {
 427                 input = append(input, modelUserMessage(text))
 428         }
 429
 430         input = append(input, modelUserMessage(state))
 431         for _, message := range messages {
 432                 input = append(input, modelUserMessage(message.Text))
 433         }
 434
 435         return input
 436 }