events: rename agent→source, embed full bead in BeadReady

Event.Agent renamed to Event.Source (json: "source") — more accurate since
the field carries a session ID for session events and a subsystem identifier
for infrastructure events.

BeadReady source is now "beads/<mount>" so listeners can filter by mount
without fetching. BeadReady data is now the full bead JSON including comments,
eliminating the round trip a listener would need to decide whether to claim.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Levi Neely 2026-03-14 20:36:38 +01:00
parent cb97b3640c
commit c58f0615a9
6 changed files with 42 additions and 22 deletions

View File

@ -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/<mount>`, `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
```

View File

@ -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)
}

View File

@ -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)
}
}

View File

@ -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/<mount>" 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":

View File

@ -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":"..."}}

View File

@ -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")