From 29f7b45adce19fb19c4d83588d969239c171eb5b Mon Sep 17 00:00:00 2001 From: Levi Neely Date: Fri, 6 Mar 2026 19:35:41 +0100 Subject: [PATCH] session: wake mail processor immediately on inbox delivery or idle transition Add wakeupCh (buffered 1) to session.Manager. The mail processing loop now selects on both the 5s ticker and the wakeup channel. Two events signal the wakeup: a message delivered to any bot inbox (via onRecv callback), and a session transitioning to idle with pending messages (via OnStateChange). When woken, processMailboxes bypasses the idleDuration threshold check. This covers all delivery paths including direct DeliverToInbox calls from the supervisor. Also flatten p9/roles to match p9/skills: remove focus-area directory layer, expose roles as a flat .md listing. Co-Authored-By: Claude Sonnet 4.6 --- internal/p9/roles.go | 216 +++++++++--------------------------- internal/p9/server.go | 47 +++----- internal/p9/tools.go | 2 +- internal/session/manager.go | 40 +++++-- 4 files changed, 102 insertions(+), 203 deletions(-) diff --git a/internal/p9/roles.go b/internal/p9/roles.go index 68c4f7a..dcaada2 100644 --- a/internal/p9/roles.go +++ b/internal/p9/roles.go @@ -5,23 +5,22 @@ import ( "fmt" "os" "path/filepath" - "slices" + "sort" "strings" "9fans.net/go/plan9" ) -// RolesFS provides 9P filesystem access to agent roles organized by focus areas. -// Roles are markdown files with YAML front-matter defining specialized agent personas. +// RolesFS provides a flat virtual filesystem for roles. +// Roles are discovered from configured directories and exposed as .md files. +// The listing is the union of all roles directories, sorted by name. type RolesFS struct { rolesDirs []string } // RoleMeta holds parsed YAML front-matter from role markdown files. type RoleMeta struct { - Name string // Physical filename (without .md) - DisplayName string // Name from frontmatter (for discovery) - FocusAreas []string + Name string Description string Path string // file path } @@ -51,8 +50,8 @@ func NewRolesFS() *RolesFS { return &RolesFS{rolesDirs: dirs} } -// parseRoleFrontMatter extracts focus-areas and description from role markdown YAML front-matter -func parseRoleFrontMatter(rolePath string, derivedFocusAreas []string) (*RoleMeta, error) { +// parseRoleFrontMatter extracts description from role markdown YAML front-matter. +func parseRoleFrontMatter(rolePath string) (*RoleMeta, error) { f, err := os.Open(rolePath) if err != nil { return nil, err @@ -60,14 +59,12 @@ func parseRoleFrontMatter(rolePath string, derivedFocusAreas []string) (*RoleMet defer f.Close() meta := &RoleMeta{ - Name: strings.TrimSuffix(filepath.Base(rolePath), ".md"), - Path: rolePath, - FocusAreas: derivedFocusAreas, + Name: strings.TrimSuffix(filepath.Base(rolePath), ".md"), + Path: rolePath, } scanner := bufio.NewScanner(f) inFrontMatter := false - hasFrontMatterFocusAreas := false hasDescription := false for scanner.Scan() { @@ -86,19 +83,9 @@ func parseRoleFrontMatter(rolePath string, derivedFocusAreas []string) (*RoleMet continue } - if focusAreas, ok := strings.CutPrefix(line, "focus-areas:"); ok { - hasFrontMatterFocusAreas = true - for fa := range strings.SplitSeq(focusAreas, ",") { - fa = strings.TrimSpace(fa) - if fa != "" && !slices.Contains(meta.FocusAreas, fa) { - meta.FocusAreas = append(meta.FocusAreas, fa) - } - } - } else if desc, ok := strings.CutPrefix(line, "description:"); ok { + if desc, ok := strings.CutPrefix(line, "description:"); ok { meta.Description = strings.TrimSpace(desc) hasDescription = true - } else if name, ok := strings.CutPrefix(line, "name:"); ok { - meta.DisplayName = strings.TrimSpace(name) } } @@ -107,146 +94,52 @@ func parseRoleFrontMatter(rolePath string, derivedFocusAreas []string) (*RoleMet return nil, fmt.Errorf("missing description") } - // If no frontmatter focus-areas and no derived, add "uncategorized" - if !hasFrontMatterFocusAreas && len(derivedFocusAreas) == 0 { - meta.FocusAreas = []string{"uncategorized"} - } - return meta, nil } -// scanRolesDir recursively scans a directory for role markdown files -func (r *RolesFS) scanRolesDir(baseDir string, currentPath string, derivedFocusAreas []string) ([]*RoleMeta, error) { +// listAllRoles scans all roles directories and returns deduplicated metadata, +// with the first occurrence (by rolesDirs order) winning on name collision. +func (r *RolesFS) listAllRoles() ([]*RoleMeta, error) { + seen := make(map[string]bool) var roles []*RoleMeta - entries, err := os.ReadDir(currentPath) - if err != nil { - return nil, err - } - - for _, entry := range entries { - fullPath := filepath.Join(currentPath, entry.Name()) - - if entry.IsDir() { - // Derive focus area from directory structure - newDerived := append([]string{}, derivedFocusAreas...) - newDerived = append(newDerived, entry.Name()) - subRoles, err := r.scanRolesDir(baseDir, fullPath, newDerived) - if err != nil { - continue - } - roles = append(roles, subRoles...) - } else if strings.HasSuffix(entry.Name(), ".md") { - meta, err := parseRoleFrontMatter(fullPath, derivedFocusAreas) - if err != nil { - continue - } - roles = append(roles, meta) - } - } - - return roles, nil -} - -// listAllRoles scans all roles directories and returns metadata -func (r *RolesFS) listAllRoles() ([]*RoleMeta, error) { - seen := make(map[string]*RoleMeta) - for _, dir := range r.rolesDirs { - roles, err := r.scanRolesDir(dir, dir, nil) + entries, err := os.ReadDir(dir) if err != nil { continue } - for _, role := range roles { - // Use physical filename (role.Name) as unique key - // Merge: keep first occurrence, union focus areas - if existing, exists := seen[role.Name]; exists { - for _, fa := range role.FocusAreas { - if !slices.Contains(existing.FocusAreas, fa) { - existing.FocusAreas = append(existing.FocusAreas, fa) - } - } - } else { - seen[role.Name] = role + for _, entry := range entries { + if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".md") { + continue } + name := strings.TrimSuffix(entry.Name(), ".md") + if seen[name] { + continue + } + rolePath := filepath.Join(dir, entry.Name()) + meta, err := parseRoleFrontMatter(rolePath) + if err != nil { + continue + } + seen[name] = true + roles = append(roles, meta) } } - var result []*RoleMeta - for _, role := range seen { - result = append(result, role) - } - return result, nil + sort.Slice(roles, func(i, j int) bool { return roles[i].Name < roles[j].Name }) + return roles, nil } -// listFocusAreas returns unique focus area names from all roles -func (r *RolesFS) listFocusAreas() ([]string, error) { - roles, err := r.listAllRoles() - if err != nil { - return nil, err - } - - focusSet := make(map[string]bool) - for _, role := range roles { - for _, fa := range role.FocusAreas { - focusSet[fa] = true - } - } - - var focusAreas []string - for fa := range focusSet { - focusAreas = append(focusAreas, fa) - } - return focusAreas, nil -} - -// listRolesInFocusArea returns roles that have the given focus area -func (r *RolesFS) listRolesInFocusArea(focusArea string) ([]*RoleMeta, error) { - roles, err := r.listAllRoles() - if err != nil { - return nil, err - } - - var result []*RoleMeta - for _, role := range roles { - if slices.Contains(role.FocusAreas, focusArea) { - result = append(result, role) - } - } - return result, nil -} - -// List returns directory entries for a roles path +// List returns directory entries for the flat roles root: agent/roles → .md files. func (r *RolesFS) List(path string) ([]plan9.Dir, error) { parts := strings.Split(strings.TrimPrefix(path, "/"), "/") - // /agent/roles - list focus areas if len(parts) == 2 && parts[0] == "agent" && parts[1] == "roles" { - focusAreas, err := r.listFocusAreas() + roles, err := r.listAllRoles() if err != nil { return nil, err } - - var dirs []plan9.Dir - for _, fa := range focusAreas { - dirs = append(dirs, plan9.Dir{ - Name: fa, - Qid: plan9.Qid{Type: plan9.QTDIR}, - Mode: plan9.DMDIR | 0555, - }) - } - return dirs, nil - } - - // /agent/roles/ - list role files - if len(parts) == 3 && parts[0] == "agent" && parts[1] == "roles" { - focusArea := parts[2] - roles, err := r.listRolesInFocusArea(focusArea) - if err != nil { - return nil, err - } - var dirs []plan9.Dir for _, role := range roles { dirs = append(dirs, plan9.Dir{ @@ -261,37 +154,34 @@ func (r *RolesFS) List(path string) ([]plan9.Dir, error) { return nil, fmt.Errorf("not found") } -// Read returns file content for a roles path +// Read returns file content for agent/roles/.md. func (r *RolesFS) Read(path string) ([]byte, error) { parts := strings.Split(strings.TrimPrefix(path, "/"), "/") - // /agent/roles//.md - if len(parts) != 4 || parts[0] != "agent" || parts[1] != "roles" { - return nil, fmt.Errorf("not found") - } - - focusArea := parts[2] - fileName := parts[3] - - if !strings.HasSuffix(fileName, ".md") { - return nil, fmt.Errorf("not a markdown file") - } - - roles, err := r.listRolesInFocusArea(focusArea) - if err != nil { - return nil, err - } - - for _, role := range roles { - if role.Name == strings.TrimSuffix(fileName, ".md") { - return os.ReadFile(role.Path) + // agent/roles/.md + if len(parts) == 3 && parts[0] == "agent" && parts[1] == "roles" { + fileName := parts[2] + if !strings.HasSuffix(fileName, ".md") { + return nil, fmt.Errorf("not found") } + roleName := strings.TrimSuffix(fileName, ".md") + + roles, err := r.listAllRoles() + if err != nil { + return nil, err + } + for _, role := range roles { + if role.Name == roleName { + return os.ReadFile(role.Path) + } + } + return nil, fmt.Errorf("role not found: %s", roleName) } - return nil, fmt.Errorf("role not found") + return nil, fmt.Errorf("not found") } -// ReadRole reads a role by name, searching all focus areas +// ReadRole reads a role by name, searching all roles directories. func (r *RolesFS) ReadRole(roleName string) (string, error) { roles, err := r.listAllRoles() if err != nil { diff --git a/internal/p9/server.go b/internal/p9/server.go index badc678..92cbb8b 100644 --- a/internal/p9/server.go +++ b/internal/p9/server.go @@ -17,7 +17,6 @@ import ( "os" "path/filepath" "regexp" - "slices" "sort" "strconv" "strings" @@ -93,9 +92,9 @@ const ( qidCompletedBase = 0x40000000 // session/{id}/completed qidMessageBase = 0x50000000 // message files qidBeadsBase = 0x60000000 // beads/{id}/file - qidToolsBase = 0x70000000 // tools/{capability}/{tool} - qidSkillsBase = 0x80000000 // skills/{intent}/{skill} - qidRolesBase = 0x90000000 // roles/{focus-area}/{role} + qidToolsBase = 0x70000000 // tools/{tool} + qidSkillsBase = 0x80000000 // skills/{skill} + qidRolesBase = 0x90000000 // roles/{role} ) // File indices within a session directory @@ -519,22 +518,24 @@ func (s *Server) walk(cs *connState, fc *plan9.Fcall) *plan9.Fcall { return errFcall(fc, "not found") } } else if path == "/roles" { - // Inside roles directory - list focus areas + // Flat role files: .md if s.roles != nil { - focusAreas, _ := s.roles.listFocusAreas() - if slices.Contains(focusAreas, name) { - qid = plan9.Qid{Type: QTDir, Path: qidRolesBase + hashID(name)} - newPath = "/roles/" + name - } else { + roles, _ := s.roles.listAllRoles() + found := false + for _, role := range roles { + if name == role.Name+".md" { + qid = plan9.Qid{Type: QTFile, Path: qidRolesBase + hashID(name)} + newPath = "/roles/" + name + found = true + break + } + } + if !found { return errFcall(fc, "not found") } } else { return errFcall(fc, "not found") } - } else if strings.HasPrefix(path, "/roles/") && strings.Count(path, "/") == 2 { - // Inside roles/ - role files - qid = plan9.Qid{Type: QTFile, Path: qidRolesBase + hashID(path+name)} - newPath = path + "/" + name } else if strings.HasPrefix(path, "/beads/") && strings.Count(path, "/") == 2 { // Inside a mount directory mountName := strings.TrimPrefix(path, "/beads/") @@ -1134,23 +1135,11 @@ func (s *Server) readDir(path string, offset uint64, count uint32) []byte { } } else if path == "/roles" { if s.roles != nil { - // Add focus area directories - focusAreas, _ := s.roles.listFocusAreas() - for _, fa := range focusAreas { - dirs = append(dirs, plan9.Dir{ - Qid: plan9.Qid{Type: QTDir, Path: qidRolesBase + hashID(fa)}, - Mode: plan9.DMDIR | 0555, Name: fa, Uid: "q", Gid: "q", Muid: "q", - }) - } - } - } else if strings.HasPrefix(path, "/roles/") && strings.Count(path, "/") == 2 { - // Inside roles/ - list role files - focusArea := strings.TrimPrefix(path, "/roles/") - if s.roles != nil { - roles, _ := s.roles.listRolesInFocusArea(focusArea) + // Flat: list .md files + roles, _ := s.roles.listAllRoles() for _, role := range roles { dirs = append(dirs, plan9.Dir{ - Qid: plan9.Qid{Type: QTFile, Path: qidRolesBase + hashID(path+role.Name)}, + Qid: plan9.Qid{Type: QTFile, Path: qidRolesBase + hashID(role.Name + ".md")}, Mode: 0444, Name: role.Name + ".md", Uid: "q", Gid: "q", Muid: "q", }) } diff --git a/internal/p9/tools.go b/internal/p9/tools.go index 45a5e78..c623950 100644 --- a/internal/p9/tools.go +++ b/internal/p9/tools.go @@ -10,7 +10,7 @@ import ( "9fans.net/go/plan9" ) -// ToolsFS provides 9P filesystem access to agent tools organized by capability level. +// ToolsFS provides a flat virtual filesystem for agent tools. // Tools are shell scripts with YAML front-matter defining their interface. type ToolsFS struct { tools []Tool diff --git a/internal/session/manager.go b/internal/session/manager.go index 9f6ec57..c6453b4 100644 --- a/internal/session/manager.go +++ b/internal/session/manager.go @@ -24,6 +24,7 @@ type Manager struct { OnStateChange func(sessionID, oldState, newState string) mu sync.RWMutex stopCh chan struct{} + wakeupCh chan struct{} wg sync.WaitGroup } @@ -35,8 +36,9 @@ func NewManager(backends map[string]backend.Backend) *Manager { backends: backends, sessions: make(map[string]backend.Session), mailManager: mailMgr, - eventBus: nil, // Set via SetEventBus + eventBus: nil, // Set via SetEventBus stopCh: make(chan struct{}), + wakeupCh: make(chan struct{}, 1), } // Set session getter for alias lookup @@ -50,6 +52,9 @@ func NewManager(backends map[string]backend.Backend) *Manager { "new_state": newState, }) } + if newState == "idle" && m.mailManager.HasPendingMessages(sessionID) { + m.signalWakeup() + } } // Wire up mailbox event callbacks @@ -71,6 +76,9 @@ func NewManager(backends map[string]backend.Backend) *Manager { } m.eventBus.Publish(receiverID, evType, msg) } + if receiverID != "user" { + m.signalWakeup() + } }, ) @@ -180,13 +188,21 @@ func (m *Manager) Remove(id string) { } } +// signalWakeup sends a non-blocking wakeup to the mail processing loop. +func (m *Manager) signalWakeup() { + select { + case m.wakeupCh <- struct{}{}: + default: + } +} + // Stop stops the mail processing loop func (m *Manager) Stop() { close(m.stopCh) m.wg.Wait() } -// mailProcessingLoop processes mailboxes every 5 seconds +// mailProcessingLoop processes mailboxes every 5 seconds, or immediately when woken. func (m *Manager) mailProcessingLoop() { defer m.wg.Done() defer func() { @@ -194,22 +210,25 @@ func (m *Manager) mailProcessingLoop() { logging.Logger().Error("panic in mailProcessingLoop", zap.Any("panic", r)) } }() - + ticker := time.NewTicker(5 * time.Second) defer ticker.Stop() - + for { select { case <-m.stopCh: return + case <-m.wakeupCh: + m.processMailboxes(true) case <-ticker.C: - m.processMailboxes() + m.processMailboxes(false) } } } -// processMailboxes handles outbox delivery and inbox processing -func (m *Manager) processMailboxes() { +// processMailboxes handles outbox delivery and inbox processing. +// bypassIdleCheck skips the idleDuration threshold, used when woken by an event. +func (m *Manager) processMailboxes(bypassIdleCheck bool) { defer func() { if r := recover(); r != nil { logging.Logger().Error("panic in processMailboxes", zap.Any("panic", r)) @@ -263,9 +282,10 @@ func (m *Manager) processMailboxes() { continue } - idleDuration := tmuxSess.IdleDuration() - if idleDuration < 5*time.Second { - continue + if !bypassIdleCheck { + if tmuxSess.IdleDuration() < 5*time.Second { + continue + } } // Check if inbox has messages