release: v0.16.1

johntheyoung · johntheyoung · commit a17939b181e1 · 2026-02-19T19:08:10.000+08:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,6 +4,16 @@ All notable changes to this project will be documented in this file.
 
 ## Unreleased
 
+## v0.16.1 - 2026-02-19
+
+### Added
+- Websocket compatibility guardrails in docs (`docs/api-notes.md`) covering reconnect semantics, control-message behavior, and polling fallback expectations.
+- Troubleshooting guidance for `rr events tail` websocket support/reconnect diagnostics (`docs/troubleshooting.md`).
+
+### Fixed
+- Added regression tests for `rr events tail` control-message filtering and reconnect/re-subscribe paths to prevent behavior drift.
+- Clarified README caveat that websocket streaming is live-only and reconnects can require follow-up list/search reconciliation.
+
 ## v0.16.0 - 2026-02-19
 
 ### Added
diff --git a/README.md b/README.md
@@ -153,6 +153,7 @@ rr events tail --all --include-control --stop-after 30s --json
 
 If your desktop build does not expose `/v1/ws`, `rr events tail` returns an explicit unsupported-version error.
 For older builds or when you need polling semantics, continue using `rr messages tail`.
+This stream is live-only; on reconnect there may be gaps, so re-query state with `rr messages list/search` when exact completeness matters.
 
 ## Messages
 
diff --git a/docs/api-notes.md b/docs/api-notes.md
@@ -51,3 +51,17 @@
 - In `--json --envelope` mode, cursor-based commands include normalized machine metadata at
   `metadata.pagination` with fields: `has_more`, `direction`, `next_cursor`, `oldest_cursor`,
   `newest_cursor`, `auto_paged`, `capped`, and `max_items` (when auto-paging is enabled).
+
+## WebSocket (Experimental) guardrails
+
+- `rr events tail` connects to `/v1/ws` and is intentionally treated as best-effort because upstream marks it experimental.
+- Unsupported websocket route behavior is explicit and stable: `rr events tail` returns an unsupported-version error if `/v1/ws` is unavailable.
+- The stream is live-only (no replay cursor). Treat reconnects as potential event gaps and re-query state with `messages list/search` when exact completeness matters.
+- `seq` is monotonic per-connection only. Do not treat it as a global durable checkpoint across reconnects.
+- Reconnect behavior is automatic by default:
+  - on disconnect/read failure, `rr events tail` reconnects and re-sends subscriptions;
+  - disable with `--reconnect=false` for strict single-connection behavior.
+- Control-message handling is explicit:
+  - default output suppresses `ready`, `subscriptions.updated`, and websocket `error` control events;
+  - pass `--include-control` to include control events in output streams.
+- If websocket semantics are not desired (or unavailable), use `rr messages tail` polling mode as a stable fallback.
diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md
@@ -30,6 +30,13 @@ rr auth set --stdin
   your Beeper Desktop build is older than the required Desktop API routes.
 - Update Beeper Desktop and retry `rr doctor`, then re-run the command.
 
+## `events tail` websocket issues
+
+- If you see `websocket events are not supported`, your Beeper Desktop build does not expose `/v1/ws` yet. Use `rr messages tail` or upgrade Desktop.
+- If your stream reconnects repeatedly, verify local connectivity and auth first (`rr auth status --check`, `rr doctor`).
+- `rr events tail` reconnects by default; disable with `--reconnect=false` if you want failures surfaced immediately.
+- For deterministic CI/script runs, bound runtime with `--stop-after` and use `--include-control` when you need subscription/control diagnostics.
+
 ## Attachment send validation errors
 
 - `--attachment-upload-id` is required when using attachment metadata override flags.
