X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/47f69638bc96100f2e446f63c7c3689023156e64..3391e4eaab1c20e3b172c443186b1947622a679e:/turn.go diff --git a/turn.go b/turn.go index 3c1c921..637d783 100644 --- a/turn.go +++ b/turn.go @@ -4,15 +4,17 @@ 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. +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 +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), } @@ -37,6 +40,8 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, client Chai state: l.state, report: l.Report, note: l.note, + fail: l.fail, + logger: logger, notify: notify, running: make(map[string]*SubagentRun), } @@ -46,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, } @@ -53,15 +60,18 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, client Chai } // Event tells the clients what the agent does as it happens. Kind is the kind -// of a logged message (user, reply, tool, echo, work), or one of ready, busy, -// idle, error, and subagent, which tells that the subagent Text starts a run. ID is the ID of a logged message. Source names the client that -// sent a user message, such as terminal or telegram, so that each client can -// show the messages of the others. +// of a logged message (user, reply, tool, echo, note, work), or one of ready, +// busy, idle, error, and subagent, which tells that the subagent Text starts a +// run. ID is the ID of a logged message. Source names the client that sent a +// user message, such as terminal or telegram, so that each client can show the +// messages of the others, or the subagent of a report. Status tells how the run +// of a report ended: finished, stopped, or failed. type Event struct { Kind string `json:"kind"` ID int `json:"id,omitempty"` Text string `json:"text,omitempty"` Source string `json:"source,omitempty"` + Status string `json:"status,omitempty"` } // TurnLoop runs the turns of one agent. A user message goes to the log when it @@ -82,6 +92,7 @@ type TurnLoop struct { runs *SubagentRuns layout *ViewLayout clock func() time.Time + logger *slog.Logger notify func(Event) wake chan struct{} @@ -101,6 +112,7 @@ func (l *TurnLoop) Submit(source, text string) (Message, error) { } l.mutex.Lock() + l.pending = append(l.pending, message) l.mutex.Unlock() @@ -112,20 +124,52 @@ func (l *TurnLoop) Submit(source, text string) (Message, error) { return message, nil } -// Report logs the report text of a subagent as a work message and tells the -// clients. Unless the subagent was stopped, the report joins the waiting -// messages and wakes the loop, as a user message of Submit does. A failed write -// stops the loop, because the log refuses all writes after it. -func (l *TurnLoop) Report(text string, stopped bool) { - message, err := l.record("work", "", text) - - l.mutex.Lock() +// Report logs the report text of the subagent name as a work message and tells +// the clients, with the status of the run. Unless the subagent was stopped, the +// report joins the waiting messages and wakes the loop, as a user message of +// 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.Work(name, text, status) switch { case err != nil: - l.failure = fmt.Errorf("%w: %w", errChainLogWrite, err) - case !stopped: - l.pending = append(l.pending, message) + 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. 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 { @@ -175,11 +219,17 @@ func (l *TurnLoop) Flush(ctx context.Context) error { } // Run runs a turn each time user messages or reports wait, until ctx ends. It -// returns the error of a failed log write, which stops the agent. At its end -// it ends the runs of the subagents and waits for them. +// returns the error of a failed log write, which stops the agent. At its start +// it resumes the subagents that a stop of the daemon cut, and at its end it +// ends the runs of the subagents and waits for them. func (l *TurnLoop) Run(ctx context.Context) error { defer l.runs.Close() + err := l.runs.Resume(ctx) + if err != nil { + return fmt.Errorf("resume subagents: %w", err) + } + for { for l.waiting() { err := l.turn(ctx) @@ -234,6 +284,19 @@ func (l *TurnLoop) waiting() bool { return len(l.pending) > 0 } +// 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() + + select { + case l.wake <- struct{}{}: + default: + } +} + // failed returns the error of a failed write of a report. func (l *TurnLoop) failed() error { l.mutex.Lock() @@ -243,19 +306,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() }() @@ -266,6 +333,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),