X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/e737005f89c7c663e1636830001730ad79396bdc..c781f79f3c8cb9a7577ee1574d196988da2719b3:/turn.go diff --git a/turn.go b/turn.go index 837ac7c..b5aa4cf 100644 --- a/turn.go +++ b/turn.go @@ -4,15 +4,66 @@ import ( "context" "errors" "fmt" + "log/slog" "path/filepath" "sync" "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. -func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *WhatsApp, schedules *Schedules, client ChainClient, clock func() time.Time, notify func(Event)) (*TurnLoop, error) { +// and notify gets each event. The clock gives the date of a turn. logger gets +// 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 { return nil, err @@ -24,6 +75,7 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *W client: client, layout: layout, clock: clock, + logger: logger, notify: notify, wake: make(chan struct{}, 1), } @@ -38,10 +90,15 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *W report: l.Report, note: l.note, fail: l.fail, + logger: logger, notify: notify, running: make(map[string]*SubagentRun), } + if agent.config.Parent != "" { + whatsapp = nil + } + l.tools = &Toolbox{ log: l, chat: memory, @@ -50,51 +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 - 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 @@ -107,6 +128,7 @@ func (l *TurnLoop) Submit(source, text string) (Message, error) { } l.mutex.Lock() + l.pending = append(l.pending, message) l.mutex.Unlock() @@ -162,6 +184,7 @@ func (l *TurnLoop) Wake(messages ...Message) { } l.mutex.Lock() + l.pending = append(l.pending, messages...) l.mutex.Unlock() @@ -224,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 err + return fmt.Errorf("check turn loop failure: %w", err) + } + + err = l.turn(ctx) + if err != nil { + return fmt.Errorf("run turn: %w", err) } } @@ -239,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) } } } @@ -280,6 +308,7 @@ func (l *TurnLoop) waiting() bool { // 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() @@ -298,19 +327,26 @@ func (l *TurnLoop) failed() error { } // turn runs one turn. A failed or interrupted model call ends the turn and -// tells the clients. 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. +// tells the clients. A failure also goes to the logger, so that it is not lost +// 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 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() l.mutex.Lock() + l.cancel = cancel l.mutex.Unlock() defer func() { l.mutex.Lock() + l.cancel = nil l.mutex.Unlock() }() @@ -319,8 +355,24 @@ 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) + } + l.notify(Event{ Kind: "error", Text: turnError(ctx, err), @@ -329,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", }) @@ -337,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)) @@ -360,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{ @@ -370,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. @@ -387,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.