X-Git-Url: http://git.giacomolli.uk/onechat/blobdiff_plain/ddd0a50f7b801112376d7fb27bfd05d2fc26c2fd..56d959f2d573eef1be5c04d0bf1659fa30d36e7c:/turn.go diff --git a/turn.go b/turn.go index 02a4a28..d6a1c41 100644 --- a/turn.go +++ b/turn.go @@ -36,6 +36,8 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, client Chai client: client, state: l.state, report: l.Report, + note: l.note, + fail: l.fail, notify: notify, running: make(map[string]*SubagentRun), } @@ -52,15 +54,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 @@ -111,26 +116,57 @@ 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) +// 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) { + _, err := l.Work(name, text, status, status != subagentStopped) + if err != nil { + l.fail(fmt.Errorf("%w: %w", errChainLogWrite, err)) + } +} - l.mutex.Lock() - switch { - case err != nil: - l.failure = fmt.Errorf("%w: %w", errChainLogWrite, err) - case !stopped: - l.pending = append(l.pending, 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) { + 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, + }) + + if !wake { + return message, nil + } + + l.mutex.Lock() + l.pending = append(l.pending, message) 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 +// report is logged, but it starts no turn. +func (l *TurnLoop) StopSubagent(name string) (bool, error) { + return l.runs.Stop(name) } // Interrupt ends the running turn. A running tool is killed, and the turn sends @@ -168,11 +204,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) @@ -194,6 +236,13 @@ func (l *TurnLoop) Run(ctx context.Context) error { } } +// note logs a note and tells the clients. A note starts no turn. +func (l *TurnLoop) note(text string) error { + _, err := l.record("note", "", text) + + return err +} + // record logs a message and tells the clients, with the client source of a user // message. func (l *TurnLoop) record(kind, source, text string) (Message, error) { @@ -220,6 +269,18 @@ 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()