X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/37d5f21b23da5fbd6b44fed41f653927669571ed..b2fc2d6a737dd5bab25e57ba822dd07258de93e6:/subagent_run.go?ds=sidebyside diff --git a/subagent_run.go b/subagent_run.go index 16cf2bd..891766c 100644 --- a/subagent_run.go +++ b/subagent_run.go @@ -5,12 +5,14 @@ import ( "encoding/json" "errors" "fmt" + "log/slog" "maps" "os" "path/filepath" "slices" "strings" "sync" + "time" ) const ( @@ -43,9 +45,10 @@ var errSubagentClosed = errors.New("the daemon stops, so no subagent starts") // 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. Stop -// ends one run, and its report says that it stopped. At the stop of the daemon -// the runs end with no report. +// report: the final answer of each chain, or the error of a failed call. The +// logger gets one line at the end of a run: done, failed with the cause, or +// stopped. Stop ends one run, and its report says that it stopped. At the stop +// of the daemon the runs end with no report. // // work.json in the directory of the agent holds the names of the running // subagents. Each start and each end of a run writes it whole, but Close does @@ -66,6 +69,7 @@ type SubagentRuns struct { report func(name, text, status string) note func(text string) error fail func(err error) + logger *slog.Logger notify func(Event) mutex sync.Mutex @@ -83,6 +87,7 @@ func (r *SubagentRuns) Spawn(ctx context.Context, tasks []string, effort *string } r.mutex.Lock() + closed := r.closed r.mutex.Unlock() @@ -133,6 +138,7 @@ func (r *SubagentRuns) Spawn(ctx context.Context, tasks []string, effort *string // 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() @@ -237,6 +243,7 @@ func (r *SubagentRuns) Resume(ctx context.Context) error { for _, name := range names { _, err = r.Tell(ctx, name, text) if err != nil { + r.logger.Error("subagent lost at restart", "name", name, "err", err) r.report(name, subagentReport(name, []string{"failed: " + err.Error()}), subagentFailed) } } @@ -249,13 +256,27 @@ func (r *SubagentRuns) Resume(ctx context.Context) error { return r.save() } +// Snapshot locks the run state, memory, and subagent logs while snapshot reads +// the agent files. The lock order agrees with Stop, which logs under the run +// lock. +func (r *SubagentRuns) Snapshot(ctx context.Context, snapshot func() error) error { + r.mutex.Lock() + defer r.mutex.Unlock() + + return r.memory.Snapshot(ctx, func() error { + return r.subagents.Snapshot(ctx, snapshot) + }) +} + // 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() @@ -313,12 +334,13 @@ func (r *SubagentRuns) effort(subagent *Subagent) (string, error) { 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 Stop and Close end it. +// start runs run in the background and logs its start. Its context keeps the +// values of ctx, but only Stop and Close end it. func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error { ctx, run.cancel = context.WithCancel(context.WithoutCancel(ctx)) r.mutex.Lock() + _, found := r.running[run.subagent.name] if r.closed || found { r.mutex.Unlock() @@ -344,6 +366,8 @@ func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error { r.workers.Add(1) r.mutex.Unlock() + run.started = r.subagents.clock() + r.logger.Info("subagent started", "name", run.subagent.name) r.notify(Event{ Kind: "subagent", Text: run.subagent.name, @@ -366,6 +390,7 @@ func (r *SubagentRuns) load() ([]string, error) { } var file SubagentWorkFile + err = json.Unmarshal(data, &file) if err != nil { return nil, fmt.Errorf("decode running subagents: %w", err) @@ -398,20 +423,25 @@ func (r *SubagentRuns) workPath() string { return filepath.Join(r.agent.dir, subagentWorkFileName) } -// 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 +// run runs the chains of run and reports and logs its end with the length of +// the run, unless Close ended it. A chain follows while messages wait after the final answer of the chain before // it, and no chain failed. A chain that Stop ends gives the reply "stopped". func (r *SubagentRuns) run(ctx context.Context, run *SubagentRun) { defer r.workers.Done() defer run.cancel() - var replies []string + var ( + replies []string + failure error + ) + stopped := false status := subagentFinished for more := true; more; { reply, err := r.chain(ctx, run) r.mutex.Lock() + stopped = run.stopped switch { case err != nil && stopped: @@ -419,17 +449,22 @@ func (r *SubagentRuns) run(ctx context.Context, run *SubagentRun) { case err != nil: reply = "failed: " + err.Error() status = subagentFailed + failure = err } more = err == nil && ctx.Err() == nil && len(run.pending) > 0 run.pending = nil + var saveErr error + if !more { delete(r.running, run.subagent.name) + if !r.closed { saveErr = r.save() } } + r.mutex.Unlock() if saveErr != nil { @@ -447,6 +482,16 @@ func (r *SubagentRuns) run(ctx context.Context, run *SubagentRun) { status = subagentStopped } + length := r.subagents.clock().Sub(run.started) + switch status { + case subagentStopped: + r.logger.Info("subagent stopped", "name", run.subagent.name, "after", length) + case subagentFailed: + r.logger.Error("subagent failed", "name", run.subagent.name, "after", length, "err", failure) + default: + r.logger.Info("subagent done", "name", run.subagent.name, "after", length) + } + r.report(run.subagent.name, subagentReport(run.subagent.name, replies), status) } @@ -470,11 +515,13 @@ func (r *SubagentRuns) chain(ctx context.Context, run *SubagentRun) (string, err input = append(input, modelUserMessage(r.state()), modelUserMessage(log)) chain := &Chain{ client: r.client, + images: r.agent.Images(), tools: &Toolbox{ log: run.subagent, chat: r.memory, subagents: r.subagents, workdir: run.subagent.WorkDir(), + images: r.agent.Images(), }, request: ModelRequest{ Model: r.agent.config.Model, @@ -505,7 +552,7 @@ func (r *SubagentRuns) take(run *SubagentRun) []Message { // SubagentRun is one run of a subagent: the view that its requests get, its // effort, the end of its context, the messages of Tell that wait for a request, -// and whether Stop ended it. The mutex of SubagentRuns protects the messages +// whether Stop ended it, and the time it started. The mutex of SubagentRuns protects the messages // and the stop. type SubagentRun struct { subagent *Subagent @@ -514,6 +561,7 @@ type SubagentRun struct { cancel context.CancelFunc pending []Message stopped bool + started time.Time } // SubagentWorkFile is the content of work.json.