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

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