feat: logs and logs --follow with WebSocket reconnect and gap-fill

README.md

@@ -86,6 +86,18 @@ Top-level settings are managed with `devlyft config get|set|list` (`api_url`, `c
 
 **Proxies and TLS inspection.** `HTTPS_PROXY`, `HTTP_PROXY` and `NO_PROXY` are honoured. If your network re-signs TLS traffic, point `DEVLYFT_CA_BUNDLE` or `devlyft config set ca_bundle` at a PEM file of the extra CA certificates; they are trusted in addition to the system roots, and every command that connects prints a notice while one is in use.
 
+## Logs
+
+```sh
+devlyft logs <deploy-id>                    # recent lines (last hour)
+devlyft logs <deploy-id> --follow           # keep streaming until Ctrl-C
+devlyft logs <deploy-id> -f --format json   # one JSON object per line
+```
+
+`--follow` reconnects on its own and fills in anything logged while it was
+disconnected. See [docs/logs.md](docs/logs.md) for the flags and how the live
+tail works.
+
 ## Privacy and update checks
 
 devlyft has no telemetry in v1: it does not collect or send usage data.

docs/logs.md

@@ -0,0 +1,78 @@
+# Logs
+
+`devlyft logs <deploy-id>` prints a deploy's recent log lines from the REST
+API. With `--follow` it then keeps printing new lines from logs-api's live-tail
+WebSocket until interrupted (Ctrl-C exits 130).
+
+```sh
+devlyft logs <deploy-id> --since 15m --pod web-abc --timestamps
+devlyft logs <deploy-id> --follow --format json | jq .
+devlyft logs <deploy-id> --follow --jq 'select(.line | test("error"; "i")) | .timestamp'
+```
+
+| Flag | Meaning |
+|---|---|
+| `-f`, `--follow` | Keep printing new lines until interrupted |
+| `--since D` | How far back the history read goes, as a Go duration (`15m`, `2h`); the server defaults to 1h |
+| `--pod NAME` | Only lines from one pod |
+| `--filter TEXT` | Only lines containing `TEXT`, case-insensitive |
+| `--match K=V` | Only JSON lines whose field matches; ops `=`, `!=`, `=~`, `!~`; repeatable |
+| `--limit N` | Most history lines (server default 500, at most 5000) |
+| `--timestamps` | Prefix each line with its timestamp and pod |
+
+Output: by default the raw line; `--format tsv` prints `timestamp`, `pod`
+and `line` tab-separated; `--format json` prints one JSON object per line
+(newline-delimited JSON, the API's `LogEntry` shape: `timestamp`, `line`,
+`pod`); `--format yaml` prints one document per line; `--jq` is applied to
+each line. Under `--format json` and `--jq`, reconnect notices go to the debug
+log instead of stderr, so stderr only ever carries the error object.
+
+## The live-tail contract
+
+The WebSocket is not described by the OpenAPI spec (devlyft/devlyft#234
+tracks publishing it), so the client is hand-written in
+`internal/api/logstream`. It relies on this contract:
+
+- **Endpoint.** `GET /v1/deploys/{deployId}/logs/stream` on the API host,
+  upgraded to a WebSocket (`wss://`). Query parameters `pod`, `filter` and
+  `match` (repeatable) narrow the stream as on the REST endpoint.
+- **Authentication.** The bearer token travels in the subprotocol header,
+  `Sec-WebSocket-Protocol: bearer, <token>`, because browsers cannot set an
+  `Authorization` header on a WebSocket handshake. The server selects the
+  `bearer` subprotocol. The token is never put in the URL, and debug logs
+  redact the header.
+- **Handshake errors.** Before upgrading, the server answers 400 (bad deploy ID
+  or field filter), 401 (missing or invalid token), 403 (not a member of the
+  owning organization) or 404 (unknown deploy) with a plain-text reason. The
+  CLI maps these to its usual errors and exit codes (3 for 401/403, 4 for 404)
+  and does not retry them.
+- **Messages.** Each text message is one log entry as JSON:
+  `{"timestamp": "<RFC 3339>", "line": "<raw line>", "pod": "<name>"}`, with
+  `pod` omitted when unknown.
+- **No resume.** The stream starts at the moment of connection. There is no
+  `since` or cursor on the WebSocket; history is only on the REST endpoint.
+- **Re-authentication.** The server closes the stream with close code `4001`
+  (`reauthenticate`) after at most 30 minutes, so the client re-checks the
+  token and organization membership. It sends the same code when its tail
+  ends for any other reason.
+
+## How `--follow` stays complete
+
+1. Read the history page from `GET /v1/deploys/{id}/logs` and print it. Its
+   `cursor` is where the live tail continues.
+2. Connect to the stream with a token fetched for this connection.
+3. After every connect, fill the gap: read REST history with
+   `since` = time since the newest line printed (plus a few seconds of slack
+   for clock skew), at the API's maximum of 5000 lines, and print what is new.
+   If the page is full, a notice says some lines may be missing.
+4. Relay the stream. Lines are de-duplicated by timestamp, pod and content,
+   since neither source has IDs, so the overlap between the gap-fill and the
+   stream is printed once.
+5. On a `4001` close, reconnect straight away with a fresh token and go back
+   to step 3. A `4001` within 10 seconds of connecting means the server's tail
+   failed rather than expired, so it is backed off like a drop.
+6. On a drop or failed connect, back off exponentially with full jitter
+   (0.5s doubling to 30s) and retry; after 8 failures in a row, exit 1.
+
+The WebSocket uses the same HTTP client as REST calls, so proxy and CA bundle
+settings apply to both.

go.mod

@@ -5,6 +5,7 @@ go 1.26.0
 toolchain go1.26.10
 
 require (
+	github.com/coder/websocket v1.8.15
 	github.com/getkin/kin-openapi v0.133.0
 	github.com/google/uuid v1.6.0
 	github.com/itchyny/gojq v0.12.19

go.sum

@@ -5,6 +5,8 @@ github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvF
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
+github.com/coder/websocket v1.8.15 h1:6B2JPeOGlpff2Uz6vOEH1Vzpi0iUz20A+lPVhPHtNUA=
+github.com/coder/websocket v1.8.15/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/danieljoos/wincred v1.2.3 h1:v7dZC2x32Ut3nEfRH+vhoZGvN72+dQ/snVXo/vMFLdQ=

internal/api/client.go

@@ -95,6 +95,10 @@ type Client interface {
 	GetPod(ctx context.Context, podID string) (*Pod, error)
 
 	GetLogs(ctx context.Context, deployID string, q *LogsQuery) (*LogsResponse, error)
+	// FollowLogs streams a deploy's new log lines to emit over the
+	// WebSocket until ctx is cancelled, reconnecting and filling gaps from
+	// GetLogs as needed. It returns ctx.Err() when cancelled.
+	FollowLogs(ctx context.Context, deployID string, opts *FollowOptions, emit func(LogEntry) error) error
 
 	ListMetricTypes(ctx context.Context) ([]string, error)
 	GetDeployMetrics(ctx context.Context, deployID string, q MetricsQuery) (*MetricsResponse, error)

internal/api/followlogs_test.go

@@ -0,0 +1,156 @@
+package api
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"slices"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/coder/websocket"
+
+	"github.com/devlyft/cli/internal/testutil/apitest"
+)
+
+// countingTokens hands out tok-1, tok-2, ... so tests can check each
+// reconnect asks for a fresh token.
+type countingTokens struct{ n atomic.Int32 }
+
+func (c *countingTokens) Token(context.Context) (string, error) {
+	return fmt.Sprintf("tok-%d", c.n.Add(1)), nil
+}
+
+func logLine(sec int, line string) string {
+	return fmt.Sprintf(`{"timestamp":"2026-10-25T12:00:%02dZ","line":%q,"pod":"web-1"}`, sec, line)
+}
+
+// End to end over a real WebSocket and the contract-checking REST fake:
+// the first stream sends a line and closes with 4001 at once; the client
+// reconnects with a fresh token, fills the gap from REST and carries on,
+// emitting every line exactly once and in order.
+func TestFollowLogsReconnectsWithFreshTokenAndFillsGap(t *testing.T) {
+	t.Parallel()
+
+	rest := apitest.NewServer(t, map[string]apitest.Response{
+		"GetV1DeploysDeployIdLogs": {Body: `{"entries":[` + logLine(1, "one") + `,` + logLine(2, "two") + `],"cursor":"2026-10-25T12:00:02Z"}`},
+	})
+
+	var (
+		mu       sync.Mutex
+		tokens   []string
+		sinces   []string
+		limits   []string
+		sessions int
+	)
+	mux := http.NewServeMux()
+	mux.HandleFunc("/v1/deploys/{id}/logs", func(w http.ResponseWriter, r *http.Request) {
+		mu.Lock()
+		sinces = append(sinces, r.URL.Query().Get("since"))
+		limits = append(limits, r.URL.Query().Get("limit"))
+		mu.Unlock()
+		// apitest matches routes against its own address.
+		r.Host = strings.TrimPrefix(rest.URL, "http://")
+		rest.ServeHTTP(w, r)
+	})
+	mux.HandleFunc("/v1/deploys/{id}/logs/stream", func(w http.ResponseWriter, r *http.Request) {
+		parts := strings.Split(r.Header.Get("Sec-WebSocket-Protocol"), ",")
+		mu.Lock()
+		tokens = append(tokens, strings.TrimSpace(parts[len(parts)-1]))
+		sessions++
+		n := sessions
+		mu.Unlock()
+
+		c, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{"bearer"}})
+		if err != nil {
+			return
+		}
+		if n == 1 {
+			_ = c.Write(r.Context(), websocket.MessageText, []byte(logLine(1, "one")))
+			_ = c.Close(4001, "reauthenticate")
+			return
+		}
+		_ = c.Write(r.Context(), websocket.MessageText, []byte(logLine(3, "three")))
+		// Hold the stream open until the client goes away.
+		_, _, _ = c.Read(r.Context())
+	})
+	srv := httptest.NewServer(mux)
+	t.Cleanup(srv.Close)
+
+	client, err := New(Options{BaseURL: srv.URL, Tokens: &countingTokens{}})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	ctx, cancel := context.WithCancel(t.Context())
+	defer cancel()
+	var got, warnings []string
+	err = client.FollowLogs(ctx, uuidA, &FollowOptions{
+		Pod:    "web-1",
+		Cursor: "2026-10-25T12:00:00Z",
+		Warn:   func(m string) { warnings = append(warnings, m) },
+	}, func(e LogEntry) error {
+		got = append(got, e.Line)
+		if e.Line == "three" {
+			cancel()
+		}
+		return nil
+	})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("FollowLogs() = %v, want context.Canceled", err)
+	}
+	if want := []string{"one", "two", "three"}; !slices.Equal(got, want) {
+		t.Errorf("emitted %q, want %q", got, want)
+	}
+
+	mu.Lock()
+	defer mu.Unlock()
+	// REST gap-fills draw from the same source, so only distinctness of
+	// the stream tokens is meaningful.
+	if len(tokens) != 2 || tokens[0] == tokens[1] {
+		t.Errorf("stream tokens = %q, want a fresh token per connection", tokens)
+	}
+	if len(sinces) != 2 {
+		t.Fatalf("gap-fills = %d, want one per connection", len(sinces))
+	}
+	for i, s := range sinces {
+		if !strings.HasSuffix(s, "s") || limits[i] != "5000" {
+			t.Errorf("gap-fill %d: since=%q limit=%q", i, s, limits[i])
+		}
+	}
+	// The early 4001 is reported like a drop rather than retried silently.
+	if len(warnings) != 1 {
+		t.Errorf("warnings = %q", warnings)
+	}
+}
+
+func TestFollowLogsRejectsBadID(t *testing.T) {
+	t.Parallel()
+
+	client, err := New(Options{BaseURL: "https://api.example.com"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = client.FollowLogs(t.Context(), "web", &FollowOptions{}, func(LogEntry) error { return nil })
+	if err == nil || !strings.Contains(err.Error(), "expected a UUID") {
+		t.Errorf("FollowLogs() = %v", err)
+	}
+}
+
+func TestRoundUpSeconds(t *testing.T) {
+	t.Parallel()
+
+	for in, want := range map[string]string{"0s": "1s", "1ns": "1s", "1s": "1s", "1500ms": "2s", "2m": "120s"} {
+		d, _ := parseDur(in)
+		if got := roundUpSeconds(d); got != want {
+			t.Errorf("roundUpSeconds(%s) = %s, want %s", in, got, want)
+		}
+	}
+}
+
+var parseDur = time.ParseDuration