Add wait-ready command with SSE-driven polling

Implements `bs wait-ready --timeout <secs>` which blocks until at least
one ready (unblocked) bead exists. Opens the SSE stream before the
initial list check to close the race window where a bead becomes ready
between check and subscription.

Supports --tag (repeatable), --assignee, --priority, --type filters.
Returns errTimeout (exit non-zero, no output) on timeout; prints to
stderr and exits non-zero on connectivity failure; exits 0 silently
when a ready bead is found.

Built with Raymond (Agent Orchestrator)

Author: vector76

internal/cli/commands_wait_ready.go

@@ -0,0 +1,126 @@
+package cli
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/spf13/cobra"
+)
+
+var errTimeout = errors.New("timed out")
+
+func newWaitReadyCmd() *cobra.Command {
+	var timeout int
+	var tags []string
+	var assignee string
+	var priority string
+	var beadType string
+
+	cmd := &cobra.Command{
+		Use:           "wait-ready",
+		Short:         "Wait until a ready bead exists",
+		SilenceErrors: true,
+		SilenceUsage:  true,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			c, err := NewClientFromEnv()
+			if err != nil {
+				fmt.Fprintln(cmd.ErrOrStderr(), "error:", err)
+				return err
+			}
+
+			var ctx context.Context
+			var cancel context.CancelFunc
+			if timeout > 0 {
+				ctx, cancel = context.WithTimeout(context.Background(), time.Duration(timeout)*time.Second)
+			} else {
+				ctx, cancel = context.WithCancel(context.Background())
+			}
+			defer cancel()
+
+			signals, errChan := c.StreamSSE(ctx)
+
+			// Build the query path once; flags are immutable after parsing.
+			params := url.Values{}
+			params.Set("ready", "true")
+			params.Set("per_page", "1")
+			if len(tags) > 0 {
+				params.Set("tag", strings.Join(tags, ","))
+			}
+			if assignee != "" {
+				params.Set("assignee", assignee)
+			}
+			if priority != "" {
+				params.Set("priority", priority)
+			}
+			if beadType != "" {
+				params.Set("type", beadType)
+			}
+			path := "/api/v1/beads?" + params.Encode()
+
+			checkReady := func() (bool, error) {
+				data, err := c.Do("GET", path, nil)
+				if err != nil {
+					return false, err
+				}
+				var result struct {
+					Total int `json:"total"`
+				}
+				if err := json.Unmarshal(data, &result); err != nil {
+					return false, fmt.Errorf("parsing response: %w", err)
+				}
+				return result.Total >= 1, nil
+			}
+
+			ready, err := checkReady()
+			if err != nil {
+				fmt.Fprintln(cmd.ErrOrStderr(), "error:", err)
+				return err
+			}
+			if ready {
+				return nil
+			}
+
+			for {
+				select {
+				case _, ok := <-signals:
+					if !ok {
+						// signals closed; errChan will deliver the outcome — stop selecting on signals.
+						signals = nil
+						continue
+					}
+					ready, err := checkReady()
+					if err != nil {
+						fmt.Fprintln(cmd.ErrOrStderr(), "error:", err)
+						return err
+					}
+					if ready {
+						return nil
+					}
+				case err, ok := <-errChan:
+					if ok && err != nil {
+						fmt.Fprintln(cmd.ErrOrStderr(), "error:", err)
+						return err
+					}
+					// nil error or channel closed: context was cancelled (deadline or explicit).
+					return errTimeout
+				case <-ctx.Done():
+					return errTimeout
+				}
+			}
+		},
+	}
+
+	cmd.Flags().IntVar(&timeout, "timeout", 0, "seconds to wait (0 = indefinite)")
+	_ = cmd.MarkFlagRequired("timeout")
+	cmd.Flags().StringArrayVar(&tags, "tag", nil, "filter by tag (repeatable)")
+	cmd.Flags().StringVar(&assignee, "assignee", "", "filter by assignee")
+	cmd.Flags().StringVar(&priority, "priority", "", "filter by priority")
+	cmd.Flags().StringVar(&beadType, "type", "", "filter by bead type")
+
+	return cmd
+}

internal/cli/commands_wait_ready_test.go

