Manage source routes with root tools

[onechat] / subagent.go

diff --git a/subagent.go b/subagent.go
index 596c03818510460456b8cbead078bf0051aeee3f..4cb638191eb1fab90544b016c575d074242187b0 100644 (file)
--- a/subagent.go
+++ b/subagent.go
@@ -28,10 +28,11 @@ const (
 // subagents.
 func newSubagents(dir string, clock func() time.Time, draw func() string) *Subagents {
        return &Subagents{
 // 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),
+               dir:      dir,
+               clock:    clock,
+               draw:     draw,
+               open:     make(map[string]*Subagent),
+               reserved: make(map[string]bool),

        }
 }
 
        }
 }
 


@@ -44,8 +45,9 @@ type Subagents struct {
        clock func() time.Time
        draw  func() string
 
        clock func() time.Time
        draw  func() string
 

-       mutex sync.Mutex
-       open  map[string]*Subagent
+       mutex    sync.Mutex
+       open     map[string]*Subagent
+       reserved map[string]bool

 }
 
 // Create makes the directory and the log of a new subagent. It draws a name
 }
 
 // Create makes the directory and the log of a new subagent. It draws a name


@@ -58,6 +60,15 @@ func (s *Subagents) Create(ctx context.Context) (*Subagent, error) {
 
        for range subagentNameDraws {
                name := s.draw()
 
        for range subagentNameDraws {
                name := s.draw()

+               s.mutex.Lock()
+
+               reserved := s.reserved[name]
+               s.mutex.Unlock()
+
+               if reserved {
+                       continue
+               }
+

                err = os.Mkdir(filepath.Join(s.dir, name), 0o700)
                if errors.Is(err, fs.ErrExist) {
                        continue
                err = os.Mkdir(filepath.Join(s.dir, name), 0o700)
                if errors.Is(err, fs.ErrExist) {
                        continue


@@ -73,6 +84,39 @@ func (s *Subagents) Create(ctx context.Context) (*Subagent, error) {
        return nil, fmt.Errorf("no free subagent name after %d draws", subagentNameDraws)
 }
 
        return nil, fmt.Errorf("no free subagent name after %d draws", subagentNameDraws)
 }
 

+// Reserve keeps a child name out of future subagent draws. An existing subagent
+// directory or a reserved name is refused.
+func (s *Subagents) Reserve(name string) error {
+       s.mutex.Lock()
+       defer s.mutex.Unlock()
+
+       if s.reserved[name] {
+               return fmt.Errorf("name %q is already used", name)
+       }
+
+       _, err := os.Lstat(filepath.Join(s.dir, name))
+       if err == nil {
+               return fmt.Errorf("name %q is already used by a subagent", name)
+       }
+
+       if !errors.Is(err, os.ErrNotExist) {
+               return fmt.Errorf("check subagent name: %w", err)
+       }
+
+       s.reserved[name] = true
+
+       return nil
+}
+
+// Release frees a reservation after a child fails to publish. It does not
+// remove a subagent directory or change an open log.
+func (s *Subagents) Release(name string) {
+       s.mutex.Lock()
+       defer s.mutex.Unlock()
+
+       delete(s.reserved, name)
+}
+

 // 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) {
 // 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) {


@@ -124,14 +168,40 @@ func (s *Subagents) Get(ctx context.Context, name string) (*Subagent, error) {
        return subagent, nil
 }
 
        return subagent, nil
 }
 

+// Snapshot flushes and locks the open logs while snapshot reads the subagent
+// files. snapshot must not call Subagents or an open Subagent.
+func (s *Subagents) Snapshot(ctx context.Context, snapshot func() error) error {
+       s.mutex.Lock()
+       defer s.mutex.Unlock()
+
+       for _, subagent := range s.open {
+               subagent.mutex.Lock()
+               defer subagent.mutex.Unlock()
+
+               err := subagent.Flush(ctx)
+               if err != nil {
+                       return fmt.Errorf("flush subagent snapshot: %w", err)
+               }
+       }
+
+       err := snapshot()
+       if err != nil {
+               return fmt.Errorf("copy subagent snapshot: %w", err)
+       }
+
+       return 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
 // 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())
        for name, subagent := range s.open {
                err = errors.Join(err, subagent.close())

+

                delete(s.open, name)
        }
 
                delete(s.open, name)
        }
 


@@ -169,6 +239,14 @@ func (a *Subagent) Flush(ctx context.Context) error {
        return a.syncer.Flush(ctx)
 }
 
        return a.syncer.Flush(ctx)
 }
 

+// First returns the first message of the log.
+func (a *Subagent) First() (Message, error) {
+       a.mutex.Lock()
+       defer a.mutex.Unlock()
+
+       return a.log.Message(0)
+}
+

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