X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/5a93b6124068c2a8748fc0bfbb2a2d151815bcf0..03ef9525ba0e62037b55606b851ee746967a003f:/turn.go?ds=inline diff --git a/turn.go b/turn.go index e0bd489..4652321 100644 --- a/turn.go +++ b/turn.go @@ -10,6 +10,56 @@ import ( "time" ) +// Event tells the clients what the agent does as it happens. Kind is the kind +// of a logged message (user, reply, tool, echo, note, work), or one of ready, +// busy, idle, error, and subagent, which tells that the subagent Text starts a +// run. ID is the ID of a logged message. Source names the client that sent a +// user message, such as terminal or telegram, so that each client can show the +// messages of the others, or the subagent of a report. Status tells how the run +// of a report ended: finished, stopped, or failed. A private notify receipt +// uses ID for its queue sequence and Status for accepted or duplicate, not +// processed. +type Event struct { + Kind string `json:"kind"` + ID int `json:"id,omitempty"` + Text string `json:"text,omitempty"` + Source string `json:"source,omitempty"` + Status string `json:"status,omitempty"` +} + +// TurnLoop runs the turns of one agent. A user message goes to the log when it +// comes, and it wakes the loop. A turn waits until every message before it is +// summarized. Then the turn runs a Chain whose first request holds the prompt, +// the tools, and the input of turnInput. +// +// The view goes out as the messages of a ViewLayout, so the first request of a +// turn reads the view of the turn before from the prompt cache. User messages +// that come during a turn join the next request of its chain after the tool +// results, and so do the reports of the subagents. The subagents run until the +// loop ends. Parent and child messages stay queued until the turn finishes. +// active belongs to the running turn. The mutex protects pending, cancel, and +// failure. +type TurnLoop struct { + agent *Agent + memory *Memory + client ChainClient + tools *Toolbox + runs *SubagentRuns + layout *ViewLayout + clock func() time.Time + logger *slog.Logger + notify func(Event) + wake chan struct{} + messages *DaemonMessages + openChild func(context.Context, *Agent) (*DaemonAgent, error) + active []Message + + mutex sync.Mutex + pending []Message + cancel context.CancelFunc + failure error +} + // newTurnLoop returns the loop of agent with its memory, its subagents, and the // WhatsApp link and the schedules of its tools. Its calls go through client, // and notify gets each event. The clock gives the date of a turn. logger gets @@ -60,79 +110,25 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *W schedules: schedules, workdir: agent.config.Workdir, agentName: agent.config.Name, + images: agent.Images(), } - return l, nil -} - -// Event tells the clients what the agent does as it happens. Kind is the kind -// of a logged message (user, reply, tool, echo, note, work), or one of ready, -// busy, idle, error, and subagent, which tells that the subagent Text starts a -// run. ID is the ID of a logged message. Source names the client that sent a -// user message, such as terminal or telegram, so that each client can show the -// messages of the others, or the subagent of a report. Status tells how the run -// of a report ended: finished, stopped, or failed. -type Event struct { - Kind string `json:"kind"` - ID int `json:"id,omitempty"` - Text string `json:"text,omitempty"` - Source string `json:"source,omitempty"` - Status string `json:"status,omitempty"` -} - -// TurnLoop runs the turns of one agent. A user message goes to the log when it -// comes, and it wakes the loop. A turn waits until every message before it is -// summarized. Then the turn runs a Chain whose first request holds the prompt, -// the tools, and the input of turnInput. -// -// The view goes out as the messages of a ViewLayout, so the first request of a -// turn reads the view of the turn before from the prompt cache. User messages -// that come during a turn join the next request of its chain after the tool -// results, and so do the reports of the subagents. The subagents run until the -// loop ends. Parent and child messages stay queued until the turn finishes. -// active belongs to the running turn. The mutex protects pending, cancel, and -// failure. -type TurnLoop struct { - agent *Agent - memory *Memory - client ChainClient - tools *Toolbox - runs *SubagentRuns - layout *ViewLayout - clock func() time.Time - logger *slog.Logger - notify func(Event) - wake chan struct{} - messages *DaemonMessages - openChild func(context.Context, *Agent) (*DaemonAgent, error) - active []Message - - mutex sync.Mutex - pending []Message - cancel context.CancelFunc - failure error -} - -// Submit logs a user message from the client source and wakes the loop. The -// message joins the next request: the first one of a new turn, or the next one -// of a running turn. -func (l *TurnLoop) Submit(source, text string) (Message, error) { - message, err := l.record("user", source, text) - if err != nil { - return Message{}, err + if agent.config.Parent == "" { + l.tools.imageShow = &ImageShow{store: agent.Images()} } - l.mutex.Lock() - - l.pending = append(l.pending, message) - l.mutex.Unlock() + return l, nil +} - select { - case l.wake <- struct{}{}: - default: +// state returns the per-turn state: the end of the view, and the date and time +// of the turn, then the names and responsibilities of its children. +func (l *TurnLoop) state() string { + state := "\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00") + if l.messages != nil { + state += l.messages.ChildState(l.agent.config.Name) } - return message, nil + return state } // Report logs the report text of the subagent name as a work message and tells @@ -150,6 +146,26 @@ func (l *TurnLoop) Report(name, text, status string) { } } +// note logs a note and tells the clients. A note starts no turn. +func (l *TurnLoop) note(text string) error { + _, err := l.record("note", "", text) + + return err +} + +// fail stops the loop with err. +func (l *TurnLoop) fail(err error) { + l.mutex.Lock() + + l.failure = errors.Join(l.failure, err) + l.mutex.Unlock() + + select { + case l.wake <- struct{}{}: + default: + } +} + // Work logs text as a work message of source, such as the name of a subagent, // and tells the clients, with status. The message starts no turn until it goes // to Wake. @@ -189,6 +205,46 @@ func (l *TurnLoop) Wake(messages ...Message) { } } +// record logs a message and tells the clients, with the client source of a user +// message. +func (l *TurnLoop) record(kind, source, text string) (Message, error) { + message, err := l.memory.Append(kind, text) + if err != nil { + return Message{}, err + } + + l.notify(Event{ + Kind: kind, + ID: message.ID, + Text: text, + Source: source, + }) + + return message, nil +} + +// Submit logs a user message from the client source and wakes the loop. The +// message joins the next request: the first one of a new turn, or the next one +// of a running turn. +func (l *TurnLoop) Submit(source, text string) (Message, error) { + message, err := l.record("user", source, text) + if err != nil { + return Message{}, err + } + + l.mutex.Lock() + + l.pending = append(l.pending, message) + l.mutex.Unlock() + + select { + case l.wake <- struct{}{}: + default: + } + + return message, nil +} + // StopSubagent ends the run of the subagent name and tells if it ran. Its // report is logged, but it starts no turn. func (l *TurnLoop) StopSubagent(name string) (bool, error) { @@ -229,10 +285,11 @@ func (l *TurnLoop) Flush(ctx context.Context) error { return l.memory.Flush(ctx) } -// Run runs a turn each time user messages or reports wait, until ctx ends. It -// returns the error of a failed log write, which stops the agent. At its start -// it resumes the subagents that a stop of the daemon cut, and at its end it -// ends the runs of the subagents and waits for them. +// Run starts a turn when user messages, reports, or queued external events +// wait, until ctx ends. It returns the error of a failed log write, which stops +// the agent. External events enter memory only between chains. At its start it +// resumes the subagents that a stop of the daemon cut, and at its end it ends +// the runs of the subagents and waits for them. func (l *TurnLoop) Run(ctx context.Context) error { defer l.runs.Close() @@ -242,12 +299,23 @@ func (l *TurnLoop) Run(ctx context.Context) error { } for { - for ctx.Err() == nil && l.waiting() { + for ctx.Err() == nil { err := l.failed() if err != nil { return fmt.Errorf("check turn loop failure: %w", err) } + if l.messages != nil { + err = l.messages.Notifications(l) + if err != nil { + return fmt.Errorf("prepare external events: %w", err) + } + } + + if !l.waiting() { + break + } + err = l.turn(ctx) if err != nil { return fmt.Errorf("run turn: %w", err) @@ -267,31 +335,6 @@ func (l *TurnLoop) Run(ctx context.Context) error { } } -// note logs a note and tells the clients. A note starts no turn. -func (l *TurnLoop) note(text string) error { - _, err := l.record("note", "", text) - - return err -} - -// record logs a message and tells the clients, with the client source of a user -// message. -func (l *TurnLoop) record(kind, source, text string) (Message, error) { - message, err := l.memory.Append(kind, text) - if err != nil { - return Message{}, err - } - - l.notify(Event{ - Kind: kind, - ID: message.ID, - Text: text, - Source: source, - }) - - return message, nil -} - // waiting tells if user messages or reports wait for a request. func (l *TurnLoop) waiting() bool { l.mutex.Lock() @@ -300,19 +343,6 @@ func (l *TurnLoop) waiting() bool { return len(l.pending) > 0 } -// fail stops the loop with err. -func (l *TurnLoop) fail(err error) { - l.mutex.Lock() - - l.failure = errors.Join(l.failure, err) - l.mutex.Unlock() - - select { - case l.wake <- struct{}{}: - default: - } -} - // failed returns the error of a failed write of a report. func (l *TurnLoop) failed() error { l.mutex.Lock() @@ -423,10 +453,19 @@ func (l *TurnLoop) chain(ctx context.Context) (string, error) { Instructions: instructions, Tools: l.agent.Tools(), }, - take: l.take, + take: l.take, + images: l.agent.Images(), } - reply, err := chain.Run(ctx, turnInput(view, l.state(), messages)) + input := turnInput(view, l.state(), messages) + for n, message := range messages { + input[len(input)-len(messages)+n], err = imageUserMessage(l.agent.Images(), message.Text) + if err != nil { + return "", fmt.Errorf("read turn images: %w", err) + } + } + + reply, err := chain.Run(ctx, input) if err != nil { return "", fmt.Errorf("run turn chain: %w", err) } @@ -448,17 +487,6 @@ func (l *TurnLoop) take() []Message { return messages } -// state returns the per-turn state: the end of the view, and the date and time -// of the turn, then the names and responsibilities of its children. -func (l *TurnLoop) state() string { - state := "\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00") - if l.messages != nil { - state += l.messages.ChildState(l.agent.config.Name) - } - - return state -} - // turnError returns the text of the error that ended a turn. func turnError(ctx context.Context, err error) string { if ctx.Err() != nil {