X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/37d5f21b23da5fbd6b44fed41f653927669571ed..6004a6d0f7af1ecf3d0ac976f50bb4f93c619887:/turn.go diff --git a/turn.go b/turn.go index 93052e2..2e5e50e 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,15 +42,22 @@ 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, } @@ -86,6 +97,7 @@ type TurnLoop struct { runs *SubagentRuns layout *ViewLayout clock func() time.Time + logger *slog.Logger notify func(Event) wake chan struct{} @@ -105,6 +117,7 @@ func (l *TurnLoop) Submit(source, text string) (Message, error) { } l.mutex.Lock() + l.pending = append(l.pending, message) l.mutex.Unlock() @@ -122,24 +135,46 @@ 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) { - message, err := l.memory.Append("work", text) - if err == nil { - l.notify(Event{ - Kind: "work", - ID: message.ID, - Text: text, - Source: name, - Status: status, - }) - } - - l.mutex.Lock() + message, err := l.Work(name, text, status) switch { case err != nil: - l.failure = fmt.Errorf("%w: %w", errChainLogWrite, err) + l.fail(fmt.Errorf("%w: %w", errChainLogWrite, err)) case status != subagentStopped: - l.pending = append(l.pending, message) + 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. 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 + } + + l.notify(Event{ + Kind: "work", + ID: message.ID, + Text: text, + Source: source, + Status: status, + }) + + 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, messages...) l.mutex.Unlock() select { @@ -257,6 +292,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() @@ -275,19 +311,23 @@ 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 write is an error of turn. func (l *TurnLoop) turn(ctx context.Context) error { 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() }() @@ -298,6 +338,10 @@ func (l *TurnLoop) turn(ctx context.Context) error { err := l.chain(ctx) 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), @@ -347,7 +391,7 @@ 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, }