"io"
"os"
"path/filepath"
+ "slices"
"strings"
)
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) {
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")
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{
// 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,
}
}
return newCompactor(client, compactorAdaPrompt, ModelConfig{
Model: "gpt-test",
Effort: "high",
- })
+ }, promptTools())
}
// compactorInputTexts returns the text of each user message of call. Each
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 {
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 {
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
}
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
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",
// 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()
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() {
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
--- /dev/null
+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
+}
--- /dev/null
+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
+}
--- /dev/null
+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)
+ })
+ }
+}
client.LimitCalls(probeCallLimit)
summarizer := ProbeSlotSummarizer{
- inner: newCompactor(client, probePrompt, config.Compactor),
+ inner: newCompactor(client, probePrompt, config.Compactor, promptTools()),
slots: make(chan struct{}, builds),
}
},
}
- 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))
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))
}
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))
}
// 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
// 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 {
running: make(map[string]*SubagentRun),
}
+ if agent.config.Parent != "" {
+ whatsapp = nil
+ }
+
l.tools = &Toolbox{
log: l,
chat: memory,
Model: l.agent.config.Model,
Effort: l.agent.config.Effort,
Instructions: instructions,
- Tools: promptTools(),
+ Tools: l.agent.Tools(),
},
take: l.take,
}
}
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)
}
projects / onechat / commitdiff