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