diff --git a/EVENTS.md b/EVENTS.md index d9ec5a7..a4c0ddb 100644 --- a/EVENTS.md +++ b/EVENTS.md @@ -21,8 +21,9 @@ This blocks and streams JSON events as they occur, one per line. Each line is a JSON object: ```json -{"id":"uuid","ts":1708598520,"agent":"a1b2c3d4","type":"StateChange","data":{"state":"running"}} -{"id":"uuid","ts":1708598525,"agent":"a1b2c3d4","type":"UserRecv","data":{"from":"user","subject":"Review request"}} +{"id":"uuid","ts":1708598520,"source":"a1b2c3d4","type":"StateChange","data":{"state":"running"}} +{"id":"uuid","ts":1708598525,"source":"a1b2c3d4","type":"UserRecv","data":{"from":"user","subject":"Review request"}} +{"id":"uuid","ts":1708598530,"source":"beads/myproject","type":"BeadReady","data":{"id":"bd-abc","title":"...","status":"open","labels":[...],"comments":[...],"mount":"myproject",...}} ``` ## Event Types @@ -32,6 +33,7 @@ Each line is a JSON object: - `UserSend` - Message sent by user - `BotRecv` - Message received by bot - `BotSend` - Message sent by bot +- `BeadReady` - A bead transitioned to open/ready; `source` is `beads/`, `data` is full bead JSON including comments ## Consuming Events @@ -49,11 +51,11 @@ Pipe to any tool: # Custom notification script 9p read anvillm/events | while read event; do - echo "$event" | jq -r '"\(.type): \(.agent)"' + echo "$event" | jq -r '"\(.type): \(.source)"' done # Desktop notifications on state changes -9p read anvillm/events | jq -r 'select(.type == "StateChange") | "\(.agent): \(.data.state)"' | \ +9p read anvillm/events | jq -r 'select(.type == "StateChange") | "\(.source): \(.data.state)"' | \ while read msg; do notify-send "AnviLLM" "$msg"; done ``` diff --git a/internal/eventbus/bus.go b/internal/eventbus/bus.go index 7d8eb13..3030edb 100644 --- a/internal/eventbus/bus.go +++ b/internal/eventbus/bus.go @@ -29,7 +29,7 @@ const allTopic = "events" type Event struct { ID string `json:"id"` TS int64 `json:"ts"` - Agent string `json:"agent"` + Source string `json:"source"` Type string `json:"type"` Data any `json:"data"` } @@ -49,11 +49,11 @@ func New() *Bus { // It is non-blocking; slow subscribers will have events dropped. func (b *Bus) Publish(agent, eventType string, data any) { e := &Event{ - ID: uuid.New().String(), - TS: time.Now().Unix(), - Agent: agent, - Type: eventType, - Data: data, + ID: uuid.New().String(), + TS: time.Now().Unix(), + Source: agent, + Type: eventType, + Data: data, } b.bus.Publish(allTopic, e) } diff --git a/internal/maildir/writer.go b/internal/maildir/writer.go index 22ca1d3..1617a22 100644 --- a/internal/maildir/writer.go +++ b/internal/maildir/writer.go @@ -37,7 +37,7 @@ func (w *Writer) run(ch <-chan *eventbus.Event) { default: continue } - w.append(ev.Agent, suffix, ev) + w.append(ev.Source, suffix, ev) } } diff --git a/internal/p9/beads.go b/internal/p9/beads.go index bde9ae3..09785b6 100644 --- a/internal/p9/beads.go +++ b/internal/p9/beads.go @@ -94,11 +94,29 @@ 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}) +// publishBeadReady emits EventBeadReady with the full bead and its comments. +// source is "beads/" so listeners can filter by mount without a round trip. +func (b *BeadsFS) publishBeadReady(store bd.Storage, mountName, beadID string) { + if b.eventbus == nil { + return } + issue, err := store.GetIssue(b.ctx, beadID) + if err != nil || issue == nil { + b.eventbus.Publish("beads/"+mountName, eventbus.EventBeadReady, map[string]string{"bead_id": beadID}) + return + } + comments, _ := store.GetIssueComments(b.ctx, beadID) + type beadReadyPayload struct { + *bd.Issue + Mount string `json:"mount"` + Comments interface{} `json:"comments,omitempty"` + } + payload := beadReadyPayload{ + Issue: issue, + Mount: mountName, + Comments: comments, + } + b.eventbus.Publish("beads/"+mountName, eventbus.EventBeadReady, payload) } // Close is a no-op since the store is managed externally. @@ -336,7 +354,7 @@ func (b *BeadsFS) Write(path string, data []byte, sessionID string) error { 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 b.executeCtlOnStore(m.store, m.name, string(data)) } return fmt.Errorf("unsupported mount write endpoint: %s", endpoint) } @@ -678,7 +696,7 @@ func (b *BeadsFS) getBlockersFromStore(store bd.Storage, id string) ([]string, e } -func (b *BeadsFS) executeCtlOnStore(store bd.Storage, cmd string) error { +func (b *BeadsFS) executeCtlOnStore(store bd.Storage, mountName, cmd string) error { command, args, err := parseCtlCommand(cmd) if err != nil { return err @@ -921,7 +939,7 @@ func (b *BeadsFS) executeCtlOnStore(store bd.Storage, cmd string) error { if err := store.UpdateIssue(b.ctx, args[0], updates, actor); err != nil { return err } - b.publishBeadReady(args[0]) + b.publishBeadReady(store, mountName, args[0]) return nil case "open": @@ -933,7 +951,7 @@ func (b *BeadsFS) executeCtlOnStore(store bd.Storage, cmd string) error { if err := store.UpdateIssue(b.ctx, args[0], map[string]any{"status": bd.StatusOpen}, actor); err != nil { return err } - b.publishBeadReady(args[0]) + b.publishBeadReady(store, mountName, args[0]) return nil case "defer": @@ -965,7 +983,7 @@ func (b *BeadsFS) executeCtlOnStore(store bd.Storage, cmd string) error { if err := store.UpdateIssue(b.ctx, args[0], updates, actor); err != nil { return err } - b.publishBeadReady(args[0]) + b.publishBeadReady(store, mountName, args[0]) return nil case "relate": diff --git a/scripts/anvillm-inbox-notify b/scripts/anvillm-inbox-notify index 0d50348..19b6d50 100755 --- a/scripts/anvillm-inbox-notify +++ b/scripts/anvillm-inbox-notify @@ -125,7 +125,7 @@ send_notification() { # addressed to the human user whose message type is in the notify_on list. # # Full event shape: -# {"id":"...", "ts":..., "agent":"user", "type":"UserRecv", +# {"id":"...", "ts":..., "source":"user", "type":"UserRecv", # "data":{"id":"...", "from":"agent-id", "to":"user", # "type":"APPROVAL_REQUEST", "subject":"...", "body":"..."}} diff --git a/scripts/anvillm-notify b/scripts/anvillm-notify index 4c9e26b..746cda7 100755 --- a/scripts/anvillm-notify +++ b/scripts/anvillm-notify @@ -21,7 +21,7 @@ get_summary() { StateChange) old_state=$(echo "$line" | jq -r '.data.old_state // empty') new_state=$(echo "$line" | jq -r '.data.new_state // empty') - agent=$(echo "$line" | jq -r '.agent // empty') + agent=$(echo "$line" | jq -r '.source // empty') if [ "$old_state" = "running" ] && [ "$new_state" = "idle" ]; then name=$(get_alias "$agent")