X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/56d959f2d573eef1be5c04d0bf1659fa30d36e7c..855a45e6448537c3eed1ad6ec7f1ca9d6caf91fc:/turn.go?ds=sidebyside diff --git a/turn.go b/turn.go index d6a1c41..e0bd489 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 and its subagents. 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, client ChainClient, clock func() time.Time, notify func(Event)) (*TurnLoop, 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. 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, client Chai client: client, layout: layout, clock: clock, + logger: logger, notify: notify, wake: make(chan struct{}, 1), } @@ -38,16 +42,24 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, client Chai 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, } return l, nil @@ -77,17 +89,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 @@ -105,6 +123,7 @@ func (l *TurnLoop) Submit(source, text string) (Message, error) { } l.mutex.Lock() + l.pending = append(l.pending, message) l.mutex.Unlock() @@ -122,18 +141,19 @@ func (l *TurnLoop) Submit(source, text string) (Message, error) { // Submit does. A failed write stops the loop, because the log refuses all // writes after it. func (l *TurnLoop) Report(name, text, status string) { - _, err := l.Work(name, text, status, status != subagentStopped) - if err != nil { + message, err := l.Work(name, text, status) + switch { + case err != nil: l.fail(fmt.Errorf("%w: %w", errChainLogWrite, err)) + case status != subagentStopped: + l.Wake(message) } } // Work logs text as a work message of source, such as the name of a subagent, -// and tells the clients, with status. With wake, the message joins the waiting -// messages and wakes the loop, as a user message of Submit does: it joins the -// first request of a new turn, or the next request of a running turn after the -// tool results. -func (l *TurnLoop) Work(source, text, status string, wake bool) (Message, error) { +// and tells the clients, with status. The message starts no turn until it goes +// to Wake. +func (l *TurnLoop) Work(source, text, status string) (Message, error) { message, err := l.memory.Append("work", text) if err != nil { return Message{}, err @@ -147,20 +167,26 @@ func (l *TurnLoop) Work(source, text, status string, wake bool) (Message, error) Status: status, }) - if !wake { - return message, nil + return message, nil +} + +// Wake adds the logged messages to the waiting messages and wakes the loop +// once, as a user message of Submit does. They join the first request of a new +// turn together, or the next request of a running turn after the tool results. +func (l *TurnLoop) Wake(messages ...Message) { + if len(messages) == 0 { + return } l.mutex.Lock() - l.pending = append(l.pending, message) + + l.pending = append(l.pending, messages...) 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 @@ -216,10 +242,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) } } @@ -231,7 +262,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) } } } @@ -272,6 +303,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() @@ -290,19 +322,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() }() @@ -311,8 +350,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), @@ -321,6 +376,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", }) @@ -329,20 +388,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)) @@ -352,7 +411,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{ @@ -362,14 +421,17 @@ 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, } - _, err = chain.Run(ctx, turnInput(view, l.state(), messages)) + reply, err := chain.Run(ctx, turnInput(view, l.state(), messages)) + 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. @@ -379,14 +441,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.