diff --git a/internal/cmd/events_tail_test.go b/internal/cmd/events_tail_test.go
@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -91,3 +92,271 @@ func TestEventsTailUnsupportedRouteMessage(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 }
+
+func TestEventsTailSkipsControlMessagesByDefault(t *testing.T) {
+	t.Setenv("BEEPER_TOKEN", "test-token")
+	t.Setenv("BEEPER_ACCESS_TOKEN", "")
+
+	upgrader := websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool { return true },
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/ws" {
+			http.NotFound(w, r)
+			return
+		}
+		if got := r.Header.Get("Authorization"); got != "Bearer test-token" {
+			http.Error(w, "unauthorized", http.StatusUnauthorized)
+			return
+		}
+
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer func() { _ = conn.Close() }()
+
+		var sub map[string]any
+		if err := conn.ReadJSON(&sub); err != nil {
+			return
+		}
+
+		_ = conn.WriteJSON(map[string]any{
+			"type":    "ready",
+			"version": 1,
+			"chatIDs": []string{"*"},
+		})
+		_ = conn.WriteJSON(map[string]any{
+			"type":   "message.upserted",
+			"seq":    2,
+			"ts":     1739320000001,
+			"chatID": "chat_a",
+			"ids":    []string{"m2"},
+		})
+		time.Sleep(100 * time.Millisecond)
+	}))
+	defer server.Close()
+
+	ctx := outfmt.WithMode(context.Background(), outfmt.Mode{JSON: true})
+	cmd := EventsTailCmd{
+		All:       true,
+		Reconnect: false,
+		StopAfter: 250 * time.Millisecond,
+	}
+
+	out, _ := captureOutput(t, func() {
+		if err := cmd.Run(ctx, &RootFlags{BaseURL: server.URL, Timeout: 5}); err != nil {
+			t.Fatalf("Run() error = %v", err)
+		}
+	})
+
+	if strings.Contains(out, `"type":"ready"`) {
+		t.Fatalf("expected ready control message to be filtered, got: %s", out)
+	}
+	if !strings.Contains(out, `"type":"message.upserted"`) {
+		t.Fatalf("expected message.upserted in output, got: %s", out)
+	}
+}
+
+func TestEventsTailIncludesControlMessagesWhenFlagSet(t *testing.T) {
+	t.Setenv("BEEPER_TOKEN", "test-token")
+	t.Setenv("BEEPER_ACCESS_TOKEN", "")
+
+	upgrader := websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool { return true },
+	}
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/ws" {
+			http.NotFound(w, r)
+			return
+		}
+		if got := r.Header.Get("Authorization"); got != "Bearer test-token" {
+			http.Error(w, "unauthorized", http.StatusUnauthorized)
+			return
+		}
+
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer func() { _ = conn.Close() }()
+
+		var sub map[string]any
+		if err := conn.ReadJSON(&sub); err != nil {
+			return
+		}
+
+		_ = conn.WriteJSON(map[string]any{
+			"type":    "ready",
+			"version": 1,
+			"chatIDs": []string{"*"},
+		})
+		time.Sleep(100 * time.Millisecond)
+	}))
+	defer server.Close()
+
+	ctx := outfmt.WithMode(context.Background(), outfmt.Mode{JSON: true})
+	cmd := EventsTailCmd{
+		All:            true,
+		IncludeControl: true,
+		Reconnect:      false,
+		StopAfter:      250 * time.Millisecond,
+	}
+
+	out, _ := captureOutput(t, func() {
+		if err := cmd.Run(ctx, &RootFlags{BaseURL: server.URL, Timeout: 5}); err != nil {
+			t.Fatalf("Run() error = %v", err)
+		}
+	})
+
+	if !strings.Contains(out, `"type":"ready"`) {
+		t.Fatalf("expected ready control message in output, got: %s", out)
+	}
+}
+
+func TestEventsTailReconnectsAndResubscribesAfterDisconnect(t *testing.T) {
+	t.Setenv("BEEPER_TOKEN", "test-token")
+	t.Setenv("BEEPER_ACCESS_TOKEN", "")
+
+	upgrader := websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool { return true },
+	}
+
+	var connectCount atomic.Int32
+	var subscriptionCount atomic.Int32
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/ws" {
+			http.NotFound(w, r)
+			return
+		}
+		if got := r.Header.Get("Authorization"); got != "Bearer test-token" {
+			http.Error(w, "unauthorized", http.StatusUnauthorized)
+			return
+		}
+
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer func() { _ = conn.Close() }()
+
+		attempt := connectCount.Add(1)
+
+		var sub map[string]any
+		if err := conn.ReadJSON(&sub); err != nil {
+			return
+		}
+		subscriptionCount.Add(1)
+
+		if attempt == 1 {
+			// Force reconnect by closing before any domain event is emitted.
+			return
+		}
+
+		_ = conn.WriteJSON(map[string]any{
+			"type":   "message.upserted",
+			"seq":    3,
+			"ts":     1739320000002,
+			"chatID": "chat_b",
+			"ids":    []string{"m3"},
+		})
+		time.Sleep(100 * time.Millisecond)
+	}))
+	defer server.Close()
+
+	ctx := outfmt.WithMode(context.Background(), outfmt.Mode{JSON: true})
+	cmd := EventsTailCmd{
+		All:            true,
+		Reconnect:      true,
+		ReconnectDelay: 10 * time.Millisecond,
+		StopAfter:      300 * time.Millisecond,
+	}
+
+	out, _ := captureOutput(t, func() {
+		if err := cmd.Run(ctx, &RootFlags{BaseURL: server.URL, Timeout: 5}); err != nil {
+			t.Fatalf("Run() error = %v", err)
+		}
+	})
+
+	if !strings.Contains(out, `"type":"message.upserted"`) {
+		t.Fatalf("expected message.upserted in output, got: %s", out)
+	}
+	if got := connectCount.Load(); got < 2 {
+		t.Fatalf("expected reconnect attempts >= 2, got %d", got)
+	}
+	if got := subscriptionCount.Load(); got < 2 {
+		t.Fatalf("expected subscriptions to be re-sent after reconnect, got %d", got)
+	}
+}
+
+func TestEventsTailReconnectsAfterTemporaryHandshakeFailure(t *testing.T) {
+	t.Setenv("BEEPER_TOKEN", "test-token")
+	t.Setenv("BEEPER_ACCESS_TOKEN", "")
+
+	upgrader := websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool { return true },
+	}
+
+	var requestCount atomic.Int32
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/v1/ws" {
+			http.NotFound(w, r)
+			return
+		}
+		if got := r.Header.Get("Authorization"); got != "Bearer test-token" {
+			http.Error(w, "unauthorized", http.StatusUnauthorized)
+			return
+		}
+
+		attempt := requestCount.Add(1)
+		if attempt == 1 {
+			http.Error(w, "temporary failure", http.StatusInternalServerError)
+			return
+		}
+
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		defer func() { _ = conn.Close() }()
+
+		var sub map[string]any
+		if err := conn.ReadJSON(&sub); err != nil {
+			return
+		}
+		_ = conn.WriteJSON(map[string]any{
+			"type":   "message.upserted",
+			"seq":    4,
+			"ts":     1739320000003,
+			"chatID": "chat_c",
+			"ids":    []string{"m4"},
+		})
+		time.Sleep(100 * time.Millisecond)
+	}))
+	defer server.Close()
+
+	ctx := outfmt.WithMode(context.Background(), outfmt.Mode{JSON: true})
+	cmd := EventsTailCmd{
+		All:            true,
+		Reconnect:      true,
+		ReconnectDelay: 10 * time.Millisecond,
+		StopAfter:      350 * time.Millisecond,
+	}
+
+	out, _ := captureOutput(t, func() {
+		if err := cmd.Run(ctx, &RootFlags{BaseURL: server.URL, Timeout: 5}); err != nil {
+			t.Fatalf("Run() error = %v", err)
+		}
+	})
+
+	if !strings.Contains(out, `"type":"message.upserted"`) {
+		t.Fatalf("expected message.upserted in output, got: %s", out)
+	}
+	if got := requestCount.Load(); got < 2 {
+		t.Fatalf("expected at least two connection attempts, got %d", got)
+	}
+}
