X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/eeb821ff17ad79c87ca9b64cd9be26f8612f4a03..add3cf6416e33c24331fd1e90e99284f44d2aae2:/subagent_run.go?ds=sidebyside diff --git a/subagent_run.go b/subagent_run.go index 4a88800..891766c 100644 --- a/subagent_run.go +++ b/subagent_run.go @@ -2,12 +2,17 @@ package main import ( "context" + "encoding/json" "errors" "fmt" + "log/slog" + "maps" "os" + "path/filepath" "slices" "strings" "sync" + "time" ) const ( @@ -17,6 +22,10 @@ const ( // subagentNoteEffort follows subagentNote when the subagent has its own // effort. subagentNoteEffort = " at effort " + + // subagentWorkFileName is the file of the names of the running + // subagents in the directory of an agent. + subagentWorkFileName = "work.json" ) // The statuses of the end of a run, in the work event of its report. @@ -36,9 +45,16 @@ 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 +// not, so after a stop of the daemon it names the runs that the stop cut, and +// Resume goes on with them. The logs alone cannot tell this, because a stop +// can cut a run just after its final reply. // // 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 @@ -52,6 +68,8 @@ type SubagentRuns struct { state func() string report func(name, text, status string) note func(text string) error + fail func(err error) + logger *slog.Logger notify func(Event) mutex sync.Mutex @@ -69,6 +87,7 @@ func (r *SubagentRuns) Spawn(ctx context.Context, tasks []string, effort *string } r.mutex.Lock() + closed := r.closed r.mutex.Unlock() @@ -119,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() @@ -204,13 +224,59 @@ func (r *SubagentRuns) Stop(name string) (bool, error) { return true, nil } +// Resume goes on with the runs that a stop of the daemon cut, as work.json +// names them. The chat gets the note "The daemon restarted, cutting NAMES; +// nothing was lost: go on.", and each cut subagent gets the same text as a +// message of Tell. A subagent that cannot go on gets a failed report. +func (r *SubagentRuns) Resume(ctx context.Context) error { + names, err := r.load() + if err != nil || len(names) == 0 { + return err + } + + text := "The daemon restarted, cutting " + strings.Join(names, ", ") + "; nothing was lost: go on." + err = r.note(text) + if err != nil { + return err + } + + 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) + } + } + + // A subagent that cannot go on does not run, so work.json no longer + // names it, and the next start does not try it again. + r.mutex.Lock() + defer r.mutex.Unlock() + + 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() @@ -268,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() @@ -287,9 +354,20 @@ func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error { } r.running[run.subagent.name] = run + err := r.save() + if err != nil { + delete(r.running, run.subagent.name) + r.mutex.Unlock() + run.cancel() + + return err + } + 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, @@ -300,20 +378,70 @@ func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error { return nil } -// 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 +// load reads the names of work.json. A missing file names none. +func (r *SubagentRuns) load() ([]string, error) { + data, err := os.ReadFile(r.workPath()) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + + if err != nil { + return nil, fmt.Errorf("read running subagents: %w", err) + } + + var file SubagentWorkFile + + err = json.Unmarshal(data, &file) + if err != nil { + return nil, fmt.Errorf("decode running subagents: %w", err) + } + + return file.Running, nil +} + +// save writes the names of the running subagents to work.json. The caller +// holds the mutex. +func (r *SubagentRuns) save() error { + file := SubagentWorkFile{ + Running: append([]string{}, slices.Sorted(maps.Keys(r.running))...), + } + + data, err := json.Marshal(file) + if err != nil { + return fmt.Errorf("encode running subagents: %w", err) + } + + err = writeFileAtomic(r.workPath(), append(data, '\n')) + if err != nil { + return fmt.Errorf("save running subagents: %w", err) + } + + return nil +} + +func (r *SubagentRuns) workPath() string { + return filepath.Join(r.agent.dir, subagentWorkFileName) +} + +// 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: @@ -321,15 +449,28 @@ 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 { + r.fail(saveErr) + } + replies = append(replies, reply) } @@ -341,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) } @@ -364,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, @@ -399,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 @@ -408,6 +561,12 @@ type SubagentRun struct { cancel context.CancelFunc pending []Message stopped bool + started time.Time +} + +// SubagentWorkFile is the content of work.json. +type SubagentWorkFile struct { + Running []string `json:"running"` } // subagentReport returns the report of the subagent name with its replies: the