Send messages to subagents with tell

[onechat] / subagent_run.go

diff --git a/subagent_run.go b/subagent_run.go
index 3ee98a076a22311424b3321e201878ce30e9148e..b2e5e62526526d60fbf2b99cd2cad711e75b8c54 100644 (file)
--- a/subagent_run.go
+++ b/subagent_run.go
@@ -23,15 +23,18 @@ const (
 var errSubagentClosed = errors.New("the daemon stops, so no subagent starts")
 
 // SubagentRuns runs the subagents of one agent in the background. Each run is
 var errSubagentClosed = errors.New("the daemon stops, so no subagent starts")
 
 // SubagentRuns runs the subagents of one agent in the background. Each run is

-// a goroutine with its own Chain and its own context, which only Close ends.
-// The first request of each chain holds the view of the agent at the spawn,
-// then the state of the turn, then the whole log of the subagent. The report of
-// the run goes to report: its final answer, or the error of a failed call. At
-// the stop of the daemon the runs end with no report.
+// a goroutine with its own context, which only Close ends. A run is one or more
+// request chains. The first request of each chain holds the view of the agent
+// at the start of the run, then the state of the turn, then the whole log of
+// the subagent. A message of Tell that comes during a chain joins its next
+// request after the tool results. One that comes after the final answer starts
+// one more chain, which reads the log again. The report of the run goes to
+// report: the final answer of each chain, or the error of a failed call. At the
+// stop of the daemon the runs end with no report.

 //
 //

-// Spawn splits the view with the layout of the agent, which has no lock. Spawn
-// is a tool, so it runs in the goroutine of the turns, as each split of a turn
-// does.
+// Spawn and Tell split the view with the layout of the agent, which has no
+// lock. They are tools, so they run in the goroutine of the turns, as each
+// split of a turn does.

 type SubagentRuns struct {
        agent     *Agent
        memory    *Memory
 type SubagentRuns struct {
        agent     *Agent
        memory    *Memory


@@ -102,6 +105,85 @@ func (r *SubagentRuns) Spawn(ctx context.Context, tasks []string, effort *string
        return "Started " + strings.Join(names, ", ") + ". Each one's report will reach you as a message when it finishes.", nil
 }
 
        return "Started " + strings.Join(names, ", ") + ". Each one's report will reach you as a message when it finishes.", nil
 }
 

+// Tell sends message to the subagent name. A running subagent reads it between
+// its tool calls, and its report answers it. A finished subagent goes on from
+// its log with a new run, with the view of now and the effort of its note.
+func (r *SubagentRuns) Tell(ctx context.Context, name, message string) (string, error) {
+       r.mutex.Lock()
+       run, found := r.running[name]
+       if found {
+               defer r.mutex.Unlock()
+
+               told, err := run.subagent.Append("user", message)
+               if err != nil {
+                       return "", err
+               }
+
+               run.pending = append(run.pending, told)
+
+               return "Sent. " + name + " reads it between its tool calls, and its report answers it.", nil
+       }
+
+       closed := r.closed
+       r.mutex.Unlock()
+
+       if closed {
+               return "", errSubagentClosed
+       }
+
+       subagent, err := r.subagents.Get(ctx, name)
+       if err != nil {
+               return "", err
+       }
+
+       effort, err := r.effort(subagent)
+       if err != nil {
+               return "", err
+       }
+
+       err = r.memory.Settle(ctx)
+       if err != nil {
+               return "", fmt.Errorf("wait for the summaries: %w", err)
+       }
+
+       view, err := r.layout.Split(r.memory.ViewLines(nil))
+       if err != nil {
+               return "", err
+       }
+
+       _, err = subagent.Append("user", message)
+       if err == nil {
+               err = subagent.Flush(ctx)
+       }
+
+       if err != nil {
+               return "", err
+       }
+
+       err = r.start(ctx, &SubagentRun{
+               subagent: subagent,
+               view:     view,
+               effort:   effort,
+       })
+       if err != nil {
+               return "", err
+       }
+
+       return "Sent. " + name + "'s reply will reach you as a message.", nil
+}
+
+// Close ends every run and waits for them. No subagent starts after Close.
+func (r *SubagentRuns) Close() {
+       r.mutex.Lock()
+       r.closed = true
+       for _, run := range r.running {
+               run.cancel()
+       }
+       r.mutex.Unlock()
+
+       r.workers.Wait()
+}
+

 // create makes a new subagent with its note, its task, and its work directory,
 // and starts its run.
 func (r *SubagentRuns) create(ctx context.Context, note, task string, view []string, effort string) (string, error) {
 // create makes a new subagent with its note, its task, and its work directory,
 // and starts its run.
 func (r *SubagentRuns) create(ctx context.Context, note, task string, view []string, effort string) (string, error) {


@@ -134,16 +216,41 @@ func (r *SubagentRuns) create(ctx context.Context, note, task string, view []str
        })
 }
 
        })
 }
 

