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 <name>.md listing. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Levi Neely
parent
15bf980298
commit
29f7b45adc
|
|
@ -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 <name>.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 → <name>.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/<focus-area> - 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/<name>.md.
|
||||
func (r *RolesFS) Read(path string) ([]byte, error) {
|
||||
parts := strings.Split(strings.TrimPrefix(path, "/"), "/")
|
||||
|
||||
// /agent/roles/<focus-area>/<role-name>.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/<name>.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 {
|
||||
|
|
|
|||
|
|
@ -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: <name>.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/<focus-area> - 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/<focus-area> - list role files
|
||||
focusArea := strings.TrimPrefix(path, "/roles/")
|
||||
if s.roles != nil {
|
||||
roles, _ := s.roles.listRolesInFocusArea(focusArea)
|
||||
// Flat: list <name>.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",
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
Reference in New Issue