From: Anderson A. Giacomolli Date: Sun, 8 Nov 2026 14:03:18 +0000 (-0300) Subject: Resume the subagents that a restart cut X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/37d5f21b23da5fbd6b44fed41f653927669571ed?ds=sidebyside Resume the subagents that a restart cut A subagent that runs when the daemon stops goes on at the next start. work.json in the agent directory holds the names of the running subagents: each start and each end of a run writes it whole, but the stop of the daemon does not, so it names the runs that the stop cut. The logs alone cannot tell this, because a stop can cut a run just after its final reply. At the start, 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 told message, so it goes on from its log with the view of now. The note starts no turn, but the reports of the subagents do. A subagent that cannot go on gets a failed report, and work.json then no longer names it, so the next start does not try it again. A failed write of work.json stops the agent, as a failed log write does. Issue: b1ac698 --- diff --git a/README.rst b/README.rst index c94f226..7712c0e 100644 --- a/README.rst +++ b/README.rst @@ -105,7 +105,10 @@ its tool calls, and a finished one goes on from its log, and its reply comes back in the same way. When the user stops a subagent, the chat gets a note, and the report of the subagent is logged, but it starts no turn. Each model call of a subagent has the name in the ``agent`` field -of ``agent/usage.jsonl``. +of ``agent/usage.jsonl``. ``agent/work.json`` holds the names of the +running subagents. A subagent that runs when the daemon stops goes on at +the next start: the chat gets the note that the daemon restarted, and +the subagent gets the same note and goes on from its log. ``agent/config.json`` holds the settings of the agent. The first start writes it with the defaults and a random ID. Do not change the ID: the diff --git a/subagent_run.go b/subagent_run.go index 4a88800..16cf2bd 100644 --- a/subagent_run.go +++ b/subagent_run.go @@ -2,9 +2,12 @@ package main import ( "context" + "encoding/json" "errors" "fmt" + "maps" "os" + "path/filepath" "slices" "strings" "sync" @@ -17,6 +20,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. @@ -40,6 +47,12 @@ var errSubagentClosed = errors.New("the daemon stops, so no subagent starts") // 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 // split of a turn does. @@ -52,6 +65,7 @@ type SubagentRuns struct { state func() string report func(name, text, status string) note func(text string) error + fail func(err error) notify func(Event) mutex sync.Mutex @@ -204,6 +218,37 @@ 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.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() +} + // Close ends every run and waits for them. No subagent starts after Close. func (r *SubagentRuns) Close() { r.mutex.Lock() @@ -287,6 +332,15 @@ 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() @@ -300,6 +354,50 @@ func (r *SubagentRuns) start(ctx context.Context, run *SubagentRun) error { return nil } +// 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 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. A chain that Stop ends gives the reply "stopped". @@ -325,11 +423,19 @@ 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 { + r.fail(saveErr) + } + replies = append(replies, reply) } @@ -410,6 +516,11 @@ type SubagentRun struct { stopped bool } +// 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 // name in brackets, a new line, the replies, and the zoom that gives its whole // chat. diff --git a/subagent_run_test.go b/subagent_run_test.go index 5caf0d8..647b7ad 100644 --- a/subagent_run_test.go +++ b/subagent_run_test.go @@ -1,12 +1,14 @@ package main import ( + "context" "io" "net/http" "os" "path/filepath" "strconv" "strings" + "sync/atomic" "testing" "time" @@ -414,6 +416,139 @@ func TestCloseEndsTheSubagentsWithNoReport(t *testing.T) { require.Len(t, entries, 1) } +func TestRestartResumesTheCutSubagents(t *testing.T) { + server := newModelFake(t) + arrived := make(chan struct{}, 1) + release := make(chan struct{}) + t.Cleanup(func() { + close(release) + }) + + var restarted atomic.Bool + server.reply = func(w http.ResponseWriter, call ModelFakeCall) { + switch { + case subagentTestCall(call) && !restarted.Load(): + arrived <- struct{}{} + <-release + case subagentTestCall(call): + modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("Built.")}, 100, 0, 5)) + default: + modelWriteEvents(w, modelCompletedEvent([]any{modelMessageItem("Noted.")}, 100, 0, 5)) + } + } + + agent, err := openAgent(t.TempDir()) + require.NoError(t, err) + + first := &TurnEventRecorder{} + loop, _, stop := subagentTestStart(t, agent, server, first) + _, err = loop.runs.Spawn(t.Context(), []string{"Build the project."}, nil) + require.NoError(t, err) + + <-arrived + name := first.Last("subagent") + stop() + require.Zero(t, first.Count("work")) + + data, err := os.ReadFile(filepath.Join(agent.dir, subagentWorkFileName)) + require.NoError(t, err) + require.JSONEq(t, `{"running": ["`+name+`"]}`, string(data)) + + restarted.Store(true) + second := &TurnEventRecorder{} + loop, memory, stop := subagentTestStart(t, agent, server, second) + defer stop() + + second.WaitFor(t, "idle", 1) + note := "The daemon restarted, cutting " + name + "; nothing was lost: go on." + log := subagentTestChat(t, memory) + require.Equal(t, []string{ + "note: " + note, + "work: " + subagentReport(name, []string{"Built."}), + "reply: Noted.", + }, log[len(log)-3:]) + + subagent, err := loop.runs.subagents.Get(t.Context(), name) + require.NoError(t, err) + + text, err := subagent.Show() + require.NoError(t, err) + require.Equal(t, "0|note: subagent\n\n1|user: Build the project.\n\n2|user: "+note+"\n\n3|reply: Built.", text) + + data, err = os.ReadFile(filepath.Join(agent.dir, subagentWorkFileName)) + require.NoError(t, err) + require.JSONEq(t, `{"running": []}`, string(data)) +} + +func TestRestartReportsASubagentThatCannotGoOn(t *testing.T) { + server := newModelFake(t) + agent, err := openAgent(t.TempDir()) + require.NoError(t, err) + require.NoError(t, os.WriteFile(filepath.Join(agent.dir, subagentWorkFileName), []byte(`{"running": ["BlueFox"]}`), 0o600)) + + events := &TurnEventRecorder{} + _, memory, stop := subagentTestStart(t, agent, server, events) + defer stop() + + events.WaitFor(t, "idle", 1) + require.Equal(t, []string{ + "note: The daemon restarted, cutting BlueFox; nothing was lost: go on.", + "work: " + subagentReport("BlueFox", []string{"failed: no agent BlueFox"}), + "reply: a line", + }, subagentTestChat(t, memory)) + require.Equal(t, subagentFailed, events.LastEvent("work").Status) + + data, err := os.ReadFile(filepath.Join(agent.dir, subagentWorkFileName)) + require.NoError(t, err) + require.JSONEq(t, `{"running": []}`, string(data)) +} + +// subagentTestStart runs a loop of agent with its memory and subagents in the +// agent directory, as the daemon does. The returned function stops the loop +// and closes the files. +func subagentTestStart(t *testing.T, agent *Agent, server *ModelFake, events *TurnEventRecorder) (*TurnLoop, *Memory, func()) { + t.Helper() + + agent.config.Workdir = t.TempDir() + clock := newTestClock() + memory, err := openMemory(t.Context(), agent.dir, clock.Now, newFakeSummarizer()) + require.NoError(t, err) + + subagents := newSubagents(filepath.Join(agent.dir, subagentDirName), clock.Now, subagentNameRandom) + client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl")) + loop, err := newTurnLoop(agent, memory, subagents, client, clock.Now, events.Add) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(t.Context()) + done := make(chan error) + go func() { + done <- loop.Run(ctx) + }() + + return loop, memory, func() { + cancel() + require.ErrorIs(t, <-done, context.Canceled) + require.NoError(t, subagents.Close()) + require.NoError(t, memory.Close()) + } +} + +// subagentTestChat returns each message of the chat in memory as +// "kind: text". +func subagentTestChat(t *testing.T, memory *Memory) []string { + t.Helper() + + var lines []string + for id := range memory.Len() { + message, err := memory.Message(id) + require.NoError(t, err) + + lines = append(lines, message.Kind+": "+message.Text) + } + + return lines +} + // subagentTestCall tells if call is a call of a subagent. func subagentTestCall(call ModelFakeCall) bool { instructions, _ := call.body["instructions"].(string) diff --git a/turn.go b/turn.go index deb47d9..93052e2 100644 --- a/turn.go +++ b/turn.go @@ -37,6 +37,7 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, client Chai state: l.state, report: l.Report, note: l.note, + fail: l.fail, notify: notify, running: make(map[string]*SubagentRun), } @@ -188,11 +189,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) @@ -247,6 +254,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()