X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/37d5f21b23da5fbd6b44fed41f653927669571ed..900ec0e7dad7a913190fcef9016e53495de9bdf8:/turn.go diff --git a/turn.go b/turn.go index 93052e2..40645b1 100644 --- a/turn.go +++ b/turn.go @@ -4,15 +4,17 @@ import ( "context" "errors" "fmt" + "log" "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. +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) { layout, err := loadViewLayout(filepath.Join(agent.dir, viewLayoutFileName)) if err != nil { return nil, err @@ -24,6 +26,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,6 +41,7 @@ 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), } @@ -47,6 +51,8 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, client Chai chat: memory, subagents: subagents, runs: l.runs, + whatsapp: whatsapp, + schedules: schedules, workdir: agent.config.Workdir, } @@ -86,6 +92,7 @@ type TurnLoop struct { runs *SubagentRuns layout *ViewLayout clock func() time.Time + logger *log.Logger notify func(Event) wake chan struct{} @@ -122,24 +129,45 @@ 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 { @@ -275,9 +303,11 @@ 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() @@ -298,6 +328,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.Printf("turn failed: %v", err) + } + l.notify(Event{ Kind: "error", Text: turnError(ctx, err),