X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/ddd0a50f7b801112376d7fb27bfd05d2fc26c2fd..a85d214071706753f358f0af23cfd525c28dbfe0:/turn.go diff --git a/turn.go b/turn.go index 02a4a28..c6ab5e9 100644 --- a/turn.go +++ b/turn.go @@ -9,10 +9,10 @@ import ( "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 of its tools. 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, whatsapp *WhatsApp, client ChainClient, clock func() time.Time, notify func(Event)) (*TurnLoop, error) { layout, err := loadViewLayout(filepath.Join(agent.dir, viewLayoutFileName)) if err != nil { return nil, err @@ -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), } @@ -45,6 +47,7 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, client Chai chat: memory, subagents: subagents, runs: l.runs, + whatsapp: whatsapp, workdir: agent.config.Workdir, } @@ -52,15 +55,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,20 +117,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) - - 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 { @@ -133,6 +170,12 @@ func (l *TurnLoop) Report(text string, stopped bool) { } } +// 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 // no further request. func (l *TurnLoop) Interrupt() bool { @@ -168,11 +211,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 +243,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 +276,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()