X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/790279faf0fda789d93a75f6f5996e65d5de4a18..80e348724041a694002fc38f57e8868b9b9a2be9:/turn.go diff --git a/turn.go b/turn.go index 637d783..b5aa4cf 100644 --- a/turn.go +++ b/turn.go @@ -10,10 +10,59 @@ 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. +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 -// the failed turns and the ends of the subagents. +// the failed turns and the ends of the subagents. A child has no WhatsApp +// tools, even when the caller supplies a link. 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) { layout, err := loadViewLayout(filepath.Join(agent.dir, viewLayoutFileName)) if err != nil { @@ -46,6 +95,10 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *W running: make(map[string]*SubagentRun), } + if agent.config.Parent != "" { + whatsapp = nil + } + l.tools = &Toolbox{ log: l, chat: memory, @@ -54,52 +107,15 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *W whatsapp: whatsapp, 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. -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{} + if agent.config.Parent == "" { + l.tools.imageShow = &ImageShow{store: agent.Images()} + } - mutex sync.Mutex - pending []Message - cancel context.CancelFunc - failure error + return l, nil } // Submit logs a user message from the client source and wakes the loop. The @@ -231,10 +247,15 @@ func (l *TurnLoop) Run(ctx context.Context) error { } for { - for l.waiting() { - err := l.turn(ctx) + for ctx.Err() == nil && l.waiting() { + err := l.failed() + if err != nil { + return fmt.Errorf("check turn loop failure: %w", err) + } + + err = l.turn(ctx) if err != nil { - return err + return fmt.Errorf("run turn: %w", err) } } @@ -246,7 +267,7 @@ func (l *TurnLoop) Run(ctx context.Context) error { err := l.failed() if err != nil { - return err + return fmt.Errorf("check turn loop failure: %w", err) } } } @@ -310,8 +331,11 @@ func (l *TurnLoop) failed() error { // when no client is connected, but an interrupt does not. The messages of the // turn stay in the log and the view, and the next turn starts only with a new // message, so a lasting failure such as a needed login does not repeat. Only a -// failed log write is an error of turn. +// failed log or queue write is an error of turn. A daemon stop leaves taken +// parent and child messages queued for restart. A user interrupt completes them +// without a report. func (l *TurnLoop) turn(ctx context.Context) error { + runCtx := ctx ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -331,7 +355,19 @@ func (l *TurnLoop) turn(ctx context.Context) error { Kind: "busy", }) - err := l.chain(ctx) + if l.messages != nil { + l.logger.Info("turn started") + } + + reply, err := l.chain(ctx) + if l.messages != nil && !errors.Is(err, errChainLogWrite) && (err == nil || runCtx.Err() == nil) { + finishErr := l.messages.Finish(l, l.active, reply) + if finishErr != nil { + err = errors.Join(err, fmt.Errorf("%w: finish agent messages: %w", errChainLogWrite, finishErr)) + } + } + + l.active = nil if err != nil && !errors.Is(err, errChainLogWrite) { if ctx.Err() == nil { l.logger.Error("turn failed", "err", err) @@ -345,6 +381,10 @@ func (l *TurnLoop) turn(ctx context.Context) error { err = nil } + if l.messages != nil { + l.logger.Info("turn done") + } + l.notify(Event{ Kind: "idle", }) @@ -353,20 +393,20 @@ func (l *TurnLoop) turn(ctx context.Context) error { } // chain runs the request chain of a turn until the final answer. -func (l *TurnLoop) chain(ctx context.Context) error { +func (l *TurnLoop) chain(ctx context.Context) (string, error) { messages := l.take() if len(messages) == 0 { - return nil + return "", nil } err := l.memory.Wait(ctx) if err != nil { - return err + return "", fmt.Errorf("wait for turn memory: %w", err) } instructions, err := l.agent.Prompt() if err != nil { - return err + return "", fmt.Errorf("read turn prompt: %w", err) } skip := make(map[int]bool, len(messages)) @@ -376,7 +416,7 @@ func (l *TurnLoop) chain(ctx context.Context) error { view, err := l.layout.Split(l.memory.ViewLines(skip)) if err != nil { - return err + return "", fmt.Errorf("split turn view: %w", err) } chain := &Chain{ @@ -386,14 +426,26 @@ func (l *TurnLoop) chain(ctx context.Context) error { Model: l.agent.config.Model, Effort: l.agent.config.Effort, Instructions: instructions, - Tools: promptTools(), + Tools: l.agent.Tools(), }, - take: l.take, + take: l.take, + images: l.agent.Images(), } - _, 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) + } + } - return err + reply, err := chain.Run(ctx, input) + if err != nil { + return "", fmt.Errorf("run turn chain: %w", err) + } + + return reply, nil } // take returns the user messages that wait for a request, and removes them. @@ -403,14 +455,22 @@ func (l *TurnLoop) take() []Message { messages := l.pending l.pending = nil + if l.messages != nil { + l.active = append(l.active, messages...) + } return messages } // state returns the per-turn state: the end of the view, and the date and time -// of the turn. +// of the turn, then the names and responsibilities of its children. func (l *TurnLoop) state() string { - return "\n\nNow: " + l.clock().Format("Monday 2006-01-02 15:04 -07:00") + 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.