X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/38a9b0502a1ccefc6ccb3472c4a21805b222caa6..b2fc2d6a737dd5bab25e57ba822dd07258de93e6:/turn.go diff --git a/turn.go b/turn.go index c6ab5e9..2c1d9c2 100644 --- a/turn.go +++ b/turn.go @@ -4,15 +4,18 @@ import ( "context" "errors" "fmt" + "log/slog" "path/filepath" "sync" "time" ) // newTurnLoop returns the loop of agent with its memory, its subagents, and the -// WhatsApp link 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, client ChainClient, clock func() time.Time, notify func(Event)) (*TurnLoop, error) { +// 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. 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 +27,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,17 +42,25 @@ 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, subagents: subagents, runs: l.runs, whatsapp: whatsapp, + schedules: schedules, workdir: agent.config.Workdir, + agentName: agent.config.Name, + images: agent.Images(), } return l, nil @@ -78,17 +90,23 @@ type Event struct { // 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. +// 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 - notify func(Event) - wake chan 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 @@ -106,6 +124,7 @@ func (l *TurnLoop) Submit(source, text string) (Message, error) { } l.mutex.Lock() + l.pending = append(l.pending, message) l.mutex.Unlock() @@ -161,6 +180,7 @@ func (l *TurnLoop) Wake(messages ...Message) { } l.mutex.Lock() + l.pending = append(l.pending, messages...) l.mutex.Unlock() @@ -223,10 +243,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) } } @@ -238,7 +263,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) } } } @@ -279,6 +304,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() @@ -297,19 +323,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() }() @@ -318,8 +351,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), @@ -328,6 +377,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", }) @@ -336,20 +389,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)) @@ -359,7 +412,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{ @@ -369,14 +422,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(), + } + + 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) + } } - _, err = chain.Run(ctx, turnInput(view, l.state(), messages)) + reply, err := chain.Run(ctx, input) + if err != nil { + return "", fmt.Errorf("run turn chain: %w", err) + } - return err + return reply, nil } // take returns the user messages that wait for a request, and removes them. @@ -386,14 +451,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.