johno/beads
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: emit rich mutation events for status changes (bd-313v)

Browse Source 
- Add emitRichMutation() function for events with metadata
- handleClose now emits MutationStatus with old/new status
- handleUpdate detects status changes and emits MutationStatus
- Add comprehensive tests for rich mutation events

Also:
- Add activity.go test coverage (bd-3jcw):
  - Tests for parseDurationString, filterEvents, formatEvent
  - Tests for all mutation type displays
- Fix silent error handling in --follow mode (bd-csnr):
  - Track consecutive daemon failures
  - Show warning after 5 failures (rate-limited to 30s)
  - Show reconnection message on recovery

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Steve Yegge
2025-12-23 04:16:26 -08:00
parent 3f51ca5454
commit da809061db
5 changed files with 662 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
cmd/bd/activity.go
View File

@@ -167,6 +167,11 @@ func runActivityFollow(sinceTime time.Time) {
ticker := time.NewTicker(activityInterval)
defer ticker.Stop()
// Track consecutive failures for error reporting
consecutiveFailures := 0
const failureWarningThreshold = 5
lastWarningTime := time.Time{}
for {
select {
case <-rootCtx.Done():
@@ -174,10 +179,39 @@ func runActivityFollow(sinceTime time.Time) {
case <-ticker.C:
newEvents, err := fetchMutations(lastPoll)
if err != nil {
// Daemon might be down, continue trying
consecutiveFailures++
// Show warning after threshold failures, but not more than once per 30 seconds
if consecutiveFailures >= failureWarningThreshold {
if time.Since(lastWarningTime) >= 30*time.Second {
if jsonOutput {
// Emit error event in JSON mode
errorEvent := map[string]interface{}{
"type": "error",
"message": fmt.Sprintf("daemon unreachable (%d failures)", consecutiveFailures),
"timestamp": time.Now().Format(time.RFC3339),
}
data, _ := json.Marshal(errorEvent)
fmt.Fprintln(os.Stderr, string(data))
} else {
timestamp := time.Now().Format("15:04:05")
fmt.Fprintf(os.Stderr, "[%s] %s daemon unreachable (%d consecutive failures)\n",
timestamp, ui.RenderWarn("!"), consecutiveFailures)
}
lastWarningTime = time.Now()
}
}
continue
}
// Reset failure counter on success
if consecutiveFailures > 0 {
if consecutiveFailures >= failureWarningThreshold && !jsonOutput {
timestamp := time.Now().Format("15:04:05")
fmt.Fprintf(os.Stderr, "[%s] %s daemon reconnected\n", timestamp, ui.RenderPass("✓"))
}
consecutiveFailures = 0
}
newEvents = filterEvents(newEvents)
for _, e := range newEvents {
if jsonOutput {

305
cmd/bd/activity_test.go Normal file
View File

@@ -0,0 +1,305 @@
package main
import (
"testing"
"time"
"github.com/steveyegge/beads/internal/rpc"
)
// TestParseDurationString tests the duration parsing function
func TestParseDurationString(t *testing.T) {
tests := []struct {
name string
input string
expected time.Duration
expectErr bool
}{
// Standard Go duration formats
{"5 minutes", "5m", 5 * time.Minute, false},
{"1 hour", "1h", time.Hour, false},
{"30 seconds", "30s", 30 * time.Second, false},
{"2h30m", "2h30m", 2*time.Hour + 30*time.Minute, false},
// Custom day format
{"2 days", "2d", 2 * 24 * time.Hour, false},
{"1 day", "1d", 24 * time.Hour, false},
{"7 days", "7d", 7 * 24 * time.Hour, false},
// Case insensitivity for custom formats
{"uppercase D", "3D", 3 * 24 * time.Hour, false},
{"uppercase H", "2H", 2 * time.Hour, false},
{"uppercase M", "15M", 15 * time.Minute, false},
{"uppercase S", "45S", 45 * time.Second, false},
// Invalid formats
{"empty string", "", 0, true},
{"invalid unit", "5x", 0, true},
{"no number", "m", 0, true},
{"text only", "five minutes", 0, true},
// Note: negative durations are actually valid in Go's time.ParseDuration
// so we test it works rather than expecting an error
{"negative duration", "-5m", -5 * time.Minute, false},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
result, err := parseDurationString(tt.input)
if tt.expectErr {
if err == nil {
t.Errorf("expected error for input %q, got nil", tt.input)
}
} else {
if err != nil {
t.Errorf("unexpected error for input %q: %v", tt.input, err)
}
if result != tt.expected {
t.Errorf("for input %q: expected %v, got %v", tt.input, tt.expected, result)
}
}
})
}
}
// TestFilterEvents tests event filtering by --mol and --type
func TestFilterEvents(t *testing.T) {
// Create test events
events := []rpc.MutationEvent{
{Type: rpc.MutationCreate, IssueID: "bd-abc123", Timestamp: time.Now()},
{Type: rpc.MutationUpdate, IssueID: "bd-abc456", Timestamp: time.Now()},
{Type: rpc.MutationDelete, IssueID: "bd-xyz789", Timestamp: time.Now()},
{Type: rpc.MutationStatus, IssueID: "bd-abc789", Timestamp: time.Now()},
{Type: rpc.MutationComment, IssueID: "bd-def123", Timestamp: time.Now()},
}
// Reset global filter vars before each test
defer func() {
activityMol = ""
activityType = ""
}()
t.Run("no filters returns all events", func(t *testing.T) {
activityMol = ""
activityType = ""
result := filterEvents(events)
if len(result) != len(events) {
t.Errorf("expected %d events, got %d", len(events), len(result))
}
})
t.Run("filter by mol prefix", func(t *testing.T) {
activityMol = "bd-abc"
activityType = ""
result := filterEvents(events)
// Should match: bd-abc123, bd-abc456, bd-abc789
if len(result) != 3 {
t.Errorf("expected 3 events matching bd-abc*, got %d", len(result))
}
for _, e := range result {
if e.IssueID[:6] != "bd-abc" {
t.Errorf("unexpected event with ID %s", e.IssueID)
}
}
})
t.Run("filter by event type", func(t *testing.T) {
activityMol = ""
activityType = rpc.MutationCreate
result := filterEvents(events)
if len(result) != 1 {
t.Errorf("expected 1 create event, got %d", len(result))
}
if len(result) > 0 && result[0].Type != rpc.MutationCreate {
t.Errorf("expected create event, got %s", result[0].Type)
}
})
t.Run("filter by both mol and type", func(t *testing.T) {
activityMol = "bd-abc"
activityType = rpc.MutationUpdate
result := filterEvents(events)
// Should match: only bd-abc456 (update)
if len(result) != 1 {
t.Errorf("expected 1 event, got %d", len(result))
}
if len(result) > 0 {
if result[0].IssueID != "bd-abc456" {
t.Errorf("expected bd-abc456, got %s", result[0].IssueID)
}
if result[0].Type != rpc.MutationUpdate {
t.Errorf("expected update type, got %s", result[0].Type)
}
}
})
t.Run("filter returns empty for no matches", func(t *testing.T) {
activityMol = "bd-nomatch"
activityType = ""
result := filterEvents(events)
if len(result) != 0 {
t.Errorf("expected 0 events, got %d", len(result))
}
})
}
// TestGetEventDisplay tests the symbol and message generation for all event types
func TestGetEventDisplay(t *testing.T) {
tests := []struct {
name string
event rpc.MutationEvent
expectedSymbol string
checkMessage func(string) bool
}{
{
name: "create event",
event: rpc.MutationEvent{Type: rpc.MutationCreate, IssueID: "bd-123"},
expectedSymbol: "+",
checkMessage: func(m string) bool { return m == "bd-123 created" },
},
{
name: "update event",
event: rpc.MutationEvent{Type: rpc.MutationUpdate, IssueID: "bd-456"},
expectedSymbol: "\u2192", // →
checkMessage: func(m string) bool { return m == "bd-456 updated" },
},
{
name: "delete event",
event: rpc.MutationEvent{Type: rpc.MutationDelete, IssueID: "bd-789"},
expectedSymbol: "\u2298", // ⊘
checkMessage: func(m string) bool { return m == "bd-789 deleted" },
},
{
name: "comment event",
event: rpc.MutationEvent{Type: rpc.MutationComment, IssueID: "bd-abc"},
expectedSymbol: "\U0001F4AC", // 💬
checkMessage: func(m string) bool { return m == "bd-abc comment added" },
},
{
name: "bonded event with step count",
event: rpc.MutationEvent{
Type: rpc.MutationBonded,
IssueID: "bd-mol",
StepCount: 5,
},
expectedSymbol: "+",
checkMessage: func(m string) bool { return m == "bd-mol bonded (5 steps)" },
},
{
name: "bonded event without step count",
event: rpc.MutationEvent{Type: rpc.MutationBonded, IssueID: "bd-mol2"},
expectedSymbol: "+",
checkMessage: func(m string) bool { return m == "bd-mol2 bonded" },
},
{
name: "squashed event",
event: rpc.MutationEvent{Type: rpc.MutationSquashed, IssueID: "bd-wisp"},
expectedSymbol: "\u25C9", // ◉
checkMessage: func(m string) bool { return m == "bd-wisp SQUASHED" },
},
{
name: "burned event",
event: rpc.MutationEvent{Type: rpc.MutationBurned, IssueID: "bd-burn"},
expectedSymbol: "\U0001F525", // 🔥
checkMessage: func(m string) bool { return m == "bd-burn burned" },
},
{
name: "status event - in_progress",
event: rpc.MutationEvent{
Type: rpc.MutationStatus,
IssueID: "bd-wip",
OldStatus: "open",
NewStatus: "in_progress",
},
expectedSymbol: "\u2192", // →
checkMessage: func(m string) bool { return m == "bd-wip in_progress" },
},
{
name: "status event - closed",
event: rpc.MutationEvent{
Type: rpc.MutationStatus,
IssueID: "bd-done",
OldStatus: "in_progress",
NewStatus: "closed",
},
expectedSymbol: "\u2713", // ✓
checkMessage: func(m string) bool { return m == "bd-done completed" },
},
{
name: "status event - reopened",
event: rpc.MutationEvent{
Type: rpc.MutationStatus,
IssueID: "bd-reopen",
OldStatus: "closed",
NewStatus: "open",
},
expectedSymbol: "\u21BA", // ↺
checkMessage: func(m string) bool { return m == "bd-reopen reopened" },
},
{
name: "unknown event type",
event: rpc.MutationEvent{Type: "custom", IssueID: "bd-custom"},
expectedSymbol: "\u2022", // •
checkMessage: func(m string) bool { return m == "bd-custom custom" },
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
symbol, message := getEventDisplay(tt.event)
if symbol != tt.expectedSymbol {
t.Errorf("expected symbol %q, got %q", tt.expectedSymbol, symbol)
}
if !tt.checkMessage(message) {
t.Errorf("unexpected message: %q", message)
}
})
}
}
// TestFormatEvent tests the ActivityEvent formatting
func TestFormatEvent(t *testing.T) {
now := time.Now()
event := rpc.MutationEvent{
Type: rpc.MutationStatus,
IssueID: "bd-test",
Timestamp: now,
OldStatus: "open",
NewStatus: "in_progress",
ParentID: "bd-parent",
StepCount: 3,
}
result := formatEvent(event)
if result.Timestamp != now {
t.Errorf("expected timestamp %v, got %v", now, result.Timestamp)
}
if result.Type != rpc.MutationStatus {
t.Errorf("expected type %s, got %s", rpc.MutationStatus, result.Type)
}
if result.IssueID != "bd-test" {
t.Errorf("expected issue ID bd-test, got %s", result.IssueID)
}
if result.OldStatus != "open" {
t.Errorf("expected OldStatus 'open', got %s", result.OldStatus)
}
if result.NewStatus != "in_progress" {
t.Errorf("expected NewStatus 'in_progress', got %s", result.NewStatus)
}
if result.ParentID != "bd-parent" {
t.Errorf("expected ParentID 'bd-parent', got %s", result.ParentID)
}
if result.StepCount != 3 {
t.Errorf("expected StepCount 3, got %d", result.StepCount)
}
if result.Symbol == "" {
t.Error("expected non-empty symbol")
}
if result.Message == "" {
t.Error("expected non-empty message")
}
}

17
internal/rpc/server_core.go
View File

@@ -139,10 +139,19 @@ func NewServer(socketPath string, store storage.Storage, workspacePath string, d
// Non-blocking: drops event if channel is full (sync will happen eventually).
// Also stores in recent mutations buffer for polling.
func (s *Server) emitMutation(eventType, issueID string) {
event := MutationEvent{
Type: eventType,
IssueID: issueID,
Timestamp: time.Now(),
s.emitRichMutation(MutationEvent{
Type: eventType,
IssueID: issueID,
})
}
// emitRichMutation sends a pre-built mutation event with optional metadata.
// Use this for events that include additional context (status changes, bonded events, etc.)
// Non-blocking: drops event if channel is full (sync will happen eventually).
func (s *Server) emitRichMutation(event MutationEvent) {
// Always set timestamp if not provided
if event.Timestamp.IsZero() {
event.Timestamp = time.Now()
}
// Send to mutation channel for daemon

27
internal/rpc/server_issues_epics.go
View File

@@ -465,7 +465,17 @@ func (s *Server) handleUpdate(req *Request) Response {
// Emit mutation event for event-driven daemon (only if any updates or label operations were performed)
if len(updates) > 0 || len(updateArgs.SetLabels) > 0 || len(updateArgs.AddLabels) > 0 || len(updateArgs.RemoveLabels) > 0 {
s.emitMutation(MutationUpdate, updateArgs.ID)
// Check if this was a status change - emit rich MutationStatus event
if updateArgs.Status != nil && *updateArgs.Status != string(issue.Status) {
s.emitRichMutation(MutationEvent{
Type: MutationStatus,
IssueID: updateArgs.ID,
OldStatus: string(issue.Status),
NewStatus: *updateArgs.Status,
})
} else {
s.emitMutation(MutationUpdate, updateArgs.ID)
}
}
updatedIssue, getErr := store.GetIssue(ctx, updateArgs.ID)
@@ -517,6 +527,12 @@ func (s *Server) handleClose(req *Request) Response {
}
}
// Capture old status for rich mutation event
oldStatus := ""
if issue != nil {
oldStatus = string(issue.Status)
}
if err := store.CloseIssue(ctx, closeArgs.ID, closeArgs.Reason, s.reqActor(req)); err != nil {
return Response{
Success: false,
@@ -524,8 +540,13 @@ func (s *Server) handleClose(req *Request) Response {
}
}
// Emit mutation event for event-driven daemon
s.emitMutation(MutationUpdate, closeArgs.ID)
// Emit rich status change event for event-driven daemon
s.emitRichMutation(MutationEvent{
Type: MutationStatus,
IssueID: closeArgs.ID,
OldStatus: oldStatus,
NewStatus: "closed",
})
closedIssue, _ := store.GetIssue(ctx, closeArgs.ID)
data, _ := json.Marshal(closedIssue)

285
internal/rpc/server_mutations_test.go
View File

@@ -234,6 +234,72 @@ func TestMutationEventTypes(t *testing.T) {
}
}
// TestEmitRichMutation verifies that rich mutation events include metadata fields
func TestEmitRichMutation(t *testing.T) {
store := memory.New("/tmp/test.jsonl")
server := NewServer("/tmp/test.sock", store, "/tmp", "/tmp/test.db")
// Emit a rich status change event
server.emitRichMutation(MutationEvent{
Type: MutationStatus,
IssueID: "bd-456",
OldStatus: "open",
NewStatus: "in_progress",
})
mutations := server.GetRecentMutations(0)
if len(mutations) != 1 {
t.Fatalf("expected 1 mutation, got %d", len(mutations))
}
m := mutations[0]
if m.Type != MutationStatus {
t.Errorf("expected type %s, got %s", MutationStatus, m.Type)
}
if m.IssueID != "bd-456" {
t.Errorf("expected issue ID bd-456, got %s", m.IssueID)
}
if m.OldStatus != "open" {
t.Errorf("expected OldStatus 'open', got %s", m.OldStatus)
}
if m.NewStatus != "in_progress" {
t.Errorf("expected NewStatus 'in_progress', got %s", m.NewStatus)
}
if m.Timestamp.IsZero() {
t.Error("expected Timestamp to be set automatically")
}
}
// TestEmitRichMutation_Bonded verifies bonded events include step count
func TestEmitRichMutation_Bonded(t *testing.T) {
store := memory.New("/tmp/test.jsonl")
server := NewServer("/tmp/test.sock", store, "/tmp", "/tmp/test.db")
// Emit a bonded event with metadata
server.emitRichMutation(MutationEvent{
Type: MutationBonded,
IssueID: "bd-789",
ParentID: "bd-parent",
StepCount: 5,
})
mutations := server.GetRecentMutations(0)
if len(mutations) != 1 {
t.Fatalf("expected 1 mutation, got %d", len(mutations))
}
m := mutations[0]
if m.Type != MutationBonded {
t.Errorf("expected type %s, got %s", MutationBonded, m.Type)
}
if m.ParentID != "bd-parent" {
t.Errorf("expected ParentID 'bd-parent', got %s", m.ParentID)
}
if m.StepCount != 5 {
t.Errorf("expected StepCount 5, got %d", m.StepCount)
}
}
func TestMutationTimestamps(t *testing.T) {
store := memory.New("/tmp/test.jsonl")
server := NewServer("/tmp/test.sock", store, "/tmp", "/tmp/test.db")
@@ -276,6 +342,225 @@ func TestEmitMutation_NonBlocking(t *testing.T) {
}
}
// TestHandleClose_EmitsStatusMutation verifies that close operations emit MutationStatus events
// with old/new status metadata (bd-313v fix)
func TestHandleClose_EmitsStatusMutation(t *testing.T) {
store := memory.New("/tmp/test.jsonl")
server := NewServer("/tmp/test.sock", store, "/tmp", "/tmp/test.db")
// Create an issue first
createArgs := CreateArgs{
Title: "Test Issue for Close",
IssueType: "bug",
Priority: 1,
}
createJSON, _ := json.Marshal(createArgs)
createReq := &Request{
Operation: OpCreate,
Args: createJSON,
Actor: "test-user",
}
createResp := server.handleCreate(createReq)
if !createResp.Success {
t.Fatalf("failed to create test issue: %s", createResp.Error)
}
var createdIssue map[string]interface{}
if err := json.Unmarshal(createResp.Data, &createdIssue); err != nil {
t.Fatalf("failed to parse created issue: %v", err)
}
issueID := createdIssue["id"].(string)
// Clear mutation buffer
time.Sleep(10 * time.Millisecond)
checkpoint := time.Now().UnixMilli()
time.Sleep(10 * time.Millisecond)
// Close the issue
closeArgs := CloseArgs{
ID: issueID,
Reason: "test complete",
}
closeJSON, _ := json.Marshal(closeArgs)
closeReq := &Request{
Operation: OpClose,
Args: closeJSON,
Actor: "test-user",
}
closeResp := server.handleClose(closeReq)
if !closeResp.Success {
t.Fatalf("close operation failed: %s", closeResp.Error)
}
// Verify MutationStatus event was emitted with correct metadata
mutations := server.GetRecentMutations(checkpoint)
var statusMutation *MutationEvent
for _, m := range mutations {
if m.Type == MutationStatus && m.IssueID == issueID {
statusMutation = &m
break
}
}
if statusMutation == nil {
t.Fatalf("expected MutationStatus event for issue %s, but none found in mutations: %+v", issueID, mutations)
}
if statusMutation.OldStatus != "open" {
t.Errorf("expected OldStatus 'open', got %s", statusMutation.OldStatus)
}
if statusMutation.NewStatus != "closed" {
t.Errorf("expected NewStatus 'closed', got %s", statusMutation.NewStatus)
}
}
// TestHandleUpdate_EmitsStatusMutationOnStatusChange verifies that status updates emit MutationStatus
func TestHandleUpdate_EmitsStatusMutationOnStatusChange(t *testing.T) {
store := memory.New("/tmp/test.jsonl")
server := NewServer("/tmp/test.sock", store, "/tmp", "/tmp/test.db")
// Create an issue first
createArgs := CreateArgs{
Title: "Test Issue for Status Update",
IssueType: "task",
Priority: 2,
}
createJSON, _ := json.Marshal(createArgs)
createReq := &Request{
Operation: OpCreate,
Args: createJSON,
Actor: "test-user",
}
createResp := server.handleCreate(createReq)
if !createResp.Success {
t.Fatalf("failed to create test issue: %s", createResp.Error)
}
var createdIssue map[string]interface{}
if err := json.Unmarshal(createResp.Data, &createdIssue); err != nil {
t.Fatalf("failed to parse created issue: %v", err)
}
issueID := createdIssue["id"].(string)
// Clear mutation buffer
time.Sleep(10 * time.Millisecond)
checkpoint := time.Now().UnixMilli()
time.Sleep(10 * time.Millisecond)
// Update status to in_progress
status := "in_progress"
updateArgs := UpdateArgs{
ID: issueID,
Status: &status,
}
updateJSON, _ := json.Marshal(updateArgs)
updateReq := &Request{
Operation: OpUpdate,
Args: updateJSON,
Actor: "test-user",
}
updateResp := server.handleUpdate(updateReq)
if !updateResp.Success {
t.Fatalf("update operation failed: %s", updateResp.Error)
}
// Verify MutationStatus event was emitted
mutations := server.GetRecentMutations(checkpoint)
var statusMutation *MutationEvent
for _, m := range mutations {
if m.Type == MutationStatus && m.IssueID == issueID {
statusMutation = &m
break
}
}
if statusMutation == nil {
t.Fatalf("expected MutationStatus event, but none found in mutations: %+v", mutations)
}
if statusMutation.OldStatus != "open" {
t.Errorf("expected OldStatus 'open', got %s", statusMutation.OldStatus)
}
if statusMutation.NewStatus != "in_progress" {
t.Errorf("expected NewStatus 'in_progress', got %s", statusMutation.NewStatus)
}
}
// TestHandleUpdate_EmitsUpdateMutationForNonStatusChanges verifies non-status updates emit MutationUpdate
func TestHandleUpdate_EmitsUpdateMutationForNonStatusChanges(t *testing.T) {
store := memory.New("/tmp/test.jsonl")
server := NewServer("/tmp/test.sock", store, "/tmp", "/tmp/test.db")
// Create an issue first
createArgs := CreateArgs{
Title: "Test Issue for Non-Status Update",
IssueType: "task",
Priority: 2,
}
createJSON, _ := json.Marshal(createArgs)
createReq := &Request{
Operation: OpCreate,
Args: createJSON,
Actor: "test-user",
}
createResp := server.handleCreate(createReq)
if !createResp.Success {
t.Fatalf("failed to create test issue: %s", createResp.Error)
}
var createdIssue map[string]interface{}
if err := json.Unmarshal(createResp.Data, &createdIssue); err != nil {
t.Fatalf("failed to parse created issue: %v", err)
}
issueID := createdIssue["id"].(string)
// Clear mutation buffer
time.Sleep(10 * time.Millisecond)
checkpoint := time.Now().UnixMilli()
time.Sleep(10 * time.Millisecond)
// Update title (not status)
newTitle := "Updated Title"
updateArgs := UpdateArgs{
ID: issueID,
Title: &newTitle,
}
updateJSON, _ := json.Marshal(updateArgs)
updateReq := &Request{
Operation: OpUpdate,
Args: updateJSON,
Actor: "test-user",
}
updateResp := server.handleUpdate(updateReq)
if !updateResp.Success {
t.Fatalf("update operation failed: %s", updateResp.Error)
}
// Verify MutationUpdate event was emitted (not MutationStatus)
mutations := server.GetRecentMutations(checkpoint)
var updateMutation *MutationEvent
for _, m := range mutations {
if m.IssueID == issueID {
updateMutation = &m
break
}
}
if updateMutation == nil {
t.Fatal("expected mutation event, but none found")
}
if updateMutation.Type != MutationUpdate {
t.Errorf("expected MutationUpdate type, got %s", updateMutation.Type)
}
}
// TestHandleDelete_EmitsMutation verifies that delete operations emit mutation events
// This is a regression test for the issue where delete operations bypass the daemon
// and don't trigger auto-sync. The delete RPC handler should emit MutationDelete events.