From: Anderson A. Giacomolli Date: Tue, 10 Nov 2026 04:38:37 +0000 (-0300) Subject: Run each child with its own files and loop X-Git-Url: http://git.giacomolli.uk/onechat/commitdiff_plain/2a276c273c21a1752d541bf87e074c5d1c27e30f?ds=sidebyside Run each child with its own files and loop The daemon opens the root and each direct child with separate memory, compaction, usage records, schedules, and subagents. Invalid child configuration stops startup, and the ready event names the children. Each child uses its own cache key and has no WhatsApp tools. Its turn and compaction calls share the same prompt and tools. Backups flush and lock every agent and its subagent logs during the copy, so the archive holds the full state of the children as well as the root. Clients still address only the root. --- diff --git a/agent.go b/agent.go index 790e18b..274806b 100644 --- a/agent.go +++ b/agent.go @@ -10,6 +10,7 @@ import ( "io" "os" "path/filepath" + "slices" "strings" ) @@ -103,6 +104,78 @@ func (a *Agent) Prompt() (string, error) { return prompt(a.config.Name, path, instructions), nil } +// Tools returns the tools shared by the turns and compactions of this agent. A +// child has no WhatsApp tools. +func (a *Agent) Tools() []ModelTool { + tools := promptTools() + if a.config.Parent != "" { + tools = slices.DeleteFunc(tools, func(tool ModelTool) bool { + return strings.HasPrefix(tool.Name, "whatsapp_") + }) + } + + return tools +} + +// Children reads the direct children in directory order. Each child needs an +// existing config with its directory name and this agent as parent. Children +// cannot have children. A missing children directory means no children. +func (a *Agent) Children() ([]*Agent, error) { + dir := filepath.Join(a.dir, "children") + entries, err := os.ReadDir(dir) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + + if err != nil { + return nil, fmt.Errorf("read children: %w", err) + } + + if a.config.Parent != "" && len(entries) != 0 { + return nil, errors.New("children cannot have children") + } + + var children []*Agent + + keys := map[string]bool{a.CacheKey(): true} + + for _, entry := range entries { + name := entry.Name() + if !entry.IsDir() { + return nil, fmt.Errorf("child %q must be a directory", name) + } + + path := filepath.Join(dir, name) + _, err = os.Stat(filepath.Join(path, "config.json")) + if err != nil { + return nil, fmt.Errorf("child %q config: %w", name, err) + } + + child, err := openAgent(path) + if err != nil { + return nil, fmt.Errorf("open child %q: %w", name, err) + } + + if child.config.Name != name || child.config.Parent != a.config.Name { + return nil, fmt.Errorf("child %q needs name %q and parent %q", name, name, a.config.Name) + } + + if child.config.Name == a.config.Name || keys[child.CacheKey()] { + return nil, fmt.Errorf("child %q must have its own name and ID", name) + } + + keys[child.CacheKey()] = true + _, err = child.Children() + if err != nil { + return nil, fmt.Errorf("read child %q: %w", name, err) + } + + children = append(children, child) + } + + return children, nil +} + // SubagentPrompt returns the system prompt of a subagent of the agent, with the // user's instructions as Prompt reads them. func (a *Agent) SubagentPrompt() (string, error) { diff --git a/backup_test.go b/backup_test.go index 440dfe2..6271f8a 100644 --- a/backup_test.go +++ b/backup_test.go @@ -110,6 +110,89 @@ func TestRunningDaemonBackupRestoresTheChatAndLinks(t *testing.T) { require.Equal(t, childText, restoredText) } +func TestBackupRestoresEachChildWithItsOwnState(t *testing.T) { + dir := t.TempDir() + daemonChildConfig(t, dir, "Garden") + daemonChildConfig(t, dir, "Library") + + server := newModelFake(t) + clock := newTestClock() + agents, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, clock.Now, discardLog()) + require.NoError(t, err) + + defer func() { + require.NoError(t, agents.Close()) + }() + + whatsapp, _, _ := whatsappStartTest(t, dir, clock) + + var loops []*TurnLoop + + for _, agent := range agents { + loop, err := agent.Loop(whatsapp, func(Event) { + }) + require.NoError(t, err) + + loops = append(loops, loop) + _, err = loop.Submit("parent", "Keep the decision of "+agent.agent.config.Name) + require.NoError(t, err) + require.NoError(t, loop.turn(t.Context())) + require.NoError(t, agent.memory.Settle(t.Context())) + + subagent, err := agent.subagents.Create(t.Context()) + require.NoError(t, err) + + _, err = subagent.Append("user", "Keep the task of "+agent.agent.config.Name) + require.NoError(t, err) + } + + archive := filepath.Join(t.TempDir(), "children.tar.gz") + _, err = backupData(t.Context(), dir, archive, whatsapp, func(ctx context.Context, snapshot func() error) error { + return daemonSnapshot(ctx, loops, snapshot) + }) + require.NoError(t, err) + + restored := t.TempDir() + require.NoError(t, backupRestore(t.Context(), archive, restored)) + + restoredAgents, err := openDaemonAgents(t.Context(), restored, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, clock.Now, discardLog()) + require.NoError(t, err) + + defer func() { + require.NoError(t, restoredAgents.Close()) + }() + + require.Len(t, restoredAgents, 3) + + for i, agent := range agents { + other := restoredAgents[i] + require.Equal(t, agent.agent.config, other.agent.config) + require.Equal(t, agent.memory.View(), other.memory.View()) + require.Equal(t, agent.memory.Len(), other.memory.Len()) + require.Equal(t, modelReadUsage(t, agent.client.UsagePath()), modelReadUsage(t, other.client.UsagePath())) + + for id := range agent.memory.Len() { + before, err := agent.memory.Message(id) + require.NoError(t, err) + + after, err := other.memory.Message(id) + require.NoError(t, err) + require.Equal(t, before, after) + } + + entries, err := os.ReadDir(filepath.Join(agent.agent.dir, subagentDirName)) + require.NoError(t, err) + require.Len(t, entries, 1) + + subagent, err := other.subagents.Get(t.Context(), entries[0].Name()) + require.NoError(t, err) + + text, err := subagent.Show() + require.NoError(t, err) + require.Contains(t, text, "Keep the task of "+agent.agent.config.Name) + } +} + func TestRestoreRefusesNonemptyDataWithoutChangingIt(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "keep") diff --git a/build_run_test.go b/build_run_test.go index fc8b4b7..b45745e 100644 --- a/build_run_test.go +++ b/build_run_test.go @@ -215,7 +215,7 @@ func newTestBuildRun(t *testing.T, server *ModelFake) *BuildRun { root := t.TempDir() usagePath := filepath.Join(root, "usage.jsonl") client := newModelClient(server.URL, ModelAPIKey(modelTestKey), "onechat-"+probeName, usagePath) - memory, err := openMemory(t.Context(), root, newTestClock().Now, newCompactor(client, probePrompt, defaultConfig().Compactor), discardLog()) + memory, err := openMemory(t.Context(), root, newTestClock().Now, newCompactor(client, probePrompt, defaultConfig().Compactor, promptTools()), discardLog()) require.NoError(t, err) return &BuildRun{ diff --git a/compactor.go b/compactor.go index 0089002..8a16c6e 100644 --- a/compactor.go +++ b/compactor.go @@ -18,13 +18,14 @@ var compactorRuler = strings.Repeat("-", nodeBytes) // newCompactor returns the compactor of an agent, which sends its calls through // client with model. instructions gives the system prompt of the agent at each -// build, so a build sends the same prompt as the turns of that time. -func newCompactor(client *ModelClient, instructions func() (string, error), model ModelConfig) *Compactor { +// build. tools are the tools of that agent, so a build sends the same prompt +// and tools as its turns. +func newCompactor(client *ModelClient, instructions func() (string, error), model ModelConfig, tools []ModelTool) *Compactor { return &Compactor{ client: client, model: model, instructions: instructions, - tools: promptTools(), + tools: tools, } } diff --git a/compactor_test.go b/compactor_test.go index eaed4fe..8540515 100644 --- a/compactor_test.go +++ b/compactor_test.go @@ -208,7 +208,7 @@ func newTestCompactor(t *testing.T, server *ModelFake) *Compactor { return newCompactor(client, compactorAdaPrompt, ModelConfig{ Model: "gpt-test", Effort: "high", - }) + }, promptTools()) } // compactorInputTexts returns the text of each user message of call. Each diff --git a/daemon.go b/daemon.go index e534aea..3a54b1a 100644 --- a/daemon.go +++ b/daemon.go @@ -24,12 +24,13 @@ const ( daemonClientEvents = 1024 ) -// daemon runs "onechat daemon": it owns the root agent and its files and serves -// the clients on the Unix socket in the data directory until ctx ends. When -// telegram.json exists, it also runs the Telegram bot. It also keeps a linked -// WhatsApp device connected. It holds the lock of the token file, so it is the -// only process that refreshes the tokens and opens the WhatsApp session. Its -// log lines go to out, through the loggers of daemonLogger. +// daemon runs "onechat daemon": it owns the root, its children, and their +// files, and serves the clients on the Unix socket in the data directory until +// ctx ends. When telegram.json exists, it also runs the Telegram bot. It also +// keeps a linked WhatsApp device connected. It holds the lock of the token +// file, so it is the only process that refreshes the tokens and opens the +// WhatsApp session. Its log lines go to out, through the loggers of +// daemonLogger. func daemon(ctx context.Context, out io.Writer) (err error) { dir, err := dataDir() if err != nil { @@ -80,30 +81,18 @@ func daemon(ctx context.Context, out io.Writer) (err error) { return err } - agent, err := openAgent(filepath.Join(dir, daemonAgentDirName)) - if err != nil { - return err - } - - client := newModelClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl")) - memory, err := openMemory(ctx, agent.dir, time.Now, newCompactor(client, agent.Prompt, config.Compactor), logger) + agents, err := openDaemonAgents(ctx, dir, address, auth, config.Compactor, time.Now, logger) if err != nil { return err } defer func() { - err = errors.Join(err, memory.Close()) + err = errors.Join(err, agents.Close()) }() - schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), time.Now, logger) - if err != nil { - return err - } - - subagents := newSubagents(filepath.Join(agent.dir, subagentDirName), time.Now, subagentNameRandom) - defer func() { - err = errors.Join(err, subagents.Close()) - }() + root := agents[0] + agent := root.agent + schedules := root.schedules listener, err := daemonListenSocket(socket) if err != nil { @@ -115,7 +104,7 @@ func daemon(ctx context.Context, out io.Writer) (err error) { linked = "linked" } - logger.Info("ready", "version", version, "socket", listener.Addr(), "model", agent.config.Model, "effort", agent.config.Effort, "workdir", agent.config.Workdir, "whatsapp", linked, "schedules", schedules.Len()) + logger.Info("ready", "version", version, "socket", listener.Addr(), "model", agent.config.Model, "effort", agent.config.Effort, "workdir", agent.config.Workdir, "whatsapp", linked, "schedules", schedules.Len(), "children", agents.ChildNames()) var telegram *TelegramBot @@ -126,14 +115,16 @@ func daemon(ctx context.Context, out io.Writer) (err error) { } return daemonServe(ctx, listener, telegram, whatsapp, schedules, func(notify func(Event)) (*TurnLoop, error) { - return newTurnLoop(agent, memory, subagents, whatsapp, schedules, client, time.Now, logger, notify) - }) + return root.Loop(whatsapp, notify) + }, agents[1:]...) } // DaemonHub holds the connected clients and sends each event to all of them. It -// runs the WhatsApp commands of the clients on whatsapp. +// runs the WhatsApp commands of the clients on whatsapp. snapshot locks all +// agents and subagents during a backup copy. type DaemonHub struct { whatsapp *WhatsApp + snapshot func(context.Context, func() error) error mutex sync.Mutex clients map[*DaemonClient]bool @@ -207,7 +198,7 @@ func (h *DaemonHub) read(ctx context.Context, client *DaemonClient, l *TurnLoop) if request.Type == "backup" { var size int64 - size, err = backupData(ctx, filepath.Dir(l.agent.dir), request.Text, h.whatsapp, l.runs.Snapshot) + size, err = backupData(ctx, filepath.Dir(l.agent.dir), request.Text, h.whatsapp, h.snapshot) if err == nil { h.Send(client, Event{ Kind: "backup", @@ -337,8 +328,10 @@ func daemonListenSocket(path string) (net.Listener, error) { // logs its batches and notes in the loop and tells all clients its notices, and // a failure of whatsapp stops the agent. It runs schedules, which log their // runs in the loop, and a failure of schedules stops the agent too. When bot is -// not nil, it runs the bot too, and a failure of the bot stops the agent. -func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, whatsapp *WhatsApp, schedules *Schedules, start func(notify func(Event)) (*TurnLoop, error)) error { +// not nil, it runs the bot too, and a failure of the bot stops the agent. Each +// child runs its own loop and schedules with no client events. A failed child +// stops the daemon, and backups lock all loops before they copy data. +func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, whatsapp *WhatsApp, schedules *Schedules, start func(notify func(Event)) (*TurnLoop, error), children ...*DaemonAgent) error { ctx, cancel := context.WithCancel(ctx) defer cancel() @@ -360,8 +353,39 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w return errors.Join(err, listener.Close()) } + loops := []*TurnLoop{l} + for _, child := range children { + loop, err := child.Loop(nil, func(Event) { + }) + if err != nil { + return errors.Join(fmt.Errorf("start child %q: %w", child.agent.config.Name, err), listener.Close()) + } + + loops = append(loops, loop) + } + + hub.snapshot = func(ctx context.Context, snapshot func() error) error { + return daemonSnapshot(ctx, loops, snapshot) + } + var workers sync.WaitGroup + childErrors := make([]error, len(children)) + for i, child := range children { + workers.Add(1) + + go func() { + defer workers.Done() + + childErrors[i] = daemonRunChild(ctx, loops[i+1], child.schedules) + if childErrors[i] != nil { + child.logger.Error("child stopped", "err", childErrors[i]) + } + + cancel() + }() + } + workers.Add(2) go func() { @@ -418,7 +442,30 @@ func daemonServe(ctx context.Context, listener net.Listener, bot *TelegramBot, w hub.CloseAll() workers.Wait() - return errors.Join(err, whatsappErr, schedulesErr, botErr) + return errors.Join(err, whatsappErr, schedulesErr, botErr, errors.Join(childErrors...)) +} + +// daemonRunChild runs a child's turn loop and its schedules until either stops. +// Its events stay in its own log. +func daemonRunChild(ctx context.Context, loop *TurnLoop, schedules *Schedules) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + + done := make(chan error, 1) + go func() { + done <- schedules.Run(ctx, loop) + + cancel() + }() + + err := loop.Run(ctx) + if errors.Is(err, context.Canceled) { + err = nil + } + + cancel() + + return errors.Join(err, <-done) } // daemonRunRequest runs one request of a client. The answers of a WhatsApp diff --git a/daemon_agent.go b/daemon_agent.go new file mode 100644 index 0000000..1161c8a --- /dev/null +++ b/daemon_agent.go @@ -0,0 +1,146 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log/slog" + "path/filepath" + "time" +) + +// openDaemonAgents opens the root and its direct children. An invalid child +// stops startup, and each opened agent is closed when startup fails. +func openDaemonAgents(ctx context.Context, dir, address string, auth ModelAuthorizer, compactor ModelConfig, clock func() time.Time, logger *slog.Logger) (agents DaemonAgents, err error) { + root, err := openAgent(filepath.Join(dir, daemonAgentDirName)) + if err != nil { + return nil, fmt.Errorf("open root agent: %w", err) + } + + if root.config.Parent != "" { + return nil, errors.New("root agent cannot have a parent") + } + + children, err := root.Children() + if err != nil { + return nil, fmt.Errorf("open child agents: %w", err) + } + + defer func() { + if err != nil { + err = errors.Join(err, agents.Close()) + } + }() + + for _, agent := range append([]*Agent{root}, children...) { + part := logger + if agent.config.Parent != "" { + part = logger.With("name", agent.config.Name) + } + + var opened *DaemonAgent + + opened, err = openDaemonAgent(ctx, agent, address, auth, compactor, clock, part) + if err != nil { + return agents, fmt.Errorf("open agent %q: %w", agent.config.Name, err) + } + + agents = append(agents, opened) + } + + return agents, nil +} + +// openDaemonAgent opens the memory, model client, schedules, and subagents of +// one agent. All model calls use its cache key and usage file. +func openDaemonAgent(ctx context.Context, agent *Agent, address string, auth ModelAuthorizer, compactor ModelConfig, clock func() time.Time, logger *slog.Logger) (*DaemonAgent, error) { + client := newModelClient(address, auth, agent.CacheKey(), filepath.Join(agent.dir, "usage.jsonl")) + memory, err := openMemory(ctx, agent.dir, clock, newCompactor(client, agent.Prompt, compactor, agent.Tools()), logger) + if err != nil { + return nil, fmt.Errorf("open memory: %w", err) + } + + schedules, err := openSchedules(filepath.Join(agent.dir, scheduleFileName), clock, logger) + if err != nil { + return nil, errors.Join(fmt.Errorf("open schedules: %w", err), memory.Close()) + } + + return &DaemonAgent{ + agent: agent, + memory: memory, + client: client, + subagents: newSubagents(filepath.Join(agent.dir, subagentDirName), clock, subagentNameRandom), + schedules: schedules, + clock: clock, + logger: logger, + }, nil +} + +// DaemonAgent owns the open files and model client of one persistent agent. Its +// loop must stop before Close releases the files. +type DaemonAgent struct { + agent *Agent + memory *Memory + client *ModelClient + subagents *Subagents + schedules *Schedules + clock func() time.Time + logger *slog.Logger +} + +// Loop makes the turn loop with its event destination. Only the root has a +// WhatsApp link. Child events stay with their own logs. +func (a *DaemonAgent) Loop(whatsapp *WhatsApp, notify func(Event)) (*TurnLoop, error) { + return newTurnLoop(a.agent, a.memory, a.subagents, whatsapp, a.schedules, a.client, a.clock, a.logger, notify) +} + +// Close flushes and closes the subagents and the memory. +func (a *DaemonAgent) Close() error { + return errors.Join(a.subagents.Close(), a.memory.Close()) +} + +// DaemonAgents holds the root first and then its children in directory order. +type DaemonAgents []*DaemonAgent + +// Close closes all agents, including after a partial startup. +func (a DaemonAgents) Close() error { + var err error + + for _, agent := range a { + err = errors.Join(err, agent.Close()) + } + + return err +} + +// ChildNames returns the names for the daemon's ready line. +func (a DaemonAgents) ChildNames() []string { + names := make([]string, 0, len(a)-1) + for _, child := range a[1:] { + names = append(names, child.agent.config.Name) + } + + return names +} + +// daemonSnapshot flushes and locks all agent and subagent logs before copy. +// Loops must stay in the same order for every snapshot. +func daemonSnapshot(ctx context.Context, loops []*TurnLoop, snapshot func() error) error { + if len(loops) == 0 { + err := snapshot() + if err != nil { + return fmt.Errorf("copy agent files: %w", err) + } + + return nil + } + + err := loops[0].runs.Snapshot(ctx, func() error { + return daemonSnapshot(ctx, loops[1:], snapshot) + }) + if err != nil { + return fmt.Errorf("snapshot agent %q: %w", loops[0].agent.config.Name, err) + } + + return nil +} diff --git a/daemon_agent_helper_test.go b/daemon_agent_helper_test.go new file mode 100644 index 0000000..8c87f83 --- /dev/null +++ b/daemon_agent_helper_test.go @@ -0,0 +1,32 @@ +package main + +import ( + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/require" +) + +// daemonChildConfig writes a complete child config in its own directory. +func daemonChildConfig(t *testing.T, dir, name string) string { + t.Helper() + + root, err := openAgent(filepath.Join(dir, daemonAgentDirName)) + require.NoError(t, err) + + path := filepath.Join(root.dir, "children", name) + child, err := openAgent(path) + require.NoError(t, err) + + config := child.config + config.Name = name + config.Model = "gpt-test-" + strings.ToLower(name) + config.Effort = "high" + config.Parent = root.config.Name + config.Responsibility = "Keep the plans and decisions for " + name + "." + config.Workdir = dir + agentWriteTestConfig(t, path, config) + + return path +} diff --git a/daemon_agent_test.go b/daemon_agent_test.go new file mode 100644 index 0000000..7b1407b --- /dev/null +++ b/daemon_agent_test.go @@ -0,0 +1,250 @@ +package main + +import ( + "context" + "io" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestDaemonOpensAndRunsTheRootAndTwoChildren(t *testing.T) { + dir := t.TempDir() + daemonChildConfig(t, dir, "Garden") + daemonChildConfig(t, dir, "Library") + + server := newModelFake(t) + clock := newTestClock() + agents, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, clock.Now, discardLog()) + require.NoError(t, err) + + t.Cleanup(func() { + require.NoError(t, agents.Close()) + }) + require.Len(t, agents, 3) + require.Equal(t, []string{"Garden", "Library"}, agents.ChildNames()) + + for _, agent := range agents { + at := clock.Now().Add(time.Minute).Format(scheduleAtLayout) + _, err = agent.schedules.Set("Record the decision for "+agent.agent.config.Name, &at, nil) + require.NoError(t, err) + } + + clock.now = clock.now.Add(time.Minute) + whatsapp, _, _ := whatsappStartTest(t, dir, clock) + listener, err := daemonListenSocket(filepath.Join(dir, socketFileName)) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + done := make(chan error, 1) + go func() { + done <- daemonServe(ctx, listener, nil, whatsapp, agents[0].schedules, func(notify func(Event)) (*TurnLoop, error) { + return agents[0].Loop(whatsapp, notify) + }, agents[1:]...) + }() + + t.Cleanup(func() { + cancel() + require.NoError(t, <-done) + }) + + require.Eventually(t, func() bool { + for _, agent := range agents { + if agent.memory.Len() != 2 { + return false + } + } + + return true + }, 5*time.Second, time.Millisecond) + + require.Equal(t, 3, server.Calls()) + + keys := make(map[string]bool) + for i := range server.Calls() { + call := server.Call(i) + keys[call.body["prompt_cache_key"].(string)] = true + } + + require.Len(t, keys, 3) + + for _, agent := range agents { + message, err := agent.memory.Message(0) + require.NoError(t, err) + require.Contains(t, message.Text, "Record the decision for "+agent.agent.config.Name) + require.True(t, keys[agent.agent.CacheKey()]) + + usage := modelReadUsage(t, filepath.Join(agent.agent.dir, "usage.jsonl")) + require.Len(t, usage, 1) + require.Equal(t, "first", usage[0].Turn) + require.Equal(t, agent.agent.config.Model, usage[0].Model) + } + + terminal := terminalStartTestClient(t, listener.Addr().String()) + terminal.WaitFor(t, "[connected to OneChat]") + require.NotContains(t, terminal.Output(), "Garden") + require.NotContains(t, terminal.Output(), "Library") +} + +func TestDaemonReadyLineNamesTheChildren(t *testing.T) { + dir := t.TempDir() + daemonChildConfig(t, dir, "Garden") + daemonChildConfig(t, dir, "Library") + t.Setenv("ONECHAT_DIR", dir) + + var out SyncBuffer + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + done := make(chan error, 1) + go func() { + done <- daemon(ctx, &out) + }() + + t.Cleanup(func() { + cancel() + + require.NoError(t, <-done) + }) + + require.Eventually(t, func() bool { + return strings.Contains(out.String(), "children=[Garden Library]") + }, 5*time.Second, time.Millisecond) +} + +func TestChildCallsShareTheirPromptAndToolsWithCompaction(t *testing.T) { + dir := t.TempDir() + daemonChildConfig(t, dir, "Garden") + + server := newModelFake(t) + agents, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, time.Now, discardLog()) + require.NoError(t, err) + + defer func() { + require.NoError(t, agents.Close()) + }() + + child := agents[1] + loop, err := child.Loop(nil, func(Event) { + }) + require.NoError(t, err) + require.Nil(t, loop.tools.whatsapp) + require.Contains(t, loop.tools.run(t.Context(), "whatsapp_send", `{"id":"fake","chat":"fake","text":"draft"}`), "unknown tool") + + _, err = loop.Submit("parent", strings.Repeat("Keep a garden decision. ", nodeBytes)) + require.NoError(t, err) + require.NoError(t, loop.turn(t.Context())) + require.NoError(t, child.memory.Settle(t.Context())) + + require.GreaterOrEqual(t, server.Calls(), 2) + + first := server.Call(0) + + for i := range server.Calls() { + call := server.Call(i) + require.Equal(t, child.agent.CacheKey(), call.body["prompt_cache_key"]) + require.Equal(t, first.body["instructions"], call.body["instructions"]) + require.Equal(t, first.body["tools"], call.body["tools"]) + require.NotContains(t, string(call.raw), "whatsapp_send") + require.NotContains(t, string(call.raw), "whatsapp_find") + } + + usage := modelReadUsage(t, child.client.UsagePath()) + require.GreaterOrEqual(t, len(usage), 2) + + var turns, nodes int + + for _, record := range usage { + if record.Node != nil { + nodes++ + } else { + turns++ + } + } + + require.Positive(t, nodes) + require.Equal(t, 1, turns) +} + +func TestBrokenChildConfigStopsDaemonStartup(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, daemonAgentDirName, "children", "Garden") + require.NoError(t, os.MkdirAll(path, 0o700)) + require.NoError(t, os.WriteFile(filepath.Join(path, "config.json"), []byte(`{"broken":true}`), 0o600)) + t.Setenv("ONECHAT_DIR", dir) + + err := daemon(t.Context(), io.Discard) + require.ErrorContains(t, err, `open child "Garden"`) + require.ErrorContains(t, err, `unknown field "broken"`) + + _, err = os.Stat(filepath.Join(dir, socketFileName)) + require.ErrorIs(t, err, os.ErrNotExist) +} + +func TestInvalidChildRelationshipsStopStartup(t *testing.T) { + server := newModelFake(t) + testCases := []struct { + name string + change func(*testing.T, string) + cause string + }{ + { + name: "missing config is not created", + change: func(t *testing.T, path string) { + require.NoError(t, os.Remove(filepath.Join(path, "config.json"))) + }, + cause: `child "Garden" config:`, + }, + { + name: "wrong parent is rejected", + change: func(t *testing.T, path string) { + agent, err := openAgent(path) + require.NoError(t, err) + + config := agent.config + config.Parent = "Other" + agentWriteTestConfig(t, path, config) + }, + cause: `needs name "Garden" and parent "OneChat"`, + }, + { + name: "a grandchild is rejected", + change: func(t *testing.T, path string) { + require.NoError(t, os.MkdirAll(filepath.Join(path, "children", "Nested"), 0o700)) + }, + cause: "children cannot have children", + }, + { + name: "a reused ID is rejected", + change: func(t *testing.T, path string) { + root, err := openAgent(filepath.Dir(filepath.Dir(path))) + require.NoError(t, err) + + child, err := openAgent(path) + require.NoError(t, err) + + config := child.config + config.ID = root.config.ID + agentWriteTestConfig(t, path, config) + }, + cause: "must have its own name and ID", + }, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + path := daemonChildConfig(t, dir, "Garden") + tc.change(t, path) + + _, err := openDaemonAgents(t.Context(), dir, server.URL, ModelAPIKey(modelTestKey), defaultConfig().Compactor, time.Now, discardLog()) + require.ErrorContains(t, err, tc.cause) + }) + } +} diff --git a/probe.go b/probe.go index 3781cd3..b004829 100644 --- a/probe.go +++ b/probe.go @@ -103,7 +103,7 @@ func probe(ctx context.Context, args []string, out io.Writer) (err error) { client.LimitCalls(probeCallLimit) summarizer := ProbeSlotSummarizer{ - inner: newCompactor(client, probePrompt, config.Compactor), + inner: newCompactor(client, probePrompt, config.Compactor, promptTools()), slots: make(chan struct{}, builds), } diff --git a/rebuild_test.go b/rebuild_test.go index 8d3c9d4..fc4ff7e 100644 --- a/rebuild_test.go +++ b/rebuild_test.go @@ -63,7 +63,7 @@ func TestRebuildMergesDatesAndKeepsAgentFiles(t *testing.T) { }, } - backup, err := rebuild(t.Context(), agent, added, client, newCompactor(client, agent.Prompt, defaultConfig().Compactor), &report) + backup, err := rebuild(t.Context(), agent, added, client, newCompactor(client, agent.Prompt, defaultConfig().Compactor, agent.Tools()), &report) require.NoError(t, err) require.Equal(t, before, rebuildTestSnapshot(t, backup)) @@ -143,7 +143,7 @@ func TestRebuildFailureAndCancellationKeepOriginal(t *testing.T) { Kind: "work", Text: strings.Repeat("long ", 200), Date: "2025-01-01T00:00:00Z", - }}, client, newCompactor(client, agent.Prompt, defaultConfig().Compactor), &strings.Builder{}) + }}, client, newCompactor(client, agent.Prompt, defaultConfig().Compactor, agent.Tools()), &strings.Builder{}) require.Error(t, err) require.Empty(t, backup) require.Equal(t, before, rebuildTestSnapshot(t, agent.dir)) @@ -179,7 +179,7 @@ func TestRebuildKeepsOriginalWhenAnInvisibleMergeFails(t *testing.T) { } client := newModelClient(server.URL, ModelAPIKey(modelTestKey), agent.CacheKey(), filepath.Join(t.TempDir(), "usage.jsonl")) - _, err := rebuild(t.Context(), agent, nil, client, newCompactor(client, agent.Prompt, defaultConfig().Compactor), &strings.Builder{}) + _, err := rebuild(t.Context(), agent, nil, client, newCompactor(client, agent.Prompt, defaultConfig().Compactor, agent.Tools()), &strings.Builder{}) require.ErrorContains(t, err, "summary tree has unfinished nodes") require.Equal(t, before, rebuildTestSnapshot(t, agent.dir)) } diff --git a/tools.go b/tools.go index 4c00350..0811981 100644 --- a/tools.go +++ b/tools.go @@ -52,8 +52,8 @@ type ToolChat interface { // directory, with no sandbox. Zoom with a name reads the log of a subagent. The // toolbox of the agent has the runs of its subagents for spawn and tell, the // WhatsApp link for the WhatsApp tools, and the schedules for the schedule -// tools. The toolbox of a subagent has none of them, so these are unknown tools -// there. +// tools. A child has no WhatsApp link. The toolbox of a subagent has none of +// them, so these are unknown tools there. type Toolbox struct { log ToolLog chat ToolChat diff --git a/turn.go b/turn.go index 637d783..2e5e50e 100644 --- a/turn.go +++ b/turn.go @@ -13,7 +13,8 @@ import ( // newTurnLoop returns the loop of agent with its memory, its subagents, and the // WhatsApp link and the schedules of its tools. Its calls go through client, // and notify gets each event. The clock gives the date of a turn. logger gets -// the failed turns and the ends of the subagents. +// the failed turns and the ends of the subagents. A child has no WhatsApp +// tools, even when the caller supplies a link. func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *WhatsApp, schedules *Schedules, client ChainClient, clock func() time.Time, logger *slog.Logger, notify func(Event)) (*TurnLoop, error) { layout, err := loadViewLayout(filepath.Join(agent.dir, viewLayoutFileName)) if err != nil { @@ -46,6 +47,10 @@ func newTurnLoop(agent *Agent, memory *Memory, subagents *Subagents, whatsapp *W running: make(map[string]*SubagentRun), } + if agent.config.Parent != "" { + whatsapp = nil + } + l.tools = &Toolbox{ log: l, chat: memory, @@ -386,7 +391,7 @@ func (l *TurnLoop) chain(ctx context.Context) error { Model: l.agent.config.Model, Effort: l.agent.config.Effort, Instructions: instructions, - Tools: promptTools(), + Tools: l.agent.Tools(), }, take: l.take, } diff --git a/whatsapp_import.go b/whatsapp_import.go index 7071c83..e1cbd02 100644 --- a/whatsapp_import.go +++ b/whatsapp_import.go @@ -142,7 +142,7 @@ func whatsappImport(ctx context.Context, args []string, out io.Writer) (err erro } client := newModelClient(address, auth, agent.CacheKey(), usagePath) - backup, err := rebuild(ctx, agent, plan.added, client, newCompactor(client, agent.Prompt, config.Compactor), out) + backup, err := rebuild(ctx, agent, plan.added, client, newCompactor(client, agent.Prompt, config.Compactor, agent.Tools()), out) if err != nil { return fmt.Errorf("rebuild imported memory: %w", err) }