X-Git-Url: https://git.giacomolli.uk/onechat/blobdiff_plain/d7a967f43fae27bcabc6072da3b5a2f923d47764..12bd9e96f3b3e67290484e3829461e054110db27:/subagent_run.go diff --git a/subagent_run.go b/subagent_run.go index 67cf79d..faa6207 100644 --- a/subagent_run.go +++ b/subagent_run.go @@ -5,7 +5,7 @@ import ( "encoding/json" "errors" "fmt" - "log" + "log/slog" "maps" "os" "path/filepath" @@ -69,7 +69,7 @@ type SubagentRuns struct { report func(name, text, status string) note func(text string) error fail func(err error) - logger *log.Logger + logger *slog.Logger notify func(Event) mutex sync.Mutex @@ -87,6 +87,7 @@ func (r *SubagentRuns) Spawn(ctx context.Context, tasks []string, effort *string } r.mutex.Lock() + closed := r.closed r.mutex.Unlock() @@ -137,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() @@ -241,7 +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.Printf("subagent %s lost at restart: %v", name, err) + r.logger.Error("subagent lost at restart", "name", name, "err", err) r.report(name, subagentReport(name, []string{"failed: " + err.Error()}), subagentFailed) } } @@ -254,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() @@ -324,6 +340,7 @@ 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() @@ -350,7 +367,7 @@ func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error { r.mutex.Unlock() run.started = r.subagents.clock() - r.logger.Printf("subagent %s started", run.subagent.name) + r.logger.Info("subagent started", "name", run.subagent.name) r.notify(Event{ Kind: "subagent", Text: run.subagent.name, @@ -373,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) @@ -406,20 +424,25 @@ func (r *SubagentRuns) workPath() string { } // 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". +// 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 failure error + 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: @@ -432,13 +455,17 @@ func (r *SubagentRuns) run(ctx context.Context, run *SubagentRun) { 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 { @@ -456,14 +483,14 @@ func (r *SubagentRuns) run(ctx context.Context, run *SubagentRun) { status = subagentStopped } - length := durationText(r.subagents.clock().Sub(run.started)) + length := r.subagents.clock().Sub(run.started) switch status { case subagentStopped: - r.logger.Printf("subagent %s stopped after %s", run.subagent.name, length) + r.logger.Info("subagent stopped", "name", run.subagent.name, "after", length) case subagentFailed: - r.logger.Printf("subagent %s failed after %s: %v", run.subagent.name, length, failure) + r.logger.Error("subagent failed", "name", run.subagent.name, "after", length, "err", failure) default: - r.logger.Printf("subagent %s done after %s", run.subagent.name, length) + r.logger.Info("subagent done", "name", run.subagent.name, "after", length) } r.report(run.subagent.name, subagentReport(run.subagent.name, replies), status) @@ -489,11 +516,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, @@ -524,8 +553,8 @@ 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, -// whether Stop ended it, and the time it started. The mutex of SubagentRuns protects the messages -// and the stop. +// whether Stop ended it, and the time it started. The mutex of SubagentRuns +// protects the messages and the stop. type SubagentRun struct { subagent *Subagent view []string