feat: reconnect the live stream with bounded backoff

ARCHITECTURE.md

@@ -22,7 +22,10 @@ powerctl-cli/
 │   │   ├── client_test.go
 │   │   ├── queries.go           # GraphQL query definitions
 │   │   ├── websocket.go         # WebSocket for live streaming
-│   │   └── websocket_test.go
+│   │   ├── websocket_test.go
+│   │   ├── reconnect.go         # Backoff, hourly budget, error taxonomy
+│   │   ├── reconnect_test.go
+│   │   └── livestream_test.go   # End-to-end against a local WebSocket server
 │   ├── commands/
 │   │   ├── root.go              # Root command, global flags
 │   │   ├── config.go            # `powerctl config` - setup wizard
@@ -94,12 +97,43 @@ format: "markdown"               # Default output format
 
 - Protocol: `graphql-transport-ws`
 - Requires `User-Agent: powerctl-cli/1.0` — Tibber rejects the default Go client
-- Reconnection: none. A read or parse error ends the stream and exits 1
 - Heartbeat: none. The connection lives as long as the server keeps it open
-- Graceful shutdown on SIGINT/SIGTERM
+- Graceful shutdown on SIGINT/SIGTERM, including mid-backoff
 - A GraphQL error payload or a null `liveMeasurement` becomes an error rather
   than a nil measurement handed to the formatter (see #13)
 
+#### Reconnection (`reconnect.go`)
+
+`Subscribe` stays a single-attempt primitive; `SubscribeWithReconnect` layers
+the retry on top, so the connection code stays testable without a retry loop
+running through it.
+
+- Backoff: 5s, doubling to a 5-minute ceiling, with equal jitter (half the delay
+  fixed, half random)
+- Budget: at most 10 connection attempts per rolling hour. Tibber allows 20, and
+  a failing stream must not spend the allowance a manual restart needs. Hitting
+  the cap ends the stream with `ErrReconnectBudget`
+- Reset: a connection that delivered data and stayed up for a minute returns the
+  backoff to 5s
+- `--no-reconnect` skips the layer entirely and calls `Subscribe` directly
+
+Whether an error is retried is the substance of it, since a wrong answer either
+drops a recoverable stream or burns the hourly budget on a request that can
+never succeed:
+
+| Condition | Behaviour |
+|-----------|-----------|
+| Read error, abnormal or server-initiated close | Retry |
+| Handshake refused with HTTP 401/403 | Fatal, exit 1 |
+| Close code `4403 Invalid token` | Fatal, exit 1 |
+| `connection_error` in place of `connection_ack` | Fatal, exit 1 |
+| Unknown home ID, other GraphQL errors (see #13) | Fatal, exit 1 |
+| Ack timeout (10s, internal deadline) | Retry |
+| `ctx` cancelled by SIGINT/SIGTERM | Clean exit 0 |
+
+Fatal errors are tagged at the point they are raised (`markFatal`) rather than
+matched on message text later.
+
 ### Commands (`internal/commands/`)
 
 | Command | Input | Output | Exit Codes |
@@ -110,7 +144,7 @@ format: "markdown"               # Default output format
 | `home` | - | Home info | 0=OK, 1=Error |
 | `prices` | - | Price list | 0=OK, 1=Error |
 | `consumption`| `--resolution`, `--last`, `--home-id` | Consumption history | 0=OK, 1=Error |
-| `live` | `--home-id` | Stream | 0=Clean exit, 1=Error |
+| `live` | `--home-id`, `--no-reconnect` | Stream | 0=Clean exit, 1=Error |
 | `version` | - | Version, commit, build date | 0=OK |
 
 ### Output Formatters (`internal/output/`)
@@ -180,6 +214,7 @@ Authorization: Bearer <token>
 | Category | Strategy |
 |----------|----------|
 | Network errors | Log and exit with code 1 |
+| Dropped live stream | Reconnect with bounded backoff; exit 1 once the hourly budget is spent |
 | Auth errors | Clear message: "Invalid token" |
 | No Pulse | Exit 1 with "Pulse not enabled" |
 | Parse errors | Log raw response, exit 1 |
@@ -210,7 +245,9 @@ windows/amd64. Release builds do not use this — see below.
 internal/
 ├── api/
 │   ├── client_test.go       # Mock HTTP responses
-│   └── websocket_test.go    # Live payload parsing and error mapping
+│   ├── websocket_test.go    # Live payload parsing and error mapping
+│   ├── reconnect_test.go    # Backoff and budget, on a faked clock
+│   └── livestream_test.go   # Reconnect over a real socket, via httptest
 ├── config/
 │   └── config_test.go       # Config resolution order
 └── output/

README.md

@@ -173,6 +173,19 @@ powerctl live --home-id 96a14971-525a-4420-aae9-e5aefaf46a81
 
 `powerctl consumption` takes the same flag.
 
+A dropped connection is retried automatically, so the stream can be left
+running. The wait starts at 5 seconds and doubles to a 5-minute ceiling, and no
+more than 10 connections are opened in any rolling hour — Tibber allows 20, and
+a manual restart needs some left. Progress notes go to stderr, so `--format
+json` output stays pipeable.
+
+Failures a reconnect cannot fix — an invalid token, an unknown home ID — still
+exit 1 immediately instead of retrying. To get that for every error:
+
+```bash
+powerctl live --no-reconnect
+```
+
 ### Output Formats
 
 Default output is beautiful colored CLI. Change format with `--format`:
@@ -247,9 +260,12 @@ make lint           # Run linter (requires golangci-lint)
 - The ID passed to `--home-id` is not on your account. List valid IDs with `powerctl home`
 
 **Live stream disconnects**
-- Rate limit is 20 connections/hour
-- There is no automatic reconnect — the command exits with code 1 and the error.
-  Re-run it, or wrap it in a supervisor if you need it to stay up
+- Dropped connections are retried automatically — watch stderr for
+  `Reconnecting in ...`
+- Rate limit is 20 connections/hour. `powerctl` spends at most 10 of them, then
+  stops with "reconnect budget exhausted" rather than taking the rest; wait for
+  the hour to roll over before restarting
+- `--no-reconnect` restores the old behaviour of exiting 1 on the first error
 
 ## Contributing
 

internal/api/livestream_test.go

@@ -0,0 +1,261 @@
+package api
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/coder/websocket"
+
+	"github.com/kristofferrisa/powerctl-cli/internal/models"
+)
+
+// liveServer is a minimal graphql-transport-ws server standing in for Tibber, so
+// the reconnect path can be exercised over a real socket rather than a stub.
+type liveServer struct {
+	*httptest.Server
+
+	mu    sync.Mutex
+	conns int
+}
+
+// startLiveServer accepts WebSocket connections and hands each one to session,
+// already past connection_init, with the 1-based connection number.
+func startLiveServer(t *testing.T, session func(ctx context.Context, conn *websocket.Conn, n int)) *liveServer {
+	t.Helper()
+
+	s := &liveServer{}
+	s.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{
+			Subprotocols: []string{"graphql-transport-ws"},
+		})
+		if err != nil {
+			return
+		}
+		defer conn.CloseNow()
+
+		s.mu.Lock()
+		s.conns++
+		n := s.conns
+		s.mu.Unlock()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer cancel()
+
+		if _, _, err := conn.Read(ctx); err != nil { // connection_init
+			return
+		}
+		session(ctx, conn, n)
+	}))
+	t.Cleanup(s.Close)
+	return s
+}
+
+func (s *liveServer) connections() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.conns
+}
+
+// client returns a LiveClient pointed at this server.
+func (s *liveServer) client() *LiveClient {
+	c := NewLiveClient("test-token", "test-home")
+	c.endpoint = "ws" + strings.TrimPrefix(s.URL, "http")
+	return c
+}
+
+func send(ctx context.Context, conn *websocket.Conn, msgType string, payload string) error {
+	data, err := json.Marshal(wsMessage{Type: msgType, ID: "1", Payload: json.RawMessage(payload)})
+	if err != nil {
+		return err
+	}
+	return conn.Write(ctx, websocket.MessageText, data)
+}
+
+func ackAndAwaitSubscribe(ctx context.Context, conn *websocket.Conn) error {
+	if err := send(ctx, conn, "connection_ack", ""); err != nil {
+		return err
+	}
+	_, _, err := conn.Read(ctx) // subscribe
+	return err
+}
+
+const measurementPayload = `{"data":{"liveMeasurement":{"timestamp":"2026-01-01T12:00:00Z","power":1234}}}`
+
+// fastPolicy retries almost immediately so the tests do not wait on real backoff.
+func fastPolicy() ReconnectPolicy {
+	p := DefaultReconnectPolicy()
+	p.BaseDelay = time.Millisecond
+	p.MaxDelay = time.Millisecond
+	return p
+}
+
+func TestSubscribeWithReconnect_RecoversFromDroppedConnection(t *testing.T) {
+	server := startLiveServer(t, func(ctx context.Context, conn *websocket.Conn, n int) {
+		if err := ackAndAwaitSubscribe(ctx, conn); err != nil {
+			return
+		}
+		if err := send(ctx, conn, "next", measurementPayload); err != nil {
+			return
+		}
+		if n == 1 {
+			conn.CloseNow() // the blip: no close handshake, just a dead socket
+			return
+		}
+		_ = send(ctx, conn, "complete", "")
+	})
+
+	var measurements []*models.LiveMeasurement
+	err := server.client().SubscribeWithReconnect(context.Background(), fastPolicy(),
+		func(m *models.LiveMeasurement) error {
+			measurements = append(measurements, m)
+			return nil
+		})
+
+	if err != nil {
+		t.Fatalf("expected the stream to recover, got %v", err)
+	}
+	if server.connections() != 2 {
+		t.Errorf("expected 2 connections, got %d", server.connections())
+	}
+	if len(measurements) != 2 {
+		t.Fatalf("expected 2 measurements across the reconnect, got %d", len(measurements))
+	}
+	for i, m := range measurements {
+		if m == nil || m.Power != 1234 {
+			t.Errorf("measurement %d: expected 1234 W, got %v", i, m)
+		}
+	}
+}
+
+func TestSubscribeWithReconnect_DoesNotRetryFatalErrors(t *testing.T) {
+	tests := []struct {
+		name    string
+		session func(ctx context.Context, conn *websocket.Conn, n int)
+		wantErr string
+	}{
+		{
+			name: "rejected token",
+			session: func(ctx context.Context, conn *websocket.Conn, _ int) {
+				_ = send(ctx, conn, "connection_error", `{"message":"Invalid token"}`)
+			},
+			wantErr: "connection rejected",
+		},
+		{
+			name: "token refused with close code 4403",
+			session: func(_ context.Context, conn *websocket.Conn, _ int) {
+				_ = conn.Close(statusInvalidToken, "Invalid token")
+			},
+			wantErr: "4403",
+		},
+		{
+			name: "unknown home ID",
+			session: func(ctx context.Context, conn *websocket.Conn, _ int) {
+				if err := ackAndAwaitSubscribe(ctx, conn); err != nil {
+					return
+				}
+				_ = send(ctx, conn, "next", `{"errors":[{"message":"home not found"}],"data":{"liveMeasurement":null}}`)
+			},
+			wantErr: "invalid or non-existing home ID",
+		},
+		{
+			name: "subscription error",
+			session: func(ctx context.Context, conn *websocket.Conn, _ int) {
+				if err := ackAndAwaitSubscribe(ctx, conn); err != nil {
+					return
+				}
+				_ = send(ctx, conn, "error", `[{"message":"query is not valid"}]`)
+			},
+			wantErr: "subscription error",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			server := startLiveServer(t, tt.session)
+
+			err := server.client().SubscribeWithReconnect(context.Background(), fastPolicy(),
+				func(*models.LiveMeasurement) error { return nil })
+
+			if err == nil {
+				t.Fatal("expected an error, got nil")
+			}
+			if !strings.Contains(err.Error(), tt.wantErr) {
+				t.Errorf("expected an error containing %q, got %q", tt.wantErr, err)
+			}
+			if errors.Is(err, ErrReconnectBudget) {
+				t.Error("retried a fatal error until the connection budget ran out")
+			}
+			if server.connections() != 1 {
+				t.Errorf("expected 1 connection attempt, got %d", server.connections())
+			}
+		})
+	}
+}
+
+func TestSubscribeWithReconnect_StopsWhenCancelled(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	server := startLiveServer(t, func(sctx context.Context, conn *websocket.Conn, _ int) {
+		if err := ackAndAwaitSubscribe(sctx, conn); err != nil {
+			return
+		}
+		for {
+			if err := send(sctx, conn, "next", measurementPayload); err != nil {
+				return
+			}
+			time.Sleep(time.Millisecond)
+		}
+	})
+
+	received := 0
+	start := time.Now()
+	err := server.client().SubscribeWithReconnect(ctx, fastPolicy(),
+		func(*models.LiveMeasurement) error {
+			received++
+			if received == 3 {
+				cancel() // stand-in for Ctrl+C
+			}
+			return nil
+		})
+
+	// Ctrl+C must not wait on a close handshake the server may never answer.
+	if elapsed := time.Since(start); elapsed > 2*time.Second {
+		t.Errorf("expected a prompt exit on cancellation, took %v", elapsed)
+	}
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("expected context.Canceled, got %v", err)
+	}
+	if server.connections() != 1 {
+		t.Errorf("expected no reconnect after cancellation, got %d connections", server.connections())
+	}
+}
+
+func TestSubscribe_NoReconnectStopsOnFirstError(t *testing.T) {
+	server := startLiveServer(t, func(ctx context.Context, conn *websocket.Conn, _ int) {
+		if err := ackAndAwaitSubscribe(ctx, conn); err != nil {
+			return
+		}
+		if err := send(ctx, conn, "next", measurementPayload); err != nil {
+			return
+		}
+		conn.CloseNow()
+	})
+
+	// What `powerctl live --no-reconnect` does: one connection, then out.
+	err := server.client().Subscribe(context.Background(), func(*models.LiveMeasurement) error { return nil })
+
+	if err == nil {
+		t.Fatal("expected a read error, got nil")
+	}
+	if server.connections() != 1 {
+		t.Errorf("expected 1 connection, got %d", server.connections())
+	}
+}