@@ -0,0 +1,122 @@
+package cli
+
+import (
+	"bytes"
+	"errors"
+	"net/http"
+	"testing"
+	"time"
+)
+
+// runWaitReadyCmd executes the wait-ready command and returns (stdout, stderr, error).
+func runWaitReadyCmd(t *testing.T, args ...string) (string, string, error) {
+	t.Helper()
+	cmd := NewRootCmd()
+	outBuf := new(bytes.Buffer)
+	errBuf := new(bytes.Buffer)
+	cmd.SetOut(outBuf)
+	cmd.SetErr(errBuf)
+	cmd.SetArgs(append([]string{"wait-ready"}, args...))
+	err := cmd.Execute()
+	return outBuf.String(), errBuf.String(), err
+}
+
+// TestWaitReady_ImmediatelyReady verifies that the command exits 0 when a ready bead
+// already exists before the SSE connection is established.
+func TestWaitReady_ImmediatelyReady(t *testing.T) {
+	ts := startTestServer(t)
+	setClientEnv(t, ts.URL)
+
+	// Create a bead (no dependencies → immediately ready)
+	runCmd(t, "add", "Ready bead")
+
+	_, _, err := runWaitReadyCmd(t, "--timeout", "5")
+	if err != nil {
+		t.Fatalf("expected nil error, got %v", err)
+	}
+}
+
+// TestWaitReady_Timeout verifies that the command returns errTimeout when no ready bead
+// appears within the timeout window.
+func TestWaitReady_Timeout(t *testing.T) {
+	ts := startTestServer(t)
+	setClientEnv(t, ts.URL)
+
+	// No beads → nothing is ready.
+	_, _, err := runWaitReadyCmd(t, "--timeout", "1")
+	if err == nil {
+		t.Fatal("expected timeout error, got nil")
+	}
+	if !errors.Is(err, errTimeout) {
+		t.Fatalf("expected errTimeout, got %v", err)
+	}
+}
+
+// TestWaitReady_BecomesReadyViaSSE verifies that the command exits 0 once a bead becomes
+// ready after the initial check (simulated via a second goroutine adding a bead).
+func TestWaitReady_BecomesReadyViaSSE(t *testing.T) {
+	ts := startTestServer(t)
+	setClientEnv(t, ts.URL)
+
+	// Add a bead with a blocking dependency so it's not initially ready.
+	out := runCmd(t, "add", "Blocker bead")
+	blocker := parseBeadFromOutput(t, out)
+
+	out = runCmd(t, "add", "Blocked bead")
+	blocked := parseBeadFromOutput(t, out)
+	runCmd(t, "link", blocked.ID, "--blocked-by", blocker.ID)
+
+	// Unblock the bead after a short delay in a goroutine.
+	// Use a direct HTTP call instead of runCmd so t.Fatalf is not called from a goroutine
+	// after the test may have completed.
+	serverURL := ts.URL
+	go func() {
+		time.Sleep(200 * time.Millisecond)
+		req, _ := http.NewRequest(http.MethodPost, serverURL+"/api/v1/beads/"+blocker.ID+"/status", bytes.NewBufferString(`{"status":"closed"}`))
+		req.Header.Set("Authorization", "Bearer "+testToken)
+		req.Header.Set("Content-Type", "application/json")
+		http.DefaultClient.Do(req) //nolint:errcheck
+	}()
+
+	_, _, err := runWaitReadyCmd(t, "--timeout", "5")
+	if err != nil {
+		t.Fatalf("expected nil error after bead became ready, got %v", err)
+	}
+}
+
+// TestWaitReady_MissingTimeout verifies that --timeout is required.
+func TestWaitReady_MissingTimeout(t *testing.T) {
+	ts := startTestServer(t)
+	setClientEnv(t, ts.URL)
+
+	_, _, err := runWaitReadyCmd(t)
+	if err == nil {
+		t.Fatal("expected error when --timeout is missing")
+	}
+}
+
+// TestWaitReady_TagFilter verifies that --tag filters beads by tag.
+func TestWaitReady_TagFilter(t *testing.T) {
+	ts := startTestServer(t)
+	setClientEnv(t, ts.URL)
+
+	// Create a bead without the required tag — should NOT match.
+	runCmd(t, "add", "Untagged bead")
+
+	// With a tag filter for a tag no bead has, it should time out.
+	_, _, err := runWaitReadyCmd(t, "--timeout", "1", "--tag", "mytag")
+	if err == nil {
+		t.Fatal("expected timeout when no bead matches tag filter")
+	}
+	if !errors.Is(err, errTimeout) {
+		t.Fatalf("expected errTimeout, got %v", err)
+	}
+
+	// Now create a bead with the matching tag.
+	runCmd(t, "add", "Tagged bead", "--tags", "mytag")
+
+	_, _, err = runWaitReadyCmd(t, "--timeout", "5", "--tag", "mytag")
+	if err != nil {
+		t.Fatalf("expected nil error with matching tagged bead, got %v", err)
+	}
+}

internal/cli/root.go

@@ -82,6 +82,7 @@ func NewRootCmd() *cobra.Command {
 		newLinkCmd(),
 		newUnlinkCmd(),
 		newDepsCmd(),
+		newWaitReadyCmd(),
 	} {
 		cmd.GroupID = "client"
 		root.AddCommand(cmd)