Manage source routes with root tools

[onechat] / daemon_agent.go

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), agent.config.Name, 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,
                openChild: func(ctx context.Context, child *Agent) (*DaemonAgent, error) {
                        return openDaemonAgent(ctx, child, address, auth, compactor, clock, logger.With("name", child.config.Name))
                },
        }, 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
        openChild func(context.Context, *Agent) (*DaemonAgent, error)
}

// 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) {
        loop, err := newTurnLoop(a.agent, a.memory, a.subagents, whatsapp, a.schedules, a.client, a.clock, a.logger, notify)
        if err != nil {
                return nil, err
        }

        loop.openChild = a.openChild

        return loop, nil
}

// 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
}