Run subagents in the background with spawn

[onechat] / subagent.go

package main

import (
        "context"
        "errors"
        "fmt"
        "io/fs"
        "os"
        "path/filepath"
        "strconv"
        "strings"
        "sync"
        "time"
)

const (
        // subagentDirName is the directory of the subagents in the directory of
        // an agent.
        subagentDirName = "subagents"

        // subagentNameDraws is the most names that Create draws before it
        // fails.
        subagentNameDraws = 1000
)

// newSubagents returns the subagents in dir, one directory for each name. The
// clock gives the dates of their messages, and draw gives the names of new
// subagents.
func newSubagents(dir string, clock func() time.Time, draw func() string) *Subagents {
        return &Subagents{
                dir:   dir,
                clock: clock,
                draw:  draw,
                open:  make(map[string]*Subagent),
        }
}

// Subagents holds the subagents of one agent. Each subagent has a directory
// named for it, which holds its log. A log takes an exclusive lock, so
// Subagents opens the log of a name once and keeps it open until Close. Its
// mutex protects the open logs.
type Subagents struct {
        dir   string
        clock func() time.Time
        draw  func() string

        mutex sync.Mutex
        open  map[string]*Subagent
}

// Create makes the directory and the log of a new subagent. It draws a name
// again when the directory of the name exists.
func (s *Subagents) Create(ctx context.Context) (*Subagent, error) {
        err := os.MkdirAll(s.dir, 0o700)
        if err != nil {
                return nil, fmt.Errorf("create subagent directory: %w", err)
        }

        for range subagentNameDraws {
                name := s.draw()
                err = os.Mkdir(filepath.Join(s.dir, name), 0o700)
                if errors.Is(err, fs.ErrExist) {
                        continue
                }

                if err != nil {
                        return nil, fmt.Errorf("create subagent directory: %w", err)
                }

                return s.Get(ctx, name)
        }

        return nil, fmt.Errorf("no free subagent name after %d draws", subagentNameDraws)
}

// Get returns the subagent name with its log open. A name that is not valid or
// that has no directory is an unknown agent.
func (s *Subagents) Get(ctx context.Context, name string) (*Subagent, error) {
        s.mutex.Lock()
        defer s.mutex.Unlock()

        subagent, found := s.open[name]
        if found {
                return subagent, nil
        }

        unknown := fmt.Errorf("no agent %s", name)
        if !subagentNameValid(name) {
                return nil, unknown
        }

        root := filepath.Join(s.dir, name)
        info, err := os.Stat(root)
        if errors.Is(err, fs.ErrNotExist) {
                return nil, unknown
        }

        if err != nil {
                return nil, fmt.Errorf("read subagent directory: %w", err)
        }

        if !info.IsDir() {
                return nil, unknown
        }

        log, err := openLog(ctx, root, s.clock)
        if err != nil {
                return nil, fmt.Errorf("open log of %s: %w", name, err)
        }

        syncer, err := openSyncer(root)
        if err != nil {
                return nil, errors.Join(fmt.Errorf("open log of %s: %w", name, err), log.Close())
        }

        subagent = &Subagent{
                name:   name,
                dir:    root,
                log:    log,
                syncer: syncer,
        }
        s.open[name] = subagent

        return subagent, nil
}

// Close makes the writes of each open log durable and closes it.
func (s *Subagents) Close() error {
        s.mutex.Lock()
        defer s.mutex.Unlock()

        var err error
        for name, subagent := range s.open {
                err = errors.Join(err, subagent.close())
                delete(s.open, name)
        }

        return err
}

// Subagent is one subagent, its directory, and its log. Its mutex protects the
// log, because the run of the subagent writes it while the agent reads it with
// zoom.
type Subagent struct {
        name   string
        dir    string
        mutex  sync.Mutex
        log    *Log
        syncer *Syncer
}

// Append logs a message of the subagent.
func (a *Subagent) Append(kind, text string) (Message, error) {
        a.mutex.Lock()
        defer a.mutex.Unlock()

        message, err := a.log.Append(kind, text)
        if err != nil {
                return Message{}, err
        }

        a.syncer.Written()

        return message, nil
}

// Flush returns when every write before the call is durable.
func (a *Subagent) Flush(ctx context.Context) error {
        return a.syncer.Flush(ctx)
}

// Show returns the whole log, each message as "i|kind: text", with a blank line
// between messages.
func (a *Subagent) Show() (string, error) {
        a.mutex.Lock()
        defer a.mutex.Unlock()

        lines := make([]string, 0, a.log.Len())
        for id := range a.log.Len() {
                message, err := a.log.Message(id)
                if err != nil {
                        return "", err
                }

                lines = append(lines, strconv.Itoa(id)+"|"+leafSource(message))
        }

        return strings.Join(lines, "\n\n"), nil
}

// WorkDir returns the directory where the tools of the subagent run: work in
// its directory.
func (a *Subagent) WorkDir() string {
        return filepath.Join(a.dir, "work")
}

// close makes the writes durable and closes the log.
func (a *Subagent) close() error {
        a.mutex.Lock()
        defer a.mutex.Unlock()

        return errors.Join(a.syncer.Close(), a.log.Close())
}