X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/522d7fd117285e4e4ce3ea5ec9b5a1722521b681..723f6e18adc6482ed913e1618c8a76e981720c2b:/turn.go diff --git a/turn.go b/turn.go index 4288604..e0bd489 100644 --- a/turn.go +++ b/turn.go @@ -4,7 +4,7 @@ import ( "context" "errors" "fmt" - "log" + "log/slog" "path/filepath" "sync" "time" @@ -13,8 +13,9 @@ import ( // 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. -func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *WhatsApp, schedules *Schedules, client ChainClient, clock func() time.Time, logger *log.Logger, notify func(Event)) (*TurnLoop, error) { +// 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 @@ -46,6 +47,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,6 +59,7 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *W whatsapp: whatsapp, schedules: schedules, workdir: agent.config.Workdir, + agentName: agent.config.Name, } return l, nil @@ -83,18 +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 - logger *log.Logger - 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 @@ -231,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) } } @@ -246,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) } } } @@ -310,8 +326,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,10 +350,22 @@ 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.Printf("turn failed: %v", err) + l.logger.Error("turn failed", "err", err) } l.notify(Event{ @@ -345,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", }) @@ -353,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)) @@ -376,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{ @@ -386,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. @@ -403,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.