X-Git-Url: http://git.giacomolli.uk/onechat/blobdiff_plain/47f69638bc96100f2e446f63c7c3689023156e64..56d959f2d573eef1be5c04d0bf1659fa30d36e7c:/turn.go diff --git a/turn.go b/turn.go index 3c1c921..d6a1c41 100644 --- a/turn.go +++ b/turn.go @@ -37,6 +37,7 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, client Chai state: l.state, report: l.Report, note: l.note, + fail: l.fail, notify: notify, running: make(map[string]*SubagentRun), } @@ -53,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 @@ -112,26 +116,51 @@ 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 @@ -175,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) @@ -234,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()