import (
"context"
+ "encoding/json"
"errors"
"fmt"
+ "maps"
"os"
+ "path/filepath"
"slices"
"strings"
"sync"
// 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.
// 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.
state func() string
report func(name, text, status string)
note func(text string) error
+ fail func(err error)
notify func(Event)
mutex sync.Mutex
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()
}
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()
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".
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)
}
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.
package main
import (
+ "context"
"io"
"net/http"
"os"
"path/filepath"
"strconv"
"strings"
+ "sync/atomic"
"testing"
"time"
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)
projects / onechat / commitdiff