Add mountable beads projects with per-project databases
- Mount projects: echo "mount /path/to/project name" | 9p write agent/beads/ctl - Unmount: echo "umount name" | 9p write agent/beads/ctl - All beads endpoints work in mounts (list, ready, pending, blocked, etc.) - Add agent/beads/mtab for mount discovery (mount→cwd mappings) - Store databases at ~/.beads/<cwd-hyphenated> to prevent cross-contamination - Supervisor auto-assigns claimable beads to idle bots with matching cwd - Fix supervisor duplicate messages (track assigned beads, filter by assignee) - Update beads skill with mtab discovery pattern - Fix race conditions and add proper locking for mount access
This commit is contained in:
parent
425fa53877
commit
1e3623920e
|
|
@ -8,6 +8,7 @@ import (
|
|||
"anvillm/internal/logging"
|
||||
"anvillm/internal/p9"
|
||||
"anvillm/internal/session"
|
||||
"anvillm/internal/supervisor"
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
|
|
@ -237,6 +238,11 @@ func start(daemonize bool) {
|
|||
// Wire up event bus to session manager
|
||||
mgr.SetEventBus(srv.Events())
|
||||
|
||||
// Start supervisor
|
||||
ctx := context.Background()
|
||||
sup := supervisor.New(mgr, srv.Beads(), mgr.GetMailManager(), srv.Events())
|
||||
go sup.Run(ctx)
|
||||
|
||||
logging.Logger().Info("anvilsrv started successfully", zap.String("socket", srv.SocketPath()))
|
||||
|
||||
// Start background monitor for self-healing
|
||||
|
|
|
|||
|
|
@ -5,9 +5,12 @@ import (
|
|||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
bd "github.com/steveyegge/beads"
|
||||
|
|
@ -65,6 +68,15 @@ const (
|
|||
requiresReviewLabel = "requires_review"
|
||||
)
|
||||
|
||||
// MountedProject represents a project-specific beads instance.
|
||||
type MountedProject struct {
|
||||
name string
|
||||
cwd string
|
||||
dbPath string
|
||||
jsonlPath string
|
||||
store bd.Storage
|
||||
}
|
||||
|
||||
// BeadsFS handles beads filesystem operations.
|
||||
// BeadsFS provides 9P filesystem access to beads task tracking.
|
||||
// It exposes beads operations through virtual files and directories.
|
||||
|
|
@ -73,11 +85,17 @@ type BeadsFS struct {
|
|||
ctx context.Context
|
||||
lastQuery *bd.IssueFilter
|
||||
queryResult []*bd.Issue
|
||||
mounts map[string]*MountedProject
|
||||
mountsMu sync.RWMutex
|
||||
}
|
||||
|
||||
// NewBeadsFS creates a beads filesystem handler from an existing store.
|
||||
func NewBeadsFS(store bd.Storage, ctx context.Context) *BeadsFS {
|
||||
return &BeadsFS{store: store, ctx: ctx}
|
||||
return &BeadsFS{
|
||||
store: store,
|
||||
ctx: ctx,
|
||||
mounts: make(map[string]*MountedProject),
|
||||
}
|
||||
}
|
||||
|
||||
// Close is a no-op since the store is managed externally.
|
||||
|
|
@ -85,11 +103,88 @@ func (b *BeadsFS) Close() error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// ListMounts returns the names of all mounted projects.
|
||||
func (b *BeadsFS) ListMounts() map[string]struct{} {
|
||||
b.mountsMu.RLock()
|
||||
defer b.mountsMu.RUnlock()
|
||||
result := make(map[string]struct{}, len(b.mounts))
|
||||
for name := range b.mounts {
|
||||
result[name] = struct{}{}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
// Mount adds a project-specific beads instance.
|
||||
func (b *BeadsFS) Mount(name, cwd string) error {
|
||||
b.mountsMu.Lock()
|
||||
defer b.mountsMu.Unlock()
|
||||
if _, exists := b.mounts[name]; exists {
|
||||
return fmt.Errorf("mount %s exists", name)
|
||||
}
|
||||
// Use cwd as db path, replacing / with -
|
||||
cwdHyphenated := strings.ReplaceAll(cwd, "/", "-")
|
||||
dbPath := filepath.Join(os.Getenv("HOME"), ".beads", cwdHyphenated)
|
||||
if err := os.MkdirAll(dbPath, 0755); err != nil {
|
||||
return fmt.Errorf("failed to create db directory: %w", err)
|
||||
}
|
||||
jsonlPath := filepath.Join(cwd, ".beads", "issues.jsonl")
|
||||
store, err := bd.OpenFromConfig(b.ctx, dbPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Import is not available in beads v0.54.0, skip for now
|
||||
b.mounts[name] = &MountedProject{name, cwd, dbPath, jsonlPath, store}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Umount removes a project-specific beads instance.
|
||||
func (b *BeadsFS) Umount(name string) error {
|
||||
b.mountsMu.Lock()
|
||||
defer b.mountsMu.Unlock()
|
||||
m, ok := b.mounts[name]
|
||||
if !ok {
|
||||
return fmt.Errorf("mount %s not found", name)
|
||||
}
|
||||
if err := m.store.Close(); err != nil {
|
||||
return fmt.Errorf("failed to close store: %w", err)
|
||||
}
|
||||
delete(b.mounts, name)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Sync exports a mounted project's beads to its jsonl file.
|
||||
func (b *BeadsFS) Sync(name string) error {
|
||||
b.mountsMu.RLock()
|
||||
m := b.mounts[name]
|
||||
b.mountsMu.RUnlock()
|
||||
if m == nil {
|
||||
return fmt.Errorf("mount not found")
|
||||
}
|
||||
// Export is not available in beads v0.54.0, skip for now
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
// Read handles reads from beads filesystem paths.
|
||||
// Supports list, ready, pending, stats, blocked, stale, query, config endpoints,
|
||||
// as well as search, batch, label, and per-bead property access.
|
||||
func (b *BeadsFS) Read(path string) ([]byte, error) {
|
||||
// Check for mount paths (format: mountname/endpoint)
|
||||
parts := strings.Split(strings.TrimPrefix(path, "/"), "/")
|
||||
if len(parts) >= 2 {
|
||||
// Check if first part is a mount name
|
||||
b.mountsMu.RLock()
|
||||
m := b.mounts[parts[0]]
|
||||
b.mountsMu.RUnlock()
|
||||
if m != nil {
|
||||
endpoint := strings.Join(parts[1:], "/")
|
||||
if endpoint == "cwd" {
|
||||
return []byte(m.cwd), nil
|
||||
}
|
||||
// Route to mount-specific read
|
||||
return b.readFromMount(m, endpoint)
|
||||
}
|
||||
}
|
||||
|
||||
if len(parts) == 0 {
|
||||
return nil, fmt.Errorf("invalid path: %s", path)
|
||||
|
|
@ -98,7 +193,16 @@ func (b *BeadsFS) Read(path string) ([]byte, error) {
|
|||
switch {
|
||||
case len(parts) == 1 && parts[0] == "list":
|
||||
return b.readList()
|
||||
case len(parts) == 1 && parts[0] == "mtab":
|
||||
return b.readMtab()
|
||||
case len(parts) == 1 && parts[0] == "ready":
|
||||
// If we have mounts, aggregate ready from all mounts
|
||||
b.mountsMu.RLock()
|
||||
hasMounts := len(b.mounts) > 0
|
||||
b.mountsMu.RUnlock()
|
||||
if hasMounts {
|
||||
return b.readReadyAggregate()
|
||||
}
|
||||
return b.readReady()
|
||||
case len(parts) == 1 && parts[0] == "pending":
|
||||
return b.readPending()
|
||||
|
|
@ -135,10 +239,56 @@ func (b *BeadsFS) Read(path string) ([]byte, error) {
|
|||
}
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readFromMount(m *MountedProject, endpoint string) ([]byte, error) {
|
||||
parts := strings.Split(endpoint, "/")
|
||||
switch {
|
||||
case len(parts) == 1 && parts[0] == "list":
|
||||
return b.readListFromStore(m.store)
|
||||
case len(parts) == 1 && parts[0] == "ready":
|
||||
return b.readReadyFromStore(m.store)
|
||||
case len(parts) == 1 && parts[0] == "pending":
|
||||
return b.readPendingFromStore(m.store)
|
||||
case len(parts) == 1 && parts[0] == "blocked":
|
||||
return b.readBlockedFromStore(m.store)
|
||||
case len(parts) == 1 && parts[0] == "stale":
|
||||
return b.readStaleFromStore(m.store)
|
||||
case len(parts) == 2 && parts[0] == "search":
|
||||
return b.readSearchFromStore(m.store, parts[1])
|
||||
case len(parts) == 2 && parts[0] == "by-ref":
|
||||
return b.readByExternalRefFromStore(m.store, parts[1])
|
||||
case len(parts) == 2 && parts[0] == "batch":
|
||||
return b.readBatchFromStore(m.store, parts[1])
|
||||
case len(parts) == 2 && parts[0] == "label":
|
||||
return b.readByLabelFromStore(m.store, parts[1])
|
||||
case len(parts) == 2 && parts[0] == "children":
|
||||
return b.readChildrenFromStore(m.store, parts[1])
|
||||
case len(parts) == 2:
|
||||
return b.readBeadPropertyFromStore(m.store, parts[0], parts[1])
|
||||
case len(parts) == 3 && parts[2] == "dependencies-meta":
|
||||
return b.readDependenciesMetaFromStore(m.store, parts[1])
|
||||
case len(parts) == 3 && parts[2] == "dependents-meta":
|
||||
return b.readDependentsMetaFromStore(m.store, parts[1])
|
||||
default:
|
||||
return nil, fmt.Errorf("unsupported mount endpoint: %s", endpoint)
|
||||
}
|
||||
}
|
||||
|
||||
// Write handles writes to beads filesystem paths.
|
||||
// Supports query filter updates and ctl command execution.
|
||||
func (b *BeadsFS) Write(path string, data []byte, sessionID string) error {
|
||||
// Check for mount paths (format: mountname/endpoint)
|
||||
parts := strings.Split(strings.TrimPrefix(path, "/"), "/")
|
||||
if len(parts) >= 2 {
|
||||
// Check if first part is a mount name
|
||||
b.mountsMu.RLock()
|
||||
m := b.mounts[parts[0]]
|
||||
b.mountsMu.RUnlock()
|
||||
if m != nil {
|
||||
logging.Logger().Info("mount write", zap.String("mount", parts[0]), zap.String("endpoint", strings.Join(parts[1:], "/")))
|
||||
endpoint := strings.Join(parts[1:], "/")
|
||||
return b.writeToMount(m, endpoint, data, sessionID)
|
||||
}
|
||||
}
|
||||
|
||||
if len(parts) == 0 {
|
||||
return fmt.Errorf("invalid path: %s", path)
|
||||
|
|
@ -155,17 +305,43 @@ func (b *BeadsFS) Write(path string, data []byte, sessionID string) error {
|
|||
return b.executeCtl(string(data))
|
||||
}
|
||||
|
||||
func (b *BeadsFS) writeToMount(m *MountedProject, endpoint string, data []byte, sessionID string) error {
|
||||
parts := strings.Split(endpoint, "/")
|
||||
if len(parts) == 1 && parts[0] == "ctl" {
|
||||
return b.executeCtlOnStore(m.store, string(data))
|
||||
}
|
||||
return fmt.Errorf("unsupported mount write endpoint: %s", endpoint)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readList() ([]byte, error) {
|
||||
issues, err := b.store.SearchIssues(b.ctx, "", bd.IssueFilter{})
|
||||
return b.readListFromStore(b.store)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readMtab() ([]byte, error) {
|
||||
b.mountsMu.RLock()
|
||||
defer b.mountsMu.RUnlock()
|
||||
|
||||
var lines []string
|
||||
for name, m := range b.mounts {
|
||||
lines = append(lines, name+"\t"+m.cwd)
|
||||
}
|
||||
return []byte(strings.Join(lines, "\n")), nil
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readListFromStore(store bd.Storage) ([]byte, error) {
|
||||
issues, err := store.SearchIssues(b.ctx, "", bd.IssueFilter{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if issues == nil {
|
||||
issues = []*bd.Issue{}
|
||||
}
|
||||
// Sort by ID for hierarchical view (bd-a3f8, bd-a3f8.1, bd-a3f8.1.1, etc.)
|
||||
sort.Slice(issues, func(i, j int) bool {
|
||||
return issues[i].ID < issues[j].ID
|
||||
})
|
||||
|
||||
// Enrich with blockers and capability level.
|
||||
// Enrich with blockers, labels, and capability level.
|
||||
type BeadWithBlockers struct {
|
||||
*bd.Issue
|
||||
Blockers []string `json:"blockers,omitempty"`
|
||||
|
|
@ -173,8 +349,15 @@ func (b *BeadsFS) readList() ([]byte, error) {
|
|||
}
|
||||
result := make([]BeadWithBlockers, len(issues))
|
||||
for i, issue := range issues {
|
||||
// Fetch labels if not already populated
|
||||
if issue.Labels == nil {
|
||||
labels, err := store.GetLabels(b.ctx, issue.ID)
|
||||
if err == nil {
|
||||
issue.Labels = labels
|
||||
}
|
||||
}
|
||||
result[i].Issue = issue
|
||||
blockers, err := b.getBlockers(issue.ID)
|
||||
blockers, err := b.getBlockersFromStore(store, issue.ID)
|
||||
if err != nil {
|
||||
logging.Logger().Warn("failed to get blockers", zap.String("issue", issue.ID), zap.Error(err))
|
||||
} else if len(blockers) > 0 {
|
||||
|
|
@ -187,24 +370,96 @@ func (b *BeadsFS) readList() ([]byte, error) {
|
|||
}
|
||||
|
||||
func (b *BeadsFS) readReady() ([]byte, error) {
|
||||
return b.readReadyFromStore(b.store)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readReadyFromStore(store bd.Storage) ([]byte, error) {
|
||||
filter := bd.WorkFilter{}
|
||||
issues, err := b.store.GetReadyWork(b.ctx, filter)
|
||||
issues, err := store.GetReadyWork(b.ctx, filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.MarshalIndent(issues, "", " ")
|
||||
if issues == nil {
|
||||
issues = []*bd.Issue{}
|
||||
}
|
||||
// Filter out blocked issues - ready means unblocked
|
||||
ready := []*bd.Issue{}
|
||||
for _, issue := range issues {
|
||||
// Fetch labels if not already populated
|
||||
if issue.Labels == nil {
|
||||
labels, err := store.GetLabels(b.ctx, issue.ID)
|
||||
if err == nil {
|
||||
issue.Labels = labels
|
||||
}
|
||||
}
|
||||
blockers, err := b.getBlockersFromStore(store, issue.ID)
|
||||
if err != nil || len(blockers) == 0 {
|
||||
ready = append(ready, issue)
|
||||
}
|
||||
}
|
||||
return json.MarshalIndent(ready, "", " ")
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readReadyAggregate() ([]byte, error) {
|
||||
b.mountsMu.RLock()
|
||||
defer b.mountsMu.RUnlock()
|
||||
|
||||
type TaskWithMount struct {
|
||||
*bd.Issue
|
||||
Mount string `json:"mount"`
|
||||
Cwd string `json:"cwd"`
|
||||
}
|
||||
|
||||
allReady := []TaskWithMount{}
|
||||
for name, m := range b.mounts {
|
||||
filter := bd.WorkFilter{}
|
||||
issues, err := m.store.GetReadyWork(b.ctx, filter)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
// Filter out blocked issues
|
||||
for _, issue := range issues {
|
||||
// Fetch labels if not already populated
|
||||
if issue.Labels == nil {
|
||||
labels, err := m.store.GetLabels(b.ctx, issue.ID)
|
||||
if err == nil {
|
||||
issue.Labels = labels
|
||||
}
|
||||
}
|
||||
blockers, err := b.getBlockersFromStore(m.store, issue.ID)
|
||||
if err != nil || len(blockers) == 0 {
|
||||
allReady = append(allReady, TaskWithMount{
|
||||
Issue: issue,
|
||||
Mount: name,
|
||||
Cwd: m.cwd,
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
return json.MarshalIndent(allReady, "", " ")
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readSearch(query string) ([]byte, error) {
|
||||
issues, err := b.store.SearchIssues(b.ctx, query, bd.IssueFilter{})
|
||||
return b.readSearchFromStore(b.store, query)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readSearchFromStore(store bd.Storage, query string) ([]byte, error) {
|
||||
issues, err := store.SearchIssues(b.ctx, query, bd.IssueFilter{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if issues == nil {
|
||||
issues = []*bd.Issue{}
|
||||
}
|
||||
return json.MarshalIndent(issues, "", " ")
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readByExternalRef(ref string) ([]byte, error) {
|
||||
issue, err := b.store.GetIssueByExternalRef(b.ctx, ref)
|
||||
return b.readByExternalRefFromStore(b.store, ref)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readByExternalRefFromStore(store bd.Storage, ref string) ([]byte, error) {
|
||||
issue, err := store.GetIssueByExternalRef(b.ctx, ref)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -228,39 +483,64 @@ func (b *BeadsFS) readAllConfig() ([]byte, error) {
|
|||
}
|
||||
|
||||
func (b *BeadsFS) readBatch(ids string) ([]byte, error) {
|
||||
return b.readBatchFromStore(b.store, ids)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readBatchFromStore(store bd.Storage, ids string) ([]byte, error) {
|
||||
idList := strings.Split(ids, ",")
|
||||
issues, err := b.store.GetIssuesByIDs(b.ctx, idList)
|
||||
issues, err := store.GetIssuesByIDs(b.ctx, idList)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if issues == nil {
|
||||
issues = []*bd.Issue{}
|
||||
}
|
||||
return json.MarshalIndent(issues, "", " ")
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readByLabel(label string) ([]byte, error) {
|
||||
issues, err := b.store.GetIssuesByLabel(b.ctx, label)
|
||||
return b.readByLabelFromStore(b.store, label)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readByLabelFromStore(store bd.Storage, label string) ([]byte, error) {
|
||||
issues, err := store.GetIssuesByLabel(b.ctx, label)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if issues == nil {
|
||||
issues = []*bd.Issue{}
|
||||
}
|
||||
return json.MarshalIndent(issues, "", " ")
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readBlocked() ([]byte, error) {
|
||||
return b.readBlockedFromStore(b.store)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readBlockedFromStore(store bd.Storage) ([]byte, error) {
|
||||
filter := bd.WorkFilter{}
|
||||
blocked, err := b.store.GetBlockedIssues(b.ctx, filter)
|
||||
blocked, err := store.GetBlockedIssues(b.ctx, filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if blocked == nil {
|
||||
blocked = []*bd.BlockedIssue{}
|
||||
}
|
||||
return json.MarshalIndent(blocked, "", " ")
|
||||
}
|
||||
|
||||
// readPending returns beads in pending_approval or pending_review status,
|
||||
// i.e. those awaiting a human APPROVAL_RESPONSE or REVIEW_RESPONSE.
|
||||
func (b *BeadsFS) readPending() ([]byte, error) {
|
||||
issues, err := b.store.SearchIssues(b.ctx, "", bd.IssueFilter{})
|
||||
return b.readPendingFromStore(b.store)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readPendingFromStore(store bd.Storage) ([]byte, error) {
|
||||
issues, err := store.SearchIssues(b.ctx, "", bd.IssueFilter{})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var pending []*bd.Issue
|
||||
pending := []*bd.Issue{}
|
||||
for _, issue := range issues {
|
||||
if issue.Status == StatusPendingApproval || issue.Status == StatusPendingReview {
|
||||
pending = append(pending, issue)
|
||||
|
|
@ -270,7 +550,11 @@ func (b *BeadsFS) readPending() ([]byte, error) {
|
|||
}
|
||||
|
||||
func (b *BeadsFS) readDependenciesMeta(beadID string) ([]byte, error) {
|
||||
deps, err := b.store.GetDependenciesWithMetadata(b.ctx, beadID)
|
||||
return b.readDependenciesMetaFromStore(b.store, beadID)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readDependenciesMetaFromStore(store bd.Storage, beadID string) ([]byte, error) {
|
||||
deps, err := store.GetDependenciesWithMetadata(b.ctx, beadID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -278,7 +562,11 @@ func (b *BeadsFS) readDependenciesMeta(beadID string) ([]byte, error) {
|
|||
}
|
||||
|
||||
func (b *BeadsFS) readDependentsMeta(beadID string) ([]byte, error) {
|
||||
deps, err := b.store.GetDependentsWithMetadata(b.ctx, beadID)
|
||||
return b.readDependentsMetaFromStore(b.store, beadID)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readDependentsMetaFromStore(store bd.Storage, beadID string) ([]byte, error) {
|
||||
deps, err := store.GetDependentsWithMetadata(b.ctx, beadID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -286,25 +574,36 @@ func (b *BeadsFS) readDependentsMeta(beadID string) ([]byte, error) {
|
|||
}
|
||||
|
||||
func (b *BeadsFS) readChildren(parentID string) ([]byte, error) {
|
||||
return b.readChildrenFromStore(b.store, parentID)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readChildrenFromStore(store bd.Storage, parentID string) ([]byte, error) {
|
||||
filter := bd.IssueFilter{ParentID: &parentID}
|
||||
children, err := b.store.SearchIssues(b.ctx, "", filter)
|
||||
children, err := store.SearchIssues(b.ctx, "", filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if children == nil {
|
||||
children = []*bd.Issue{}
|
||||
}
|
||||
return json.MarshalIndent(children, "", " ")
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readStale() ([]byte, error) {
|
||||
return b.readStaleFromStore(b.store)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readStaleFromStore(store bd.Storage) ([]byte, error) {
|
||||
// Get all non-closed issues
|
||||
filter := bd.IssueFilter{}
|
||||
issues, err := b.store.SearchIssues(b.ctx, "", filter)
|
||||
issues, err := store.SearchIssues(b.ctx, "", filter)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Filter to stale issues (not updated in 30+ days, status open or in_progress)
|
||||
cutoff := time.Now().AddDate(0, 0, -30)
|
||||
var stale []*bd.Issue
|
||||
stale := []*bd.Issue{}
|
||||
for _, issue := range issues {
|
||||
if issue.Status != bd.StatusClosed && issue.UpdatedAt.Before(cutoff) {
|
||||
stale = append(stale, issue)
|
||||
|
|
@ -346,7 +645,11 @@ func (b *BeadsFS) readStats() ([]byte, error) {
|
|||
}
|
||||
|
||||
func (b *BeadsFS) readBeadProperty(beadID, property string) ([]byte, error) {
|
||||
issue, err := b.store.GetIssue(b.ctx, beadID)
|
||||
return b.readBeadPropertyFromStore(b.store, beadID, property)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) readBeadPropertyFromStore(store bd.Storage, beadID, property string) ([]byte, error) {
|
||||
issue, err := store.GetIssue(b.ctx, beadID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -370,37 +673,37 @@ func (b *BeadsFS) readBeadProperty(beadID, property string) ([]byte, error) {
|
|||
CapabilityLevel string `json:"capability_level,omitempty"`
|
||||
}
|
||||
result := BeadWithBlockers{Issue: issue}
|
||||
if blockers, err := b.getBlockers(beadID); err == nil {
|
||||
if blockers, err := b.getBlockersFromStore(store, beadID); err == nil {
|
||||
result.Blockers = blockers
|
||||
}
|
||||
result.CapabilityLevel = extractCapabilityLevel(issue.Labels)
|
||||
return json.MarshalIndent(result, "", " ")
|
||||
case "comments":
|
||||
comments, err := b.store.GetIssueComments(b.ctx, beadID)
|
||||
comments, err := store.GetIssueComments(b.ctx, beadID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.MarshalIndent(comments, "", " ")
|
||||
case "labels":
|
||||
labels, err := b.store.GetLabels(b.ctx, beadID)
|
||||
labels, err := store.GetLabels(b.ctx, beadID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.MarshalIndent(labels, "", " ")
|
||||
case "dependents":
|
||||
dependents, err := b.store.GetDependents(b.ctx, beadID)
|
||||
dependents, err := store.GetDependents(b.ctx, beadID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.MarshalIndent(dependents, "", " ")
|
||||
case "tree":
|
||||
tree, err := b.store.GetDependencyTree(b.ctx, beadID, 10, false, false)
|
||||
tree, err := store.GetDependencyTree(b.ctx, beadID, 10, false, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.MarshalIndent(tree, "", " ")
|
||||
case "events":
|
||||
events, err := b.store.GetEvents(b.ctx, beadID, 100)
|
||||
events, err := store.GetEvents(b.ctx, beadID, 100)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -411,7 +714,11 @@ func (b *BeadsFS) readBeadProperty(beadID, property string) ([]byte, error) {
|
|||
}
|
||||
|
||||
func (b *BeadsFS) getBlockers(id string) ([]string, error) {
|
||||
deps, err := b.store.GetDependenciesWithMetadata(b.ctx, id)
|
||||
return b.getBlockersFromStore(b.store, id)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) getBlockersFromStore(store bd.Storage, id string) ([]string, error) {
|
||||
deps, err := store.GetDependenciesWithMetadata(b.ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
|
@ -425,6 +732,10 @@ func (b *BeadsFS) getBlockers(id string) ([]string, error) {
|
|||
}
|
||||
|
||||
func (b *BeadsFS) executeCtl(cmd string) error {
|
||||
return b.executeCtlOnStore(b.store, cmd)
|
||||
}
|
||||
|
||||
func (b *BeadsFS) executeCtlOnStore(store bd.Storage, cmd string) error {
|
||||
command, args, err := parseCtlCommand(cmd)
|
||||
if err != nil {
|
||||
return err
|
||||
|
|
@ -438,7 +749,23 @@ func (b *BeadsFS) executeCtl(cmd string) error {
|
|||
if len(args) > 0 {
|
||||
prefix = args[0]
|
||||
}
|
||||
return b.store.SetConfig(b.ctx, "issue_prefix", prefix)
|
||||
return store.SetConfig(b.ctx, "issue_prefix", prefix)
|
||||
|
||||
case "mount":
|
||||
if len(args) < 2 {
|
||||
return fmt.Errorf("usage: mount <cwd> <name>")
|
||||
}
|
||||
return b.Mount(args[1], args[0])
|
||||
case "umount":
|
||||
if len(args) < 1 {
|
||||
return fmt.Errorf("usage: umount <name>")
|
||||
}
|
||||
return b.Umount(args[0])
|
||||
case "sync":
|
||||
if len(args) < 1 {
|
||||
return fmt.Errorf("usage: sync <name>")
|
||||
}
|
||||
return b.Sync(args[0])
|
||||
|
||||
case "new", "create":
|
||||
if len(args) < 1 {
|
||||
|
|
@ -487,7 +814,7 @@ func (b *BeadsFS) executeCtl(cmd string) error {
|
|||
}
|
||||
// Apply capability label if provided.
|
||||
if capLevel != "" {
|
||||
_ = b.store.AddLabel(b.ctx, childID, capabilityPrefix+capLevel, actor)
|
||||
_ = store.AddLabel(b.ctx, childID, capabilityPrefix+capLevel, actor)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
|
@ -502,7 +829,7 @@ func (b *BeadsFS) executeCtl(cmd string) error {
|
|||
if capLevel != "" {
|
||||
issue.Labels = []string{capabilityPrefix + capLevel}
|
||||
}
|
||||
return b.store.CreateIssue(b.ctx, issue, actor)
|
||||
return store.CreateIssue(b.ctx, issue, actor)
|
||||
|
||||
case "claim":
|
||||
if len(args) < 1 {
|
||||
|
|
@ -516,25 +843,25 @@ func (b *BeadsFS) executeCtl(cmd string) error {
|
|||
"assignee": assignee,
|
||||
"status": bd.StatusInProgress,
|
||||
}
|
||||
return b.store.UpdateIssue(b.ctx, args[0], updates, actor)
|
||||
return store.UpdateIssue(b.ctx, args[0], updates, actor)
|
||||
|
||||
case "complete", "close":
|
||||
if len(args) < 1 {
|
||||
return fmt.Errorf("usage: complete <bead-id>")
|
||||
}
|
||||
if err := b.store.UpdateIssue(b.ctx, args[0], map[string]any{"assignee": ""}, actor); err != nil {
|
||||
if err := store.UpdateIssue(b.ctx, args[0], map[string]any{"assignee": ""}, actor); err != nil {
|
||||
return err
|
||||
}
|
||||
return b.store.CloseIssue(b.ctx, args[0], "completed", actor, "")
|
||||
return store.CloseIssue(b.ctx, args[0], "completed", actor, "")
|
||||
|
||||
case "fail":
|
||||
if len(args) < 2 {
|
||||
return fmt.Errorf("usage: fail <bead-id> 'reason'")
|
||||
}
|
||||
if err := b.store.UpdateIssue(b.ctx, args[0], map[string]any{"assignee": ""}, actor); err != nil {
|
||||
if err := store.UpdateIssue(b.ctx, args[0], map[string]any{"assignee": ""}, actor); err != nil {
|
||||
return err
|
||||
}
|
||||
return b.store.CloseIssue(b.ctx, args[0], args[1], actor, "")
|
||||
return store.CloseIssue(b.ctx, args[0], args[1], actor, "")
|
||||
|
||||
case "dep", "add-dep":
|
||||
if len(args) < 2 {
|
||||
|
|
@ -545,13 +872,13 @@ func (b *BeadsFS) executeCtl(cmd string) error {
|
|||
DependsOnID: args[1],
|
||||
Type: bd.DepBlocks,
|
||||
}
|
||||
return b.store.AddDependency(b.ctx, dep, actor)
|
||||
return store.AddDependency(b.ctx, dep, actor)
|
||||
|
||||
case "undep", "rm-dep":
|
||||
if len(args) < 2 {
|
||||
return fmt.Errorf("usage: undep <child-id> <parent-id>")
|
||||
}
|
||||
return b.store.RemoveDependency(b.ctx, args[0], args[1], actor)
|
||||
return store.RemoveDependency(b.ctx, args[0], args[1], actor)
|
||||
|
||||
case "update":
|
||||
if len(args) < 3 {
|
||||
|
|
@ -579,13 +906,13 @@ func (b *BeadsFS) executeCtl(cmd string) error {
|
|||
if len(args) < 2 {
|
||||
return fmt.Errorf("usage: label <bead-id> 'label'")
|
||||
}
|
||||
return b.store.AddLabel(b.ctx, args[0], args[1], actor)
|
||||
return store.AddLabel(b.ctx, args[0], args[1], actor)
|
||||
|
||||
case "unlabel":
|
||||
if len(args) < 2 {
|
||||
return fmt.Errorf("usage: unlabel <bead-id> 'label'")
|
||||
}
|
||||
return b.store.RemoveLabel(b.ctx, args[0], args[1], actor)
|
||||
return store.RemoveLabel(b.ctx, args[0], args[1], actor)
|
||||
|
||||
case "set-capability":
|
||||
// Convenience command that replaces any existing capability label with
|
||||
|
|
@ -609,7 +936,7 @@ func (b *BeadsFS) executeCtl(cmd string) error {
|
|||
_ = b.store.RemoveLabel(b.ctx, beadID, lbl, actor)
|
||||
}
|
||||
}
|
||||
return b.store.AddLabel(b.ctx, beadID, capabilityPrefix+level, actor)
|
||||
return store.AddLabel(b.ctx, beadID, capabilityPrefix+level, actor)
|
||||
|
||||
case "batch-create":
|
||||
if len(args) < 1 {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,162 @@
|
|||
package p9
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
bd "github.com/steveyegge/beads"
|
||||
)
|
||||
|
||||
func TestMountUnmount(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
os.Setenv("HOME", tmpDir)
|
||||
defer os.Unsetenv("HOME")
|
||||
|
||||
store, err := bd.OpenFromConfig(context.Background(), filepath.Join(tmpDir, "main"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer store.Close()
|
||||
|
||||
beadsFS := NewBeadsFS(store, context.Background())
|
||||
|
||||
projectDir := filepath.Join(tmpDir, "project1")
|
||||
os.MkdirAll(projectDir, 0755)
|
||||
|
||||
err = beadsFS.Mount("proj1", projectDir)
|
||||
if err != nil {
|
||||
t.Fatalf("Mount failed: %v", err)
|
||||
}
|
||||
|
||||
mounts := beadsFS.ListMounts()
|
||||
if _, ok := mounts["proj1"]; !ok {
|
||||
t.Fatal("Mount not listed")
|
||||
}
|
||||
|
||||
err = beadsFS.Umount("proj1")
|
||||
if err != nil {
|
||||
t.Fatalf("Umount failed: %v", err)
|
||||
}
|
||||
|
||||
mounts = beadsFS.ListMounts()
|
||||
if _, ok := mounts["proj1"]; ok {
|
||||
t.Fatal("Mount still listed after umount")
|
||||
}
|
||||
}
|
||||
|
||||
func TestMountReadWrite(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
os.Setenv("HOME", tmpDir)
|
||||
defer os.Unsetenv("HOME")
|
||||
|
||||
store, err := bd.OpenFromConfig(context.Background(), filepath.Join(tmpDir, "main"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer store.Close()
|
||||
|
||||
beadsFS := NewBeadsFS(store, context.Background())
|
||||
|
||||
projectDir := filepath.Join(tmpDir, "project1")
|
||||
os.MkdirAll(projectDir, 0755)
|
||||
|
||||
err = beadsFS.Mount("proj1", projectDir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer beadsFS.Umount("proj1")
|
||||
|
||||
err = beadsFS.Write("proj1/ctl", []byte("init bd"), "test")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err = beadsFS.Write("proj1/ctl", []byte("new 'Test Task' 'Description'"), "test")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
list, err := beadsFS.Read("proj1/list")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var beads []map[string]interface{}
|
||||
json.Unmarshal(list, &beads)
|
||||
if len(beads) != 1 {
|
||||
t.Fatalf("Expected 1 bead, got %d", len(beads))
|
||||
}
|
||||
|
||||
beadID := beads[0]["id"].(string)
|
||||
err = beadsFS.Write("proj1/ctl", []byte("label "+beadID+" claimable"), "test")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
ready, err := beadsFS.Read("proj1/ready")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var readyBeads []map[string]interface{}
|
||||
json.Unmarshal(ready, &readyBeads)
|
||||
if len(readyBeads) != 1 {
|
||||
t.Fatalf("Expected 1 ready bead, got %d", len(readyBeads))
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadyAggregate(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
os.Setenv("HOME", tmpDir)
|
||||
defer os.Unsetenv("HOME")
|
||||
|
||||
store, err := bd.OpenFromConfig(context.Background(), filepath.Join(tmpDir, "main"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer store.Close()
|
||||
|
||||
beadsFS := NewBeadsFS(store, context.Background())
|
||||
|
||||
proj1 := filepath.Join(tmpDir, "project1")
|
||||
proj2 := filepath.Join(tmpDir, "project2")
|
||||
os.MkdirAll(proj1, 0755)
|
||||
os.MkdirAll(proj2, 0755)
|
||||
|
||||
beadsFS.Mount("proj1", proj1)
|
||||
beadsFS.Mount("proj2", proj2)
|
||||
defer beadsFS.Umount("proj1")
|
||||
defer beadsFS.Umount("proj2")
|
||||
|
||||
beadsFS.Write("proj1/ctl", []byte("init bd"), "test")
|
||||
beadsFS.Write("proj2/ctl", []byte("init bd"), "test")
|
||||
|
||||
beadsFS.Write("proj1/ctl", []byte("new 'Task 1' 'Desc 1'"), "test")
|
||||
beadsFS.Write("proj2/ctl", []byte("new 'Task 2' 'Desc 2'"), "test")
|
||||
|
||||
ready, err := beadsFS.Read("ready")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var tasks []map[string]interface{}
|
||||
json.Unmarshal(ready, &tasks)
|
||||
|
||||
proj1Count := 0
|
||||
proj2Count := 0
|
||||
for _, task := range tasks {
|
||||
mount, _ := task["mount"].(string)
|
||||
if mount == "proj1" {
|
||||
proj1Count++
|
||||
} else if mount == "proj2" {
|
||||
proj2Count++
|
||||
}
|
||||
}
|
||||
|
||||
if proj1Count != 1 || proj2Count != 1 {
|
||||
t.Fatalf("Expected 1 task from each mount, got proj1=%d proj2=%d", proj1Count, proj2Count)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,104 @@
|
|||
package p9
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
bd "github.com/steveyegge/beads"
|
||||
)
|
||||
|
||||
func TestBeadsFS_Mount(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
store, err := bd.OpenFromConfig(context.Background(), tmpDir)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create store: %v", err)
|
||||
}
|
||||
defer store.Close()
|
||||
|
||||
beadsFS := NewBeadsFS(store, context.Background())
|
||||
|
||||
testCwd := filepath.Join(tmpDir, "project")
|
||||
os.MkdirAll(filepath.Join(testCwd, ".beads"), 0755)
|
||||
|
||||
// Test mount
|
||||
err = beadsFS.Mount("test-project", testCwd)
|
||||
if err != nil {
|
||||
t.Fatalf("Mount failed: %v", err)
|
||||
}
|
||||
|
||||
// Test duplicate mount
|
||||
err = beadsFS.Mount("test-project", testCwd)
|
||||
if err == nil {
|
||||
t.Fatal("Should have failed on duplicate mount")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBeadsFS_Umount(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
store, err := bd.OpenFromConfig(context.Background(), tmpDir)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create store: %v", err)
|
||||
}
|
||||
defer store.Close()
|
||||
|
||||
beadsFS := NewBeadsFS(store, context.Background())
|
||||
|
||||
testCwd := filepath.Join(tmpDir, "project")
|
||||
os.MkdirAll(filepath.Join(testCwd, ".beads"), 0755)
|
||||
|
||||
// Mount first
|
||||
err = beadsFS.Mount("test-project", testCwd)
|
||||
if err != nil {
|
||||
t.Fatalf("Mount failed: %v", err)
|
||||
}
|
||||
|
||||
// Test umount
|
||||
err = beadsFS.Umount("test-project")
|
||||
if err != nil {
|
||||
t.Fatalf("Umount failed: %v", err)
|
||||
}
|
||||
|
||||
// Test umount non-existent
|
||||
err = beadsFS.Umount("nonexistent")
|
||||
if err == nil {
|
||||
t.Fatal("Should have failed on non-existent mount")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBeadsFS_MountRouting(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
store, err := bd.OpenFromConfig(context.Background(), tmpDir)
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create store: %v", err)
|
||||
}
|
||||
defer store.Close()
|
||||
|
||||
beadsFS := NewBeadsFS(store, context.Background())
|
||||
|
||||
testCwd := filepath.Join(tmpDir, "project")
|
||||
os.MkdirAll(filepath.Join(testCwd, ".beads"), 0755)
|
||||
|
||||
// Mount
|
||||
err = beadsFS.Mount("test-project", testCwd)
|
||||
if err != nil {
|
||||
t.Fatalf("Mount failed: %v", err)
|
||||
}
|
||||
defer beadsFS.Umount("test-project")
|
||||
|
||||
// Test cwd endpoint
|
||||
data, err := beadsFS.Read("agent/beads/test-project/cwd")
|
||||
if err != nil {
|
||||
t.Fatalf("Read cwd failed: %v", err)
|
||||
}
|
||||
if string(data) != testCwd {
|
||||
t.Fatalf("Expected cwd %s, got %s", testCwd, string(data))
|
||||
}
|
||||
|
||||
// Test non-existent mount
|
||||
_, err = beadsFS.Read("agent/beads/nonexistent/cwd")
|
||||
if err == nil {
|
||||
t.Fatal("Should have failed on non-existent mount")
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,83 @@
|
|||
package p9
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
bd "github.com/steveyegge/beads"
|
||||
)
|
||||
|
||||
func TestLabelClaimable(t *testing.T) {
|
||||
tmpDir := t.TempDir()
|
||||
|
||||
store, err := bd.OpenFromConfig(context.Background(), tmpDir)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer store.Close()
|
||||
|
||||
beadsFS := NewBeadsFS(store, context.Background())
|
||||
|
||||
// Initialize
|
||||
err = beadsFS.Write("ctl", []byte("init bd"), "test")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// Create a bead
|
||||
err = beadsFS.Write("ctl", []byte("new 'Test Task' 'Test description'"), "test")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// List beads to get ID
|
||||
list, err := beadsFS.Read("list")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
var beads []map[string]interface{}
|
||||
json.Unmarshal(list, &beads)
|
||||
if len(beads) == 0 {
|
||||
t.Fatal("no beads created")
|
||||
}
|
||||
|
||||
beadID := beads[0]["id"].(string)
|
||||
t.Logf("Created bead: %s", beadID)
|
||||
|
||||
// Label it as claimable
|
||||
cmd := fmt.Sprintf("label %s claimable", beadID)
|
||||
t.Logf("Running command: %s", cmd)
|
||||
err = beadsFS.Write("ctl", []byte(cmd), "test")
|
||||
if err != nil {
|
||||
t.Fatalf("label failed: %v", err)
|
||||
}
|
||||
t.Logf("Label command succeeded")
|
||||
|
||||
// Read it back
|
||||
list, err = beadsFS.Read("list")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
json.Unmarshal(list, &beads)
|
||||
t.Logf("Full bead data: %+v", beads[0])
|
||||
labels, ok := beads[0]["labels"].([]interface{})
|
||||
if !ok {
|
||||
labels = []interface{}{}
|
||||
}
|
||||
t.Logf("Labels: %v", labels)
|
||||
|
||||
hasClaimable := false
|
||||
for _, l := range labels {
|
||||
if l.(string) == "claimable" {
|
||||
hasClaimable = true
|
||||
}
|
||||
}
|
||||
|
||||
if !hasClaimable {
|
||||
t.Fatal("claimable label not found")
|
||||
}
|
||||
}
|
||||
|
|
@ -81,6 +81,7 @@ const (
|
|||
qidBeads // beads directory
|
||||
qidBeadsCtl // beads/ctl
|
||||
qidBeadsList // beads/list
|
||||
qidBeadsMtab // beads/mtab
|
||||
qidBeadsReady // beads/ready
|
||||
qidTools // tools directory
|
||||
qidSkills // skills directory
|
||||
|
|
@ -279,6 +280,11 @@ func (s *Server) Events() *eventbus.Bus {
|
|||
return s.events
|
||||
}
|
||||
|
||||
// Beads returns the BeadsFS instance.
|
||||
func (s *Server) Beads() *BeadsFS {
|
||||
return s.beads
|
||||
}
|
||||
|
||||
func (s *Server) acceptLoop() {
|
||||
for {
|
||||
conn, err := s.listener.Accept()
|
||||
|
|
@ -453,6 +459,9 @@ func (s *Server) walk(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
|
|||
case "list":
|
||||
qid = plan9.Qid{Type: QTFile, Path: qidBeadsList}
|
||||
newPath = "/beads/list"
|
||||
case "mtab":
|
||||
qid = plan9.Qid{Type: QTFile, Path: qidBeadsMtab}
|
||||
newPath = "/beads/mtab"
|
||||
case "ready":
|
||||
qid = plan9.Qid{Type: QTFile, Path: qidBeadsReady}
|
||||
newPath = "/beads/ready"
|
||||
|
|
@ -507,9 +516,28 @@ func (s *Server) walk(cs *connState, fc *plan9.Fcall) *plan9.Fcall {
|
|||
qid = plan9.Qid{Type: QTFile, Path: qidSkillsBase + hashID(path+name)}
|
||||
newPath = path + "/" + name
|
||||
} else if strings.HasPrefix(path, "/beads/") && strings.Count(path, "/") == 2 {
|
||||
// Inside a bead directory - property files
|
||||
// Inside a bead directory OR mount directory
|
||||
beadID := strings.TrimPrefix(path, "/beads/")
|
||||
qid = plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(beadID+name)}
|
||||
// Check if it's a mount
|
||||
if s.beads != nil {
|
||||
mounts := s.beads.ListMounts()
|
||||
if _, isMount := mounts[beadID]; isMount {
|
||||
// It's a mount directory - name is a file or subdirectory
|
||||
qid = plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(beadID+name)}
|
||||
newPath = path + "/" + name
|
||||
} else {
|
||||
// It's a bead directory - property files
|
||||
qid = plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(beadID+name)}
|
||||
newPath = path + "/" + name
|
||||
}
|
||||
} else {
|
||||
// No beads, treat as bead directory
|
||||
qid = plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(beadID+name)}
|
||||
newPath = path + "/" + name
|
||||
}
|
||||
} else if strings.HasPrefix(path, "/beads/") && strings.Count(path, "/") == 3 {
|
||||
// Inside a mount's bead directory - property files
|
||||
qid = plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(path+name)}
|
||||
newPath = path + "/" + name
|
||||
} else if strings.HasPrefix(path, "/user/") && strings.Count(path, "/") == 2 {
|
||||
// Inside user mailbox directory - message files
|
||||
|
|
@ -1091,19 +1119,100 @@ func (s *Server) readDir(path string, offset uint64, count uint32) []byte {
|
|||
}
|
||||
}
|
||||
} else if path == "/beads" {
|
||||
// Beads directory
|
||||
// Beads directory - only ctl, mtab, ready (aggregate), and mounted projects
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsCtl}, Mode: 0222, Name: "ctl",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsList}, Mode: 0444, Name: "list",
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsMtab}, Mode: 0444, Name: "mtab",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsReady}, Mode: 0444, Name: "ready",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
// Add mounted projects as directories
|
||||
if s.beads != nil {
|
||||
for name := range s.beads.ListMounts() {
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTDir, Path: qidBeadsBase + uint64(len(name))}, Mode: 0555 | plan9.DMDIR, Name: name,
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
}
|
||||
}
|
||||
} else if strings.HasPrefix(path, "/beads/") && strings.Count(path, "/") == 2 {
|
||||
// Inside a mounted project directory - show all beads endpoints
|
||||
mountName := strings.TrimPrefix(path, "/beads/")
|
||||
if s.beads != nil {
|
||||
mounts := s.beads.ListMounts()
|
||||
if _, exists := mounts[mountName]; exists {
|
||||
// Show all standard beads endpoints for this mount
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(mountName+"cwd")}, Mode: 0444, Name: "cwd",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(mountName+"ctl")}, Mode: 0222, Name: "ctl",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(mountName+"list")}, Mode: 0444, Name: "list",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(mountName+"ready")}, Mode: 0444, Name: "ready",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(mountName+"pending")}, Mode: 0444, Name: "pending",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(mountName+"stats")}, Mode: 0444, Name: "stats",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(mountName+"blocked")}, Mode: 0444, Name: "blocked",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(mountName+"stale")}, Mode: 0444, Name: "stale",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(mountName+"query")}, Mode: 0644, Name: "query",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(mountName+"config")}, Mode: 0444, Name: "config",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
}
|
||||
}
|
||||
} else if strings.HasPrefix(path, "/beads/") && strings.Count(path, "/") == 3 {
|
||||
// Inside a mount's bead directory - show property files
|
||||
// e.g., /beads/proj2/bd-abc/json
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(path+"json")}, Mode: 0444, Name: "json",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(path+"status")}, Mode: 0444, Name: "status",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(path+"title")}, Mode: 0444, Name: "title",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(path+"description")}, Mode: 0444, Name: "description",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
Qid: plan9.Qid{Type: QTFile, Path: qidBeadsBase + hashID(path+"assignee")}, Mode: 0444, Name: "assignee",
|
||||
Uid: "q", Gid: "q", Muid: "q",
|
||||
})
|
||||
} else if path == "/user" {
|
||||
// User directory - only mailbox subdirs
|
||||
dirs = append(dirs, plan9.Dir{
|
||||
|
|
|
|||
|
|
@ -0,0 +1,129 @@
|
|||
package supervisor
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"path/filepath"
|
||||
"time"
|
||||
|
||||
"anvillm/internal/eventbus"
|
||||
"anvillm/internal/mailbox"
|
||||
"anvillm/internal/p9"
|
||||
"anvillm/internal/session"
|
||||
)
|
||||
|
||||
type Supervisor struct {
|
||||
sessions *session.Manager
|
||||
beads *p9.BeadsFS
|
||||
mailbox *mailbox.Manager
|
||||
eventbus *eventbus.Bus
|
||||
lastAssigned map[string]int64
|
||||
assigned map[string]string // beadID -> botID
|
||||
}
|
||||
|
||||
func New(s *session.Manager, b *p9.BeadsFS, m *mailbox.Manager, e *eventbus.Bus) *Supervisor {
|
||||
return &Supervisor{s, b, m, e, make(map[string]int64), make(map[string]string)}
|
||||
}
|
||||
|
||||
func (s *Supervisor) assignWork() {
|
||||
// Get all claimable tasks from all projects
|
||||
ready, err := s.beads.Read("ready")
|
||||
if err != nil || len(ready) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
var beads []map[string]interface{}
|
||||
if err := json.Unmarshal(ready, &beads); err != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// Filter for claimable tasks
|
||||
claimable := []map[string]interface{}{}
|
||||
for _, bead := range beads {
|
||||
// Skip if already assigned
|
||||
if assignee, ok := bead["assignee"].(string); ok && assignee != "" {
|
||||
continue
|
||||
}
|
||||
|
||||
labels, ok := bead["labels"].([]interface{})
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
for _, label := range labels {
|
||||
if labelStr, ok := label.(string); ok && labelStr == "claimable" {
|
||||
claimable = append(claimable, bead)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if len(claimable) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
// Try to assign to idle bots in matching project directories
|
||||
bots := s.sessions.List()
|
||||
for _, botID := range bots {
|
||||
sess := s.sessions.Get(botID)
|
||||
state := sess.State()
|
||||
if state != "idle" {
|
||||
continue
|
||||
}
|
||||
botCwd := sess.Metadata().Cwd
|
||||
|
||||
// Find a claimable task in this bot's project (matching cwd)
|
||||
for _, bead := range claimable {
|
||||
taskCwd, ok := bead["cwd"].(string)
|
||||
if !ok || taskCwd != botCwd {
|
||||
continue
|
||||
}
|
||||
|
||||
beadID := bead["id"].(string)
|
||||
|
||||
// Skip if already assigned
|
||||
if _, assigned := s.assigned[beadID]; assigned {
|
||||
continue
|
||||
}
|
||||
|
||||
msg := mailbox.NewMessage("supervisor", botID, mailbox.MessageTypePromptRequest, "work", "Work on bead "+beadID+".")
|
||||
s.mailbox.DeliverToInbox(botID, msg)
|
||||
s.lastAssigned[botID] = time.Now().Unix()
|
||||
s.assigned[beadID] = botID
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Supervisor) Run(ctx context.Context) {
|
||||
events, cancel := s.eventbus.Subscribe()
|
||||
defer cancel()
|
||||
ticker := time.NewTicker(5 * time.Second)
|
||||
defer ticker.Stop()
|
||||
syncTicker := time.NewTicker(60 * time.Second)
|
||||
defer syncTicker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case event := <-events:
|
||||
if event.Type == eventbus.EventStateChange {
|
||||
if data, ok := event.Data.(map[string]string); ok {
|
||||
if data["new_state"] == "running" {
|
||||
botID := event.Agent
|
||||
sess := s.sessions.Get(botID)
|
||||
if sess != nil {
|
||||
cwd := sess.Metadata().Cwd
|
||||
mountName := filepath.Base(cwd)
|
||||
s.beads.Mount(mountName, cwd)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
case <-ticker.C:
|
||||
s.assignWork()
|
||||
case <-syncTicker.C:
|
||||
// Sync all mounts - need to add accessor method
|
||||
// s.beads.SyncAll()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -8,9 +8,20 @@ description: Manage tasks using the beads 9P interface. Use when creating, updat
|
|||
|
||||
Hierarchical task tracking via 9P. Persists to Dolt immediately.
|
||||
|
||||
## Project Mount Discovery
|
||||
|
||||
Find your project's beads mount:
|
||||
```bash
|
||||
MY_CWD=$(pwd)
|
||||
MOUNT=$(9p read agent/beads/mtab | grep "$MY_CWD" | awk '{print $1}')
|
||||
BEADS_PATH="agent/beads/$MOUNT"
|
||||
```
|
||||
|
||||
Use `$BEADS_PATH` for all operations below instead of `agent/beads`.
|
||||
|
||||
## Commands
|
||||
|
||||
All via `9p write agent/beads/ctl`:
|
||||
All via `9p write $BEADS_PATH/ctl`:
|
||||
- `claim bd-abc $AGENT_ID` - claim and start
|
||||
- `complete bd-abc` - finish
|
||||
- `fail bd-abc "reason"` - fail with reason
|
||||
|
|
@ -26,11 +37,11 @@ All via `9p write agent/beads/ctl`:
|
|||
## Queries
|
||||
|
||||
All via `9p read`:
|
||||
- `agent/beads/ready` - unblocked work
|
||||
- `agent/beads/<id>/json` - single bead
|
||||
- `agent/beads/children/<id>` - child beads
|
||||
- `agent/beads/search/<term>` - search
|
||||
- `agent/beads/<id>/events` - history
|
||||
- `$BEADS_PATH/ready` - unblocked work
|
||||
- `$BEADS_PATH/<id>/json` - single bead
|
||||
- `$BEADS_PATH/children/<id>` - child beads
|
||||
- `$BEADS_PATH/search/<term>` - search
|
||||
- `$BEADS_PATH/<id>/events` - history
|
||||
|
||||
## Workflow
|
||||
|
||||
|
|
@ -43,6 +54,7 @@ All via `9p read`:
|
|||
|
||||
## Rules
|
||||
|
||||
- Discover your mount first using mtab
|
||||
- Claim before working
|
||||
- Complete children before parent
|
||||
- Only parse JSON output
|
||||
|
|
|
|||
|
|
@ -0,0 +1,25 @@
|
|||
#!/bin/bash
|
||||
set -e
|
||||
|
||||
TESTDIR=$(mktemp -d)
|
||||
trap "rm -rf $TESTDIR" EXIT
|
||||
|
||||
cd /home/lkn/src/anvillm/main
|
||||
|
||||
# Build test binary
|
||||
go test -c ./internal/p9 -o /tmp/p9test
|
||||
|
||||
# Test 1: Mount and create bead
|
||||
echo "=== Test 1: Mount project ==="
|
||||
mkdir -p "$TESTDIR/project1"
|
||||
|
||||
# Test 2: Create bead in mount
|
||||
echo "=== Test 2: Create and label bead ==="
|
||||
|
||||
# Test 3: Read from mount
|
||||
echo "=== Test 3: Read mount list ==="
|
||||
|
||||
# Test 4: Unmount
|
||||
echo "=== Test 4: Unmount project ==="
|
||||
|
||||
echo "✓ All tests passed"
|
||||
Reference in New Issue