+// effort returns the effort of subagent from the note at the start of its log:
+// the effort of the agent, or the effort that the note names.
+func (r *SubagentRuns) effort(subagent *Subagent) (string, error) {
+       note, err := subagent.First()
+       if err != nil {
+               return "", err
+       }
+
+       effort, own := strings.CutPrefix(note.Text, subagentNote+subagentNoteEffort)
+       switch {
+       case note.Kind != "note":
+       case own:
+               return effort, nil
+       case note.Text == subagentNote:
+               return r.agent.config.Effort, nil
+       }
+
+       return "", fmt.Errorf("the log of %s does not start with a subagent note", subagent.name)
+}
+

 // start runs run in the background. Its context keeps the values of ctx, but
 // only Close ends it.
 func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error {
        ctx, run.cancel = context.WithCancel(context.WithoutCancel(ctx))
 
        r.mutex.Lock()
 // start runs run in the background. Its context keeps the values of ctx, but
 // only Close ends it.
 func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error {
        ctx, run.cancel = context.WithCancel(context.WithoutCancel(ctx))
 
        r.mutex.Lock()

-       if r.closed {
+       _, found := r.running[run.subagent.name]
+       if r.closed || found {

                r.mutex.Unlock()
                run.cancel()
 
                r.mutex.Unlock()
                run.cancel()
 

+               if found {
+                       return fmt.Errorf("%s runs already", run.subagent.name)
+               }
+

                return errSubagentClosed
        }
 
                return errSubagentClosed
        }
 


@@ -161,35 +268,48 @@ func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error {
        return nil
 }
 
        return nil
 }
 

-// run runs the chain of run and reports its end, unless Close ended it.
+// run runs the chains of run and reports its end, unless Close ended it. A
+// chain follows while messages wait after the final answer of the chain before
+// it, and no chain failed.

 func (r *SubagentRuns) run(ctx context.Context, run *SubagentRun) {
        defer r.workers.Done()
        defer run.cancel()
 
 func (r *SubagentRuns) run(ctx context.Context, run *SubagentRun) {
        defer r.workers.Done()
        defer run.cancel()
 

-       reply := r.chain(ctx, run)
+       var replies []string
+       for more := true; more; {
+               reply, err := r.chain(ctx, run)
+               if err != nil {
+                       reply = "failed: " + err.Error()
+               }
+
+               replies = append(replies, reply)

 
 

-       r.mutex.Lock()
-       delete(r.running, run.subagent.name)
-       r.mutex.Unlock()
+               r.mutex.Lock()
+               more = err == nil && len(run.pending) > 0
+               run.pending = nil
+               if !more {
+                       delete(r.running, run.subagent.name)
+               }
+               r.mutex.Unlock()
+       }

 
        if ctx.Err() != nil {
                return
        }
 
 
        if ctx.Err() != nil {
                return
        }
 

-       r.report(subagentReport(run.subagent.name, []string{reply}), false)
+       r.report(subagentReport(run.subagent.name, replies), false)

 }
 
 }
 

-// chain runs one request chain of run and returns its final answer, or the
-// error of the chain after "failed: ".
-func (r *SubagentRuns) chain(ctx context.Context, run *SubagentRun) string {
+// chain runs one request chain of run and returns its final answer.
+func (r *SubagentRuns) chain(ctx context.Context, run *SubagentRun) (string, error) {

        instructions, err := r.agent.SubagentPrompt()
        if err != nil {
        instructions, err := r.agent.SubagentPrompt()
        if err != nil {

-               return "failed: " + err.Error()
+               return "", err

        }
 
        log, err := run.subagent.Show()
        if err != nil {
        }
 
        log, err := run.subagent.Show()
        if err != nil {

-               return "failed: " + err.Error()
+               return "", err

        }
 
        input := make([]any, 0, len(run.view)+2)
        }
 
        input := make([]any, 0, len(run.view)+2)


@@ -214,37 +334,34 @@ func (r *SubagentRuns) chain(ctx context.Context, run *SubagentRun) string {
                        Agent:        run.subagent.name,
                },
                take: func() []Message {
                        Agent:        run.subagent.name,
                },
                take: func() []Message {

-                       return nil
+                       return r.take(run)

                },
        }
 
                },
        }
 

-       reply, err := chain.Run(ctx, input)
-       if err != nil {
-               return "failed: " + err.Error()
-       }
-
-       return reply
+       return chain.Run(ctx, input)

 }
 
 }
 

-// Close ends every run and waits for them. No subagent starts after Close.
-func (r *SubagentRuns) Close() {
+// take returns the messages of Tell that wait for a request of run, and
+// removes them.
+func (r *SubagentRuns) take(run *SubagentRun) []Message {

        r.mutex.Lock()
        r.mutex.Lock()

-       r.closed = true
-       for _, run := range r.running {
-               run.cancel()
-       }
-       r.mutex.Unlock()
+       defer r.mutex.Unlock()

 
 

-       r.workers.Wait()
+       messages := run.pending
+       run.pending = nil
+
+       return messages

 }
 
 // SubagentRun is one run of a subagent: the view that its requests get, its
 }
 
 // SubagentRun is one run of a subagent: the view that its requests get, its

-// effort, and the end of its context.
+// effort, the end of its context, and the messages of Tell that wait for a
+// request. The mutex of SubagentRuns protects the messages.

 type SubagentRun struct {
        subagent *Subagent
        view     []string
        effort   string
        cancel   context.CancelFunc
 type SubagentRun struct {
        subagent *Subagent
        view     []string
        effort   string
        cancel   context.CancelFunc

+       pending  []Message

 }
 
 // subagentReport returns the report of the subagent name with its replies: the
 }
 
 // subagentReport returns the report of the subagent name with its replies: the