supervisor: replace polling with event-driven work assignment
- Add EventBeadReady to eventbus, published by BeadsFS on open/reopen/unclaim
- Wire eventbus into BeadsFS via SetEventBus (called after server init)
- Supervisor now reacts to EventStateChange{idle} and EventBeadReady
instead of a 5s ticker; 60s fallback ticker retained for safety
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
d39a7f118d
commit
68cc9445a9
|
|
@ -220,8 +220,9 @@ func start(daemonize bool) {
|
|||
}
|
||||
defer srv.Close()
|
||||
|
||||
// Wire up event bus to session manager
|
||||
// Wire up event bus to session manager and beads filesystem
|
||||
mgr.SetEventBus(srv.Events())
|
||||
beadsFS.SetEventBus(srv.Events())
|
||||
|
||||
// Start maildir writer for message persistence
|
||||
mailDir := filepath.Join(os.Getenv("HOME"), ".local", "share", "anvillm", "mail")
|
||||
|
|
|
|||
|
|
@ -19,6 +19,7 @@ const (
|
|||
EventUserSend = "UserSend"
|
||||
EventBotRecv = "BotRecv"
|
||||
EventBotSend = "BotSend"
|
||||
EventBeadReady = "BeadReady" // a bead transitioned to open/ready
|
||||
)
|
||||
|
||||
// allTopic is the single topic used for all events.
|
||||
|
|
|
|||
|
|
@ -1,6 +1,7 @@
|
|||
package p9
|
||||
|
||||
import (
|
||||
"anvillm/internal/eventbus"
|
||||
"anvillm/internal/logging"
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
|
@ -86,6 +87,7 @@ type BeadsFS struct {
|
|||
queryResult []*bd.Issue
|
||||
mounts map[string]*MountedProject
|
||||
mountsMu sync.RWMutex
|
||||
eventbus *eventbus.Bus
|
||||
}
|
||||
|
||||
// NewBeadsFS creates a beads filesystem handler from an existing store.
|
||||
|
|
@ -96,6 +98,18 @@ func NewBeadsFS(store bd.Storage, ctx context.Context) *BeadsFS {
|
|||
}
|
||||
}
|
||||
|
||||
// SetEventBus wires an event bus for publishing bead state change events.
|
||||
func (b *BeadsFS) SetEventBus(eb *eventbus.Bus) {
|
||||
b.eventbus = eb
|
||||
}
|
||||
|
||||
// publishBeadReady emits EventBeadReady if an event bus is wired.
|
||||
func (b *BeadsFS) publishBeadReady(beadID string) {
|
||||
if b.eventbus != nil {
|
||||
b.eventbus.Publish("beads", eventbus.EventBeadReady, map[string]string{"bead_id": beadID})
|
||||
}
|
||||
}
|
||||
|
||||
// Close is a no-op since the store is managed externally.
|
||||
func (b *BeadsFS) Close() error {
|
||||
return nil
|
||||
|
|
@ -998,7 +1012,11 @@ func (b *BeadsFS) executeCtlOnStore(store bd.Storage, cmd string) error {
|
|||
"assignee": "",
|
||||
"status": bd.StatusOpen,
|
||||
}
|
||||
return store.UpdateIssue(b.ctx, args[0], updates, actor)
|
||||
if err := store.UpdateIssue(b.ctx, args[0], updates, actor); err != nil {
|
||||
return err
|
||||
}
|
||||
b.publishBeadReady(args[0])
|
||||
return nil
|
||||
|
||||
case "open":
|
||||
// Promote a deferred bead to open (ready for scheduling).
|
||||
|
|
@ -1006,7 +1024,11 @@ func (b *BeadsFS) executeCtlOnStore(store bd.Storage, cmd string) error {
|
|||
if len(args) < 1 {
|
||||
return fmt.Errorf("usage: open <bead-id>")
|
||||
}
|
||||
return store.UpdateIssue(b.ctx, args[0], map[string]any{"status": bd.StatusOpen}, actor)
|
||||
if err := store.UpdateIssue(b.ctx, args[0], map[string]any{"status": bd.StatusOpen}, actor); err != nil {
|
||||
return err
|
||||
}
|
||||
b.publishBeadReady(args[0])
|
||||
return nil
|
||||
|
||||
case "defer":
|
||||
// Defer a bead, optionally until a specific time (RFC3339).
|
||||
|
|
@ -1034,7 +1056,11 @@ func (b *BeadsFS) executeCtlOnStore(store bd.Storage, cmd string) error {
|
|||
"status": bd.StatusOpen,
|
||||
"closed_at": (*time.Time)(nil),
|
||||
}
|
||||
return store.UpdateIssue(b.ctx, args[0], updates, actor)
|
||||
if err := store.UpdateIssue(b.ctx, args[0], updates, actor); err != nil {
|
||||
return err
|
||||
}
|
||||
b.publishBeadReady(args[0])
|
||||
return nil
|
||||
|
||||
case "relate":
|
||||
// Add a bidirectional relates-to dependency between two beads.
|
||||
|
|
|
|||
|
|
@ -122,20 +122,29 @@ func (s *Supervisor) assignWork() {
|
|||
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()
|
||||
// Fallback ticker: catches anything missed by events (e.g. startup, edge cases).
|
||||
fallback := time.NewTicker(60 * time.Second)
|
||||
defer fallback.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-events:
|
||||
case <-ticker.C:
|
||||
case event := <-events:
|
||||
if event == nil {
|
||||
continue
|
||||
}
|
||||
switch event.Type {
|
||||
case eventbus.EventStateChange:
|
||||
// Session went idle — it may be able to take work.
|
||||
if data, ok := event.Data.(map[string]string); ok && data["new_state"] == "idle" {
|
||||
s.assignWork()
|
||||
}
|
||||
case eventbus.EventBeadReady:
|
||||
// A bead just became open/ready — try to assign it.
|
||||
s.assignWork()
|
||||
}
|
||||
case <-fallback.C:
|
||||
s.assignWork()
|
||||
case <-syncTicker.C:
|
||||
// Sync all mounts - need to add accessor method
|
||||
// s.beads.SyncAll()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Reference in New Issue