From 689430e2018145024390347ddc4aba8a20877f30 Mon Sep 17 00:00:00 2001
From: RogerCecilpot <james.edward@devlyft.io>
Date: Sun, 25 Oct 2026 14:29:34 +0000
Subject: [PATCH] feat: logs and logs --follow with WebSocket reconnect and
 gap-fill

`devlyft logs <deploy-id>` prints recent lines from the REST endpoint;
--follow then streams new lines from logs-api's live-tail WebSocket until
interrupted. Flags: --since, --pod, --filter, --match, --limit,
--timestamps. --format json prints NDJSON and --jq applies per line.

internal/api/logstream holds the hand-written client:
- Dial authenticates with the token in the bearer subprotocol (never the
  URL), maps rejected handshakes to clierr (401/403 exit 3, 404 exit 4)
  and logs the handshake through redaction.
- Follow reconnects after the server's 4001 re-auth close with a fresh
  token, backs off drops and failed dials with full jitter, and after
  every connect fills the gap from REST `since` (a duration on this API),
  de-duplicating by timestamp, pod and line. A 4001 that arrives at once
  is backed off, since the server also sends it when its tail fails.

The facade gains FollowLogs, built on the same http.Client, token source
and logger as REST calls so proxy and CA settings apply to the stream.
output.Printer gains PrintRecord for streamed records.

Refs devlyft/devlyft#254

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 README.md                                     |  12 +
 docs/logs.md                                  |  78 ++++
 go.mod                                        |   1 +
 go.sum                                        |   2 +
 internal/api/client.go                        |   4 +
 internal/api/followlogs_test.go               | 156 ++++++++
 internal/api/logstream/dial.go                | 179 +++++++++
 internal/api/logstream/dial_test.go           | 220 +++++++++++
 internal/api/logstream/follow.go              | 316 ++++++++++++++++
 internal/api/logstream/follow_test.go         | 349 ++++++++++++++++++
 internal/api/mock/client.go                   |  14 +
 internal/api/observe.go                       |  81 +++-
 internal/cli/logs/logs.go                     | 153 ++++++++
 internal/cli/logs_test.go                     | 133 +++++++
 internal/cli/root.go                          |   2 +
 internal/cli/testdata/logs_default.golden     |   7 +
 internal/cli/testdata/logs_filters.golden     |   7 +
 internal/cli/testdata/logs_follow.golden      |   9 +
 .../testdata/logs_follow_interrupted.golden   |  10 +
 internal/cli/testdata/logs_follow_json.golden |   9 +
 internal/cli/testdata/logs_follow_pod.golden  |   9 +
 .../testdata/logs_follow_stream_error.golden  |  11 +
 internal/cli/testdata/logs_jq.golden          |   5 +
 internal/cli/testdata/logs_json.golden        |   7 +
 internal/cli/testdata/logs_not_found.golden   |   6 +
 internal/cli/testdata/logs_timestamps.golden  |   7 +
 internal/cli/testdata/logs_tsv.golden         |   7 +
 .../cli/testdata/logs_usage_bad_since.golden  |   6 +
 internal/cli/testdata/logs_usage_limit.golden |   6 +
 .../testdata/logs_usage_negative_since.golden |   6 +
 .../cli/testdata/logs_usage_no_deploy.golden  |   6 +
 internal/cli/testdata/logs_usage_quiet.golden |   6 +
 internal/cli/testdata/logs_yaml.golden        |  15 +
 internal/output/output.go                     |  39 ++
 34 files changed, 1877 insertions(+), 1 deletion(-)
 create mode 100644 docs/logs.md
 create mode 100644 internal/api/followlogs_test.go
 create mode 100644 internal/api/logstream/dial.go
 create mode 100644 internal/api/logstream/dial_test.go
 create mode 100644 internal/api/logstream/follow.go
 create mode 100644 internal/api/logstream/follow_test.go
 create mode 100644 internal/cli/logs/logs.go
 create mode 100644 internal/cli/logs_test.go
 create mode 100644 internal/cli/testdata/logs_default.golden
 create mode 100644 internal/cli/testdata/logs_filters.golden
 create mode 100644 internal/cli/testdata/logs_follow.golden
 create mode 100644 internal/cli/testdata/logs_follow_interrupted.golden
 create mode 100644 internal/cli/testdata/logs_follow_json.golden
 create mode 100644 internal/cli/testdata/logs_follow_pod.golden
 create mode 100644 internal/cli/testdata/logs_follow_stream_error.golden
 create mode 100644 internal/cli/testdata/logs_jq.golden
 create mode 100644 internal/cli/testdata/logs_json.golden
 create mode 100644 internal/cli/testdata/logs_not_found.golden
 create mode 100644 internal/cli/testdata/logs_timestamps.golden
 create mode 100644 internal/cli/testdata/logs_tsv.golden
 create mode 100644 internal/cli/testdata/logs_usage_bad_since.golden
 create mode 100644 internal/cli/testdata/logs_usage_limit.golden
 create mode 100644 internal/cli/testdata/logs_usage_negative_since.golden
 create mode 100644 internal/cli/testdata/logs_usage_no_deploy.golden
 create mode 100644 internal/cli/testdata/logs_usage_quiet.golden
 create mode 100644 internal/cli/testdata/logs_yaml.golden

diff --git a/README.md b/README.md
index 1291142..a68f6b4 100644
--- a/README.md
+++ b/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.
diff --git a/docs/logs.md b/docs/logs.md
new file mode 100644
index 0000000..760b16c
--- /dev/null
+++ b/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.
diff --git a/go.mod b/go.mod
index 88d9b9f..9a8cb97 100644
--- a/go.mod
+++ b/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
diff --git a/go.sum b/go.sum
index 60ab272..8a463a9 100644
--- a/go.sum
+++ b/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=
diff --git a/internal/api/client.go b/internal/api/client.go
index 999cefa..63897bd 100644
--- a/internal/api/client.go
+++ b/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)
diff --git a/internal/api/followlogs_test.go b/internal/api/followlogs_test.go
new file mode 100644
index 0000000..8d1434a
--- /dev/null
+++ b/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
diff --git a/internal/api/logstream/dial.go b/internal/api/logstream/dial.go
new file mode 100644
index 0000000..41bfdb7
--- /dev/null
+++ b/internal/api/logstream/dial.go
@@ -0,0 +1,179 @@
+package logstream
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"github.com/coder/websocket"
+
+	"github.com/devlyft/cli/internal/clierr"
+	"github.com/devlyft/cli/internal/redact"
+)
+
+// CloseReauth is the close code logs-api sends when a stream reaches its
+// maximum lifetime (30 minutes) or its tail ends.
+const CloseReauth websocket.StatusCode = 4001
+
+// bearerSubprotocol is the first of the two subprotocol values carrying
+// the token: `Sec-WebSocket-Protocol: bearer, <token>`. Browsers cannot set
+// an Authorization header on a WebSocket handshake, so logs-api reads the
+// token from there instead.
+const bearerSubprotocol = "bearer"
+
+// readLimit is the largest frame accepted; one frame is one log line.
+const readLimit = 1 << 20
+
+// DialOptions describe a live-tail connection.
+type DialOptions struct {
+	// BaseURL is the API origin, e.g. https://api.devlyft.io.
+	BaseURL  string
+	DeployID string
+	// Pod, Filter and Match narrow the stream as on the REST endpoint.
+	Pod    string
+	Filter string
+	Match  []string
+	// Token returns a valid bearer token; it is asked on every dial.
+	Token func(ctx context.Context) (string, error)
+	// HTTPClient carries proxy and CA settings; nil uses a default client.
+	HTTPClient *http.Client
+	UserAgent  string
+	// Logger receives redacted debug logs of each handshake; nil discards.
+	Logger *slog.Logger
+}
+
+// StreamURL returns the WebSocket URL for opts: wss for an https API and
+// ws for a plain-http loopback API used in tests.
+func StreamURL(opts *DialOptions) (string, error) {
+	u, err := url.Parse(strings.TrimRight(opts.BaseURL, "/"))
+	if err != nil || u.Host == "" {
+		return "", fmt.Errorf("invalid API URL %q", opts.BaseURL)
+	}
+	switch u.Scheme {
+	case "https":
+		u.Scheme = "wss"
+	case "http":
+		u.Scheme = "ws"
+	default:
+		return "", fmt.Errorf("invalid API URL %q", opts.BaseURL)
+	}
+	u.Path += "/v1/deploys/" + url.PathEscape(opts.DeployID) + "/logs/stream"
+	q := url.Values{}
+	if opts.Pod != "" {
+		q.Set("pod", opts.Pod)
+	}
+	if opts.Filter != "" {
+		q.Set("filter", opts.Filter)
+	}
+	for _, m := range opts.Match {
+		q.Add("match", m)
+	}
+	u.RawQuery = q.Encode()
+	return u.String(), nil
+}
+
+// Dial opens a live-tail connection, authenticating with a fresh token in
+// the bearer subprotocol. The token never appears in the URL. A handshake
+// the server rejects returns a *clierr.Error carrying its HTTP status.
+func Dial(ctx context.Context, opts *DialOptions) (Conn, error) {
+	streamURL, err := StreamURL(opts)
+	if err != nil {
+		return nil, err
+	}
+	token, err := opts.Token(ctx)
+	if err != nil {
+		return nil, err
+	}
+	logger := opts.Logger
+	if logger == nil {
+		logger = slog.New(slog.DiscardHandler)
+	}
+	header := http.Header{}
+	if opts.UserAgent != "" {
+		header.Set("User-Agent", opts.UserAgent)
+	}
+
+	hc := opts.HTTPClient
+	if hc == nil {
+		hc = &http.Client{}
+	}
+	start := time.Now()
+	logger.Debug("websocket dial", "url", streamURL, "header", redact.Header(withProtocol(header, token)))
+	c, resp, err := websocket.Dial(ctx, streamURL, &websocket.DialOptions{
+		HTTPClient:   hc,
+		HTTPHeader:   header,
+		Subprotocols: []string{bearerSubprotocol, token},
+	})
+	if resp != nil && resp.Body != nil {
+		defer func() { _ = resp.Body.Close() }()
+	}
+	if err != nil {
+		if resp != nil && resp.StatusCode != http.StatusSwitchingProtocols {
+			logger.Debug("websocket rejected", "status", resp.StatusCode, "duration", time.Since(start))
+			return nil, handshakeError(resp)
+		}
+		logger.Debug("websocket dial failed", "error", redact.String(err.Error()), "duration", time.Since(start))
+		return nil, fmt.Errorf("connecting to the log stream: %w", err)
+	}
+	logger.Debug("websocket connected", "duration", time.Since(start))
+	c.SetReadLimit(readLimit)
+	return &wsConn{c: c}, nil
+}
+
+// withProtocol returns h plus the subprotocol header as sent, for logging
+// through redaction.
+func withProtocol(h http.Header, token string) http.Header {
+	out := h.Clone()
+	out.Set("Sec-WebSocket-Protocol", bearerSubprotocol+", "+token)
+	return out
+}
+
+// handshakeError maps a rejected handshake to the CLI's errors. logs-api
+// answers with a plain-text reason before upgrading.
+func handshakeError(resp *http.Response) error {
+	var msg string
+	if resp.Body != nil {
+		b, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
+		msg = strings.TrimSpace(redact.String(string(b)))
+	}
+	if msg == "" {
+		msg = http.StatusText(resp.StatusCode)
+	}
+	return clierr.FromHTTP(resp.StatusCode, "log stream: "+msg, resp.Header.Get("X-Request-Id"))
+}
+
+type wsConn struct{ c *websocket.Conn }
+
+func (w *wsConn) Read(ctx context.Context) (Entry, error) {
+	for {
+		typ, data, err := w.c.Read(ctx)
+		if err != nil {
+			if websocket.CloseStatus(err) == CloseReauth {
+				return Entry{}, fmt.Errorf("%w: %w", ErrReauth, err)
+			}
+			return Entry{}, err
+		}
+		if typ != websocket.MessageText {
+			continue
+		}
+		var e Entry
+		if err := json.Unmarshal(data, &e); err != nil {
+			return Entry{}, fmt.Errorf("log stream: malformed message: %w", err)
+		}
+		return e, nil
+	}
+}
+
+// Close drops the connection without waiting for a close handshake: by
+// the time Follow closes a connection it has either been closed by the
+// server or is being abandoned.
+func (w *wsConn) Close() error {
+	_ = w.c.CloseNow()
+	return nil
+}
diff --git a/internal/api/logstream/dial_test.go b/internal/api/logstream/dial_test.go
new file mode 100644
index 0000000..be91f60
--- /dev/null
+++ b/internal/api/logstream/dial_test.go
@@ -0,0 +1,220 @@
+package logstream
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+
+	"github.com/coder/websocket"
+
+	"github.com/devlyft/cli/internal/clierr"
+	"github.com/devlyft/cli/internal/exitcode"
+)
+
+const deployID = "6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b"
+
+// wsServer is a minimal logs-api stream: it checks the bearer subprotocol,
+// sends frames and closes with the given code.
+func wsServer(t *testing.T, handle func(w http.ResponseWriter, r *http.Request)) *httptest.Server {
+	t.Helper()
+	srv := httptest.NewServer(http.HandlerFunc(handle))
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func acceptAndSend(t *testing.T, w http.ResponseWriter, r *http.Request, frames []string, code websocket.StatusCode) {
+	t.Helper()
+	c, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{"bearer"}})
+	if err != nil {
+		t.Errorf("accept: %v", err)
+		return
+	}
+	for _, f := range frames {
+		if err := c.Write(r.Context(), websocket.MessageText, []byte(f)); err != nil {
+			t.Errorf("write: %v", err)
+			return
+		}
+	}
+	_ = c.Close(code, "reauthenticate")
+}
+
+func TestDialSendsTokenInSubprotocolAndReadsFrames(t *testing.T) {
+	t.Parallel()
+
+	srv := wsServer(t, func(w http.ResponseWriter, r *http.Request) {
+		// logs-api splits the header on commas and trims each value.
+		got := strings.Split(r.Header.Get("Sec-WebSocket-Protocol"), ",")
+		if len(got) != 2 || strings.TrimSpace(got[0]) != "bearer" || strings.TrimSpace(got[1]) != "tok-1" {
+			t.Errorf("Sec-WebSocket-Protocol = %q, want bearer and the token", got)
+		}
+		if r.URL.Path != "/v1/deploys/"+deployID+"/logs/stream" {
+			t.Errorf("path = %q", r.URL.Path)
+		}
+		if q := r.URL.Query(); q.Get("pod") != "web-1" || q.Get("filter") != "err" || len(q["match"]) != 2 {
+			t.Errorf("query = %v", q)
+		}
+		if strings.Contains(r.URL.RawQuery, "tok-1") {
+			t.Error("token in the URL")
+		}
+		if ua := r.Header.Get("User-Agent"); ua != "devlyft/test" {
+			t.Errorf("User-Agent = %q", ua)
+		}
+		acceptAndSend(t, w, r, []string{
+			`{"timestamp":"2026-10-25T12:00:00.5Z","line":"hello","pod":"web-1"}`,
+		}, CloseReauth)
+	})
+
+	var logs bytes.Buffer
+	c, err := Dial(t.Context(), &DialOptions{
+		BaseURL:   srv.URL,
+		DeployID:  deployID,
+		Pod:       "web-1",
+		Filter:    "err",
+		Match:     []string{"level=error", "service=~api"},
+		Token:     func(context.Context) (string, error) { return "tok-1", nil },
+		UserAgent: "devlyft/test",
+		Logger:    slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug})),
+	})
+	if err != nil {
+		t.Fatalf("Dial() = %v", err)
+	}
+	defer func() { _ = c.Close() }()
+
+	e, err := c.Read(t.Context())
+	if err != nil || e.Line != "hello" || *e.Pod != "web-1" || e.Timestamp != "2026-10-25T12:00:00.5Z" {
+		t.Fatalf("Read() = %+v, %v", e, err)
+	}
+	if _, err := c.Read(t.Context()); !errors.Is(err, ErrReauth) {
+		t.Errorf("Read() after 4001 = %v, want ErrReauth", err)
+	}
+	if strings.Contains(logs.String(), "tok-1") {
+		t.Errorf("debug log leaked the token:\n%s", logs.String())
+	}
+	if !strings.Contains(logs.String(), "websocket dial") {
+		t.Errorf("no debug log of the dial:\n%s", logs.String())
+	}
+}
+
+func TestReadOtherCloseIsNotReauth(t *testing.T) {
+	t.Parallel()
+
+	srv := wsServer(t, func(w http.ResponseWriter, r *http.Request) {
+		acceptAndSend(t, w, r, []string{"not json"}, websocket.StatusGoingAway)
+	})
+	c, err := Dial(t.Context(), &DialOptions{BaseURL: srv.URL, DeployID: deployID, Token: staticToken})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := c.Read(t.Context()); err == nil || !strings.Contains(err.Error(), "malformed") {
+		t.Errorf("Read() = %v, want a malformed-message error", err)
+	}
+	if _, err := c.Read(t.Context()); err == nil || errors.Is(err, ErrReauth) {
+		t.Errorf("Read() = %v, want a non-reauth close", err)
+	}
+	_ = c.Close()
+}
+
+func TestReadSkipsBinaryFrames(t *testing.T) {
+	t.Parallel()
+
+	srv := wsServer(t, func(w http.ResponseWriter, r *http.Request) {
+		c, err := websocket.Accept(w, r, &websocket.AcceptOptions{Subprotocols: []string{"bearer"}})
+		if err != nil {
+			return
+		}
+		_ = c.Write(r.Context(), websocket.MessageBinary, []byte{1, 2})
+		b, _ := json.Marshal(Entry{Timestamp: "2026-10-25T12:00:00Z", Line: "text"})
+		_ = c.Write(r.Context(), websocket.MessageText, b)
+		_ = c.Close(websocket.StatusNormalClosure, "")
+	})
+	c, err := Dial(t.Context(), &DialOptions{BaseURL: srv.URL, DeployID: deployID, Token: staticToken})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if e, err := c.Read(t.Context()); err != nil || e.Line != "text" {
+		t.Errorf("Read() = %+v, %v", e, err)
+	}
+}
+
+func TestDialRejectedHandshake(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		status int
+		body   string
+		exit   int
+	}{
+		{http.StatusUnauthorized, "unauthorized\n", exitcode.Auth},
+		{http.StatusForbidden, "forbidden\n", exitcode.Auth},
+		{http.StatusNotFound, "not found\n", exitcode.NotFound},
+		{http.StatusBadRequest, "invalid field filter: bad\n", exitcode.Error},
+		{http.StatusServiceUnavailable, "", exitcode.Error},
+	}
+	for _, tt := range tests {
+		srv := wsServer(t, func(w http.ResponseWriter, _ *http.Request) {
+			w.Header().Set("X-Request-Id", "req-7")
+			http.Error(w, strings.TrimSuffix(tt.body, "\n"), tt.status)
+		})
+		_, err := Dial(t.Context(), &DialOptions{BaseURL: srv.URL, DeployID: deployID, Token: staticToken})
+		e, ok := errors.AsType[*clierr.Error](err)
+		if !ok {
+			t.Errorf("%d: Dial() = %v, want *clierr.Error", tt.status, err)
+			continue
+		}
+		if e.HTTPStatus != tt.status || e.ExitCode() != tt.exit || e.RequestID != "req-7" {
+			t.Errorf("%d: got status %d exit %d request %q", tt.status, e.HTTPStatus, e.ExitCode(), e.RequestID)
+		}
+		if !strings.HasPrefix(e.Message, "log stream: ") {
+			t.Errorf("%d: message %q", tt.status, e.Message)
+		}
+	}
+}
+
+func TestDialNetworkAndTokenErrors(t *testing.T) {
+	t.Parallel()
+
+	srv := httptest.NewServer(http.NotFoundHandler())
+	url := srv.URL
+	srv.Close()
+	if _, err := Dial(t.Context(), &DialOptions{BaseURL: url, DeployID: deployID, Token: staticToken}); err == nil ||
+		!strings.Contains(err.Error(), "connecting to the log stream") {
+		t.Errorf("Dial() to a closed server = %v", err)
+	}
+
+	boom := errors.New("not logged in")
+	if _, err := Dial(t.Context(), &DialOptions{
+		BaseURL: "https://api.example.com", DeployID: deployID,
+		Token: func(context.Context) (string, error) { return "", boom },
+	}); !errors.Is(err, boom) {
+		t.Errorf("Dial() with a failing token source = %v", err)
+	}
+}
+
+func TestStreamURL(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		base, want string
+		wantErr    bool
+	}{
+		{base: "https://api.devlyft.io", want: "wss://api.devlyft.io/v1/deploys/" + deployID + "/logs/stream"},
+		{base: "https://api.devlyft.io/", want: "wss://api.devlyft.io/v1/deploys/" + deployID + "/logs/stream"},
+		{base: "http://127.0.0.1:8080", want: "ws://127.0.0.1:8080/v1/deploys/" + deployID + "/logs/stream"},
+		{base: "ftp://example.com", wantErr: true},
+		{base: "::", wantErr: true},
+	}
+	for _, tt := range tests {
+		got, err := StreamURL(&DialOptions{BaseURL: tt.base, DeployID: deployID})
+		if (err != nil) != tt.wantErr || got != tt.want {
+			t.Errorf("StreamURL(%q) = %q, %v", tt.base, got, err)
+		}
+	}
+}
+
+func staticToken(context.Context) (string, error) { return "tok", nil }
diff --git a/internal/api/logstream/follow.go b/internal/api/logstream/follow.go
new file mode 100644
index 0000000..fa5d760
--- /dev/null
+++ b/internal/api/logstream/follow.go
@@ -0,0 +1,316 @@
+// Package logstream follows a deploy's logs live over logs-api's WebSocket.
+//
+// The stream has no resume cursor: it starts at the moment of connection,
+// and the server closes it with code 4001 at least every 30 minutes so the
+// client re-authenticates. Follow therefore reconnects with a fresh token
+// after every close or drop and fills the gap from the REST history
+// endpoint, dropping lines it has already emitted. See docs/logs.md.
+package logstream
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"hash/fnv"
+	"math"
+	"math/rand/v2"
+	"time"
+
+	"github.com/devlyft/cli/internal/api/gen/logs"
+	"github.com/devlyft/cli/internal/clierr"
+)
+
+// Entry is one log line, the same shape on the WebSocket and the REST API.
+type Entry = logs.LogEntry
+
+// ErrReauth is returned by Conn.Read when the server closed the stream with
+// code 4001, asking the client to re-authenticate and reconnect.
+var ErrReauth = errors.New("server asked the client to re-authenticate")
+
+// Conn is one live-tail connection.
+type Conn interface {
+	// Read returns the next entry. It returns ErrReauth (possibly wrapped)
+	// on a 4001 close and another error on any other close or drop.
+	Read(ctx context.Context) (Entry, error)
+	Close() error
+}
+
+// Config wires Follow to the network. Dial and History are required.
+type Config struct {
+	// Dial opens a new connection with a fresh token.
+	Dial func(ctx context.Context) (Conn, error)
+	// History returns the entries of the last `since`, oldest first, and
+	// whether the server's line limit may have cut the page short.
+	History func(ctx context.Context, since time.Duration) (entries []Entry, truncated bool, err error)
+
+	// Last is the timestamp of the newest entry already shown (or the
+	// cursor from the history read); lines after it are filled from REST
+	// after each connect. Zero skips the first gap-fill.
+	Last time.Time
+	// Seen are entries already shown, so overlaps are not repeated.
+	Seen []Entry
+
+	// Warn reports recoverable problems, such as a reconnect; nil drops them.
+	Warn func(msg string)
+	// MaxAttempts is how many dials or sessions in a row may fail before
+	// Follow gives up. Zero means 8.
+	MaxAttempts int
+	// BaseDelay and MaxDelay bound the reconnect backoff. Zero means 500ms
+	// and 30s.
+	BaseDelay, MaxDelay time.Duration
+
+	// Replaced in tests.
+	Now    func() time.Time
+	Sleep  func(ctx context.Context, d time.Duration) error
+	Jitter func(n int64) int64
+}
+
+// gapSlack widens each gap-fill window to cover clock skew between the
+// client and the log store.
+const gapSlack = 5 * time.Second
+
+// minSession is how long a stream must last for its 4001 close to count as
+// the routine re-authentication. The server also closes with 4001 when its
+// tail fails, so an early close is backed off like a drop rather than
+// retried in a tight loop.
+const minSession = 10 * time.Second
+
+// maxSeen bounds how many lines are remembered for de-duplication.
+const maxSeen = 10000
+
+// Follow streams new entries to emit until ctx is cancelled, emit returns
+// an error, a dial fails permanently (401, 403, 404), or reconnecting
+// fails MaxAttempts times in a row. It returns ctx.Err() when cancelled.
+func Follow(ctx context.Context, cfg *Config, emit func(Entry) error) error {
+	f := newFollower(cfg, emit)
+	failures := 0
+	for {
+		started := f.now()
+		conn, err := cfg.Dial(ctx)
+		if err == nil {
+			err = f.session(ctx, conn)
+			_ = conn.Close()
+		} else if !retryableDial(err) && ctx.Err() == nil {
+			return err
+		}
+
+		var emitErr *emitError
+		switch {
+		case ctx.Err() != nil:
+			return ctx.Err()
+		case errors.As(err, &emitErr):
+			return emitErr.err
+		case errors.Is(err, ErrReauth) && f.now().Sub(started) >= minSession:
+			// Expected at least every 30 minutes: reconnect at once.
+			failures = 0
+			continue
+		}
+
+		failures++
+		if failures >= f.maxAttempts {
+			return fmt.Errorf("log stream: giving up after %d attempts: %w", failures, err)
+		}
+		if errors.Is(err, ErrReauth) {
+			f.warn("Log stream closed by the server; reconnecting.")
+		} else {
+			f.warn(fmt.Sprintf("Log stream interrupted (%v); reconnecting.", err))
+		}
+		if err := f.sleep(ctx, f.backoff(failures)); err != nil {
+			return err
+		}
+	}
+}
+
+// emitError carries an error returned by the caller's emit function.
+type emitError struct{ err error }
+
+func (e *emitError) Error() string { return e.err.Error() }
+
+type follower struct {
+	cfg  *Config
+	emit func(Entry) error
+
+	last time.Time
+	seen map[uint64]time.Time
+
+	maxAttempts         int
+	baseDelay, maxDelay time.Duration
+	now                 func() time.Time
+	sleep               func(context.Context, time.Duration) error
+	jitter              func(int64) int64
+}
+
+func newFollower(cfg *Config, emit func(Entry) error) *follower {
+	f := &follower{
+		cfg:         cfg,
+		emit:        emit,
+		last:        cfg.Last,
+		seen:        make(map[uint64]time.Time),
+		maxAttempts: cfg.MaxAttempts,
+		baseDelay:   cfg.BaseDelay,
+		maxDelay:    cfg.MaxDelay,
+		now:         cfg.Now,
+		sleep:       cfg.Sleep,
+		jitter:      cfg.Jitter,
+	}
+	if f.maxAttempts <= 0 {
+		f.maxAttempts = 8
+	}
+	if f.baseDelay <= 0 {
+		f.baseDelay = 500 * time.Millisecond
+	}
+	if f.maxDelay <= 0 {
+		f.maxDelay = 30 * time.Second
+	}
+	if f.now == nil {
+		f.now = time.Now
+	}
+	if f.sleep == nil {
+		f.sleep = sleepCtx
+	}
+	if f.jitter == nil {
+		f.jitter = rand.Int64N //nolint:gosec // jitter needs no cryptographic randomness
+	}
+	for i := range cfg.Seen {
+		e := &cfg.Seen[i]
+		f.remember(e)
+		if ts := timestamp(e); ts.After(f.last) {
+			f.last = ts
+		}
+	}
+	return f
+}
+
+// session gap-fills from REST, then relays the connection until it ends.
+// The gap-fill runs after connecting so nothing logged in between is lost;
+// overlaps with the stream are de-duplicated.
+func (f *follower) session(ctx context.Context, conn Conn) error {
+	if err := f.fillGap(ctx); err != nil {
+		return err
+	}
+	for {
+		e, err := conn.Read(ctx)
+		if err != nil {
+			return err
+		}
+		if err := f.offer(&e); err != nil {
+			return err
+		}
+	}
+}
+
+// fillGap emits what was logged between the newest entry shown and now.
+func (f *follower) fillGap(ctx context.Context) error {
+	if f.last.IsZero() {
+		return nil
+	}
+	since := f.now().Sub(f.last) + gapSlack
+	entries, truncated, err := f.cfg.History(ctx, since)
+	if err != nil {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		// The live stream is still useful without the gap; say so and go on.
+		f.warn(fmt.Sprintf("Could not fetch log lines missed while reconnecting: %v", err))
+		return nil
+	}
+	if truncated {
+		f.warn("Some log lines missed while reconnecting may not be shown: there were more than the server returns at once.")
+	}
+	for i := range entries {
+		if err := f.offer(&entries[i]); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// offer emits e unless it was already emitted.
+func (f *follower) offer(e *Entry) error {
+	if !f.remember(e) {
+		return nil
+	}
+	if err := f.emit(*e); err != nil {
+		return &emitError{err: err}
+	}
+	if ts := timestamp(e); ts.After(f.last) {
+		f.last = ts
+	}
+	f.prune()
+	return nil
+}
+
+// remember records e and reports whether it is new. Lines are identified
+// by timestamp, pod and content because the stream carries no IDs.
+func (f *follower) remember(e *Entry) bool {
+	ts := timestamp(e)
+	h := fnv.New64a()
+	_, _ = fmt.Fprintf(h, "%d\x00", ts.UnixNano())
+	if e.Pod != nil {
+		_, _ = h.Write([]byte(*e.Pod))
+	}
+	_, _ = h.Write([]byte{0})
+	_, _ = h.Write([]byte(e.Line))
+	key := h.Sum64()
+	if _, ok := f.seen[key]; ok {
+		return false
+	}
+	f.seen[key] = ts
+	return true
+}
+
+// prune forgets lines too old to reappear in a gap-fill, keeping memory
+// bounded on long follows.
+func (f *follower) prune() {
+	if len(f.seen) <= maxSeen {
+		return
+	}
+	cutoff := f.last.Add(-2 * gapSlack)
+	for k, ts := range f.seen {
+		if ts.Before(cutoff) {
+			delete(f.seen, k)
+		}
+	}
+}
+
+// backoff is exponential with full jitter.
+func (f *follower) backoff(attempt int) time.Duration {
+	d := float64(f.baseDelay) * math.Pow(2, float64(attempt-1))
+	d = math.Min(d, float64(f.maxDelay))
+	return time.Duration(f.jitter(int64(d)) + 1)
+}
+
+func (f *follower) warn(msg string) {
+	if f.cfg.Warn != nil {
+		f.cfg.Warn(msg)
+	}
+}
+
+// timestamp parses an entry's RFC 3339 timestamp; unparsable ones are zero.
+func timestamp(e *Entry) time.Time {
+	ts, err := time.Parse(time.RFC3339Nano, e.Timestamp)
+	if err != nil {
+		return time.Time{}
+	}
+	return ts
+}
+
+// retryableDial reports whether a failed dial is worth retrying: network
+// errors and 429/5xx are, while 401, 403 and 404 will not change.
+func retryableDial(err error) bool {
+	if e, ok := errors.AsType[*clierr.Error](err); ok && e.HTTPStatus != 0 {
+		return e.Retryable
+	}
+	return true
+}
+
+func sleepCtx(ctx context.Context, d time.Duration) error {
+	t := time.NewTimer(d)
+	defer t.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-t.C:
+		return nil
+	}
+}
diff --git a/internal/api/logstream/follow_test.go b/internal/api/logstream/follow_test.go
new file mode 100644
index 0000000..710230f
--- /dev/null
+++ b/internal/api/logstream/follow_test.go
@@ -0,0 +1,349 @@
+package logstream
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/devlyft/cli/internal/clierr"
+)
+
+var t0 = time.Date(2026, 10, 25, 12, 0, 0, 0, time.UTC)
+
+func entry(sec int, line string) Entry {
+	pod := "web-1"
+	return Entry{Timestamp: t0.Add(time.Duration(sec) * time.Second).Format(time.RFC3339Nano), Line: line, Pod: &pod}
+}
+
+// fakeConn replays entries, then ends with err.
+type fakeConn struct {
+	entries []Entry
+	err     error
+	closed  bool
+}
+
+func (c *fakeConn) Read(ctx context.Context) (Entry, error) {
+	if ctx.Err() != nil {
+		return Entry{}, ctx.Err()
+	}
+	if len(c.entries) == 0 {
+		return Entry{}, c.err
+	}
+	e := c.entries[0]
+	c.entries = c.entries[1:]
+	return e, nil
+}
+
+func (c *fakeConn) Close() error { c.closed = true; return nil }
+
+// harness drives Follow with scripted dials and history, a fake clock that
+// advances by one minute per dial, and recorded sleeps.
+type harness struct {
+	dials   []func() (Conn, error)
+	history func(since time.Duration) ([]Entry, bool, error)
+
+	now      time.Time
+	sinces   []time.Duration
+	sleeps   []time.Duration
+	warnings []string
+	emitted  []string
+}
+
+func (h *harness) config() *Config {
+	n := 0
+	return &Config{
+		Dial: func(context.Context) (Conn, error) {
+			if n >= len(h.dials) {
+				return nil, io.EOF
+			}
+			d := h.dials[n]
+			n++
+			h.now = h.now.Add(time.Minute)
+			return d()
+		},
+		History: func(_ context.Context, since time.Duration) ([]Entry, bool, error) {
+			h.sinces = append(h.sinces, since)
+			if h.history == nil {
+				return nil, false, nil
+			}
+			return h.history(since)
+		},
+		Warn:        func(msg string) { h.warnings = append(h.warnings, msg) },
+		MaxAttempts: 3,
+		Now:         func() time.Time { return h.now },
+		Sleep: func(_ context.Context, d time.Duration) error {
+			h.sleeps = append(h.sleeps, d)
+			return nil
+		},
+		Jitter: func(n int64) int64 { return n - 1 },
+	}
+}
+
+func (h *harness) emit(e Entry) error {
+	h.emitted = append(h.emitted, e.Line)
+	return nil
+}
+
+func conn(err error, entries ...Entry) func() (Conn, error) {
+	return func() (Conn, error) { return &fakeConn{entries: entries, err: err}, nil }
+}
+
+func TestFollowReauthReconnectsAndFillsGapWithoutDuplicates(t *testing.T) {
+	t.Parallel()
+
+	// A long first session (the clock moves past minSession) ends with the
+	// routine 4001; the gap-fill returns one line already streamed and one
+	// that was logged while disconnected.
+	h := &harness{now: t0}
+	h.dials = []func() (Conn, error){
+		conn(fmt.Errorf("closed: %w", ErrReauth), entry(10, "a"), entry(20, "b")),
+		conn(io.EOF, entry(40, "d")),
+	}
+	calls := 0
+	h.history = func(time.Duration) ([]Entry, bool, error) {
+		calls++
+		if calls == 2 { // the gap-fill after the first reconnect
+			return []Entry{entry(20, "b"), entry(30, "c")}, false, nil
+		}
+		return nil, false, nil
+	}
+	cfg := h.config()
+	cfg.Last = t0
+	cfg.Now = func() time.Time { h.now = h.now.Add(time.Minute); return h.now }
+
+	err := Follow(t.Context(), cfg, h.emit)
+	if err == nil || !strings.Contains(err.Error(), "giving up after 3 attempts") {
+		t.Fatalf("Follow() = %v, want giving up", err)
+	}
+	if want := []string{"a", "b", "c", "d"}; !slices.Equal(h.emitted, want) {
+		t.Errorf("emitted %q, want %q", h.emitted, want)
+	}
+	// The second gap-fill reaches back from "now" to the newest line seen
+	// (b at +20s), plus slack.
+	if len(h.sinces) < 2 || h.sinces[1] < gapSlack {
+		t.Errorf("gap-fill windows = %v", h.sinces)
+	}
+}
+
+func TestFollowEarlyReauthIsBackedOff(t *testing.T) {
+	t.Parallel()
+
+	// The server also closes with 4001 when its tail fails; an immediate
+	// close must not reconnect in a tight loop.
+	h := &harness{now: t0}
+	h.dials = []func() (Conn, error){
+		conn(ErrReauth), conn(ErrReauth), conn(ErrReauth),
+	}
+	cfg := h.config()
+	cfg.Now = func() time.Time { return t0 } // no time passes
+
+	err := Follow(t.Context(), cfg, h.emit)
+	if !errors.Is(err, ErrReauth) {
+		t.Fatalf("Follow() = %v, want ErrReauth after giving up", err)
+	}
+	if len(h.sleeps) != 2 {
+		t.Errorf("slept %d times, want 2 (backoff before each retry)", len(h.sleeps))
+	}
+}
+
+func TestFollowBackoffGrowsAndIsCapped(t *testing.T) {
+	t.Parallel()
+
+	h := &harness{now: t0}
+	for range 6 {
+		h.dials = append(h.dials, func() (Conn, error) { return nil, errors.New("connection refused") })
+	}
+	cfg := h.config()
+	cfg.MaxAttempts = 6
+	cfg.BaseDelay = time.Second
+	cfg.MaxDelay = 4 * time.Second
+
+	err := Follow(t.Context(), cfg, h.emit)
+	if err == nil || !strings.Contains(err.Error(), "connection refused") {
+		t.Fatalf("Follow() = %v", err)
+	}
+	want := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second, 4 * time.Second, 4 * time.Second}
+	if !slices.Equal(h.sleeps, want) {
+		t.Errorf("sleeps = %v, want %v", h.sleeps, want)
+	}
+	if len(h.warnings) != 5 {
+		t.Errorf("warnings = %q, want one per retry", h.warnings)
+	}
+}
+
+func TestFollowPermanentDialErrorStops(t *testing.T) {
+	t.Parallel()
+
+	for _, status := range []int{401, 403, 404} {
+		h := &harness{now: t0}
+		h.dials = []func() (Conn, error){
+			func() (Conn, error) { return nil, clierr.FromHTTP(status, "no", "") },
+		}
+		err := Follow(t.Context(), h.config(), h.emit)
+		if e, ok := errors.AsType[*clierr.Error](err); !ok || e.HTTPStatus != status {
+			t.Errorf("status %d: Follow() = %v, want the HTTP error", status, err)
+		}
+		if len(h.sleeps) != 0 {
+			t.Errorf("status %d: retried a permanent error", status)
+		}
+	}
+}
+
+func TestFollowRetriesUnavailable(t *testing.T) {
+	t.Parallel()
+
+	h := &harness{now: t0}
+	h.dials = []func() (Conn, error){
+		func() (Conn, error) { return nil, clierr.FromHTTP(503, "busy", "") },
+		conn(io.EOF, entry(1, "after")),
+	}
+	_ = Follow(t.Context(), h.config(), h.emit)
+	if !slices.Equal(h.emitted, []string{"after"}) {
+		t.Errorf("emitted %q", h.emitted)
+	}
+}
+
+func TestFollowSkipsSeenEntries(t *testing.T) {
+	t.Parallel()
+
+	h := &harness{now: t0}
+	h.dials = []func() (Conn, error){conn(io.EOF, entry(1, "old"), entry(2, "new"))}
+	cfg := h.config()
+	cfg.Seen = []Entry{entry(1, "old")}
+	_ = Follow(t.Context(), cfg, h.emit)
+	if !slices.Equal(h.emitted, []string{"new"}) {
+		t.Errorf("emitted %q, want only the unseen line", h.emitted)
+	}
+	// Seen entries set the gap-fill start even without a cursor.
+	if len(h.sinces) == 0 {
+		t.Error("no gap-fill after the first connect")
+	}
+}
+
+func TestFollowSameLineFromDifferentPodsIsNotDuplicate(t *testing.T) {
+	t.Parallel()
+
+	other := "web-2"
+	e2 := entry(1, "ready")
+	e2.Pod = &other
+	h := &harness{now: t0}
+	h.dials = []func() (Conn, error){conn(io.EOF, entry(1, "ready"), e2)}
+	_ = Follow(t.Context(), h.config(), h.emit)
+	if len(h.emitted) != 2 {
+		t.Errorf("emitted %q, want both pods' lines", h.emitted)
+	}
+}
+
+func TestFollowGapFillProblemsWarn(t *testing.T) {
+	t.Parallel()
+
+	h := &harness{now: t0}
+	h.dials = []func() (Conn, error){conn(io.EOF, entry(100, "live"))}
+	h.history = func(time.Duration) ([]Entry, bool, error) { return nil, false, errors.New("boom") }
+	cfg := h.config()
+	cfg.Last = t0
+	_ = Follow(t.Context(), cfg, h.emit)
+	if !slices.Equal(h.emitted, []string{"live"}) {
+		t.Errorf("emitted %q, want the live line despite the failed gap-fill", h.emitted)
+	}
+	if len(h.warnings) == 0 || !strings.Contains(h.warnings[0], "boom") {
+		t.Errorf("warnings = %q", h.warnings)
+	}
+
+	h = &harness{now: t0}
+	h.dials = []func() (Conn, error){conn(io.EOF)}
+	h.history = func(time.Duration) ([]Entry, bool, error) { return []Entry{entry(1, "x")}, true, nil }
+	cfg = h.config()
+	cfg.Last = t0
+	_ = Follow(t.Context(), cfg, h.emit)
+	if len(h.warnings) == 0 || !strings.Contains(h.warnings[0], "may not be shown") {
+		t.Errorf("truncated gap-fill did not warn: %q", h.warnings)
+	}
+}
+
+func TestFollowNoGapFillWithoutStart(t *testing.T) {
+	t.Parallel()
+
+	h := &harness{now: t0}
+	h.dials = []func() (Conn, error){conn(io.EOF, entry(1, "x"))}
+	_ = Follow(t.Context(), h.config(), h.emit)
+	if len(h.sinces) != 0 {
+		t.Errorf("gap-filled %v with nothing shown yet", h.sinces)
+	}
+}
+
+func TestFollowEmitErrorStops(t *testing.T) {
+	t.Parallel()
+
+	h := &harness{now: t0}
+	h.dials = []func() (Conn, error){conn(io.EOF, entry(1, "x"), entry(2, "y"))}
+	stop := errors.New("broken pipe")
+	err := Follow(t.Context(), h.config(), func(Entry) error { return stop })
+	if !errors.Is(err, stop) {
+		t.Errorf("Follow() = %v, want the emit error", err)
+	}
+}
+
+func TestFollowCancelled(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(t.Context())
+	h := &harness{now: t0}
+	c := &fakeConn{entries: []Entry{entry(1, "x")}}
+	h.dials = []func() (Conn, error){func() (Conn, error) { return c, nil }}
+	err := Follow(ctx, h.config(), func(Entry) error {
+		cancel()
+		return nil
+	})
+	if !errors.Is(err, context.Canceled) {
+		t.Errorf("Follow() = %v, want context.Canceled", err)
+	}
+	if !c.closed {
+		t.Error("connection not closed")
+	}
+
+	// Cancelled while backing off.
+	ctx, cancel = context.WithCancel(t.Context())
+	h = &harness{now: t0}
+	h.dials = []func() (Conn, error){func() (Conn, error) { return nil, errors.New("down") }}
+	cfg := h.config()
+	cfg.Sleep = func(ctx context.Context, d time.Duration) error {
+		cancel()
+		return sleepCtx(ctx, d)
+	}
+	if err := Follow(ctx, cfg, h.emit); !errors.Is(err, context.Canceled) {
+		t.Errorf("Follow() while backing off = %v, want context.Canceled", err)
+	}
+}
+
+func TestPruneBoundsMemory(t *testing.T) {
+	t.Parallel()
+
+	f := newFollower(&Config{}, func(Entry) error { return nil })
+	for i := range maxSeen + 100 {
+		if err := f.offer(&Entry{Timestamp: t0.Add(time.Duration(i) * time.Second).Format(time.RFC3339Nano), Line: "l"}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if len(f.seen) > maxSeen {
+		t.Errorf("remembering %d lines, want at most %d", len(f.seen), maxSeen)
+	}
+	// Recent lines are still remembered.
+	recent := Entry{Timestamp: t0.Add(time.Duration(maxSeen+99) * time.Second).Format(time.RFC3339Nano), Line: "l"}
+	if f.remember(&recent) {
+		t.Error("forgot the newest line")
+	}
+}
+
+func TestSleepCtx(t *testing.T) {
+	t.Parallel()
+
+	if err := sleepCtx(t.Context(), time.Millisecond); err != nil {
+		t.Errorf("sleepCtx() = %v", err)
+	}
+}
diff --git a/internal/api/mock/client.go b/internal/api/mock/client.go
index 5db8ab6..9e9cb7a 100644
--- a/internal/api/mock/client.go
+++ b/internal/api/mock/client.go
@@ -157,6 +157,20 @@ func (mr *MockClientMockRecorder) DeleteSecret(ctx, projectID, envID, name any)
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteSecret", reflect.TypeOf((*MockClient)(nil).DeleteSecret), ctx, projectID, envID, name)
 }
 
+// FollowLogs mocks base method.
+func (m *MockClient) FollowLogs(ctx context.Context, deployID string, opts *api.FollowOptions, emit func(api.LogEntry) error) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "FollowLogs", ctx, deployID, opts, emit)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// FollowLogs indicates an expected call of FollowLogs.
+func (mr *MockClientMockRecorder) FollowLogs(ctx, deployID, opts, emit any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "FollowLogs", reflect.TypeOf((*MockClient)(nil).FollowLogs), ctx, deployID, opts, emit)
+}
+
 // GetBuild mocks base method.
 func (m *MockClient) GetBuild(ctx context.Context, buildID string) (*api.Build, error) {
 	m.ctrl.T.Helper()
diff --git a/internal/api/observe.go b/internal/api/observe.go
index f2a6b81..934dedd 100644
--- a/internal/api/observe.go
+++ b/internal/api/observe.go
@@ -2,13 +2,16 @@ package api
 
 import (
 	"context"
+	"strconv"
+	"time"
 
 	"github.com/devlyft/cli/internal/api/gen/logs"
 	"github.com/devlyft/cli/internal/api/gen/metrics"
+	"github.com/devlyft/cli/internal/api/logstream"
 )
 
 // Logs are served by logs-api and metrics by metrics-api. Following logs
-// over the WebSocket is hand-written in internal/api/logstream (#254).
+// over the WebSocket is hand-written in internal/api/logstream.
 
 // LogsQuery narrows a log read. Empty fields use the server's defaults.
 type LogsQuery struct {
@@ -48,6 +51,82 @@ func (c *client) GetLogs(ctx context.Context, deployID string, q *LogsQuery) (*L
 	return expect(r.HTTPResponse, r.Body, r.JSON200, *n)
 }
 
+// gapFillLimit is the most lines one gap-fill asks for: the API's maximum.
+const gapFillLimit = 5000
+
+// FollowOptions configure FollowLogs.
+type FollowOptions struct {
+	// Pod, Filter and Match narrow the stream as in LogsQuery.
+	Pod    string
+	Filter string
+	Match  []string
+	// Cursor is the cursor of the GetLogs page already shown; the stream
+	// continues from it. Empty starts from the moment of connection.
+	Cursor string
+	// Seen are the entries already shown, so they are not repeated.
+	Seen []LogEntry
+	// Warn reports reconnects and gaps; nil drops them.
+	Warn func(msg string)
+}
+
+func (c *client) FollowLogs(ctx context.Context, deployID string, opts *FollowOptions, emit func(LogEntry) error) error {
+	id, err := parseID("deploy", deployID)
+	if err != nil {
+		return err
+	}
+	tokens := c.doer.tokens
+	if tokens == nil {
+		tokens = StaticToken("")
+	}
+	dial := &logstream.DialOptions{
+		BaseURL:  c.baseURL,
+		DeployID: id.String(),
+		Pod:      opts.Pod,
+		Filter:   opts.Filter,
+		Match:    opts.Match,
+		Token:    tokens.Token,
+		// The same client as every REST call, so proxy and CA settings
+		// apply to the WebSocket too.
+		HTTPClient: c.doer.hc,
+		UserAgent:  c.doer.ua,
+		Logger:     c.doer.log,
+	}
+	cfg := &logstream.Config{
+		Dial: func(ctx context.Context) (logstream.Conn, error) { return logstream.Dial(ctx, dial) },
+		History: func(ctx context.Context, since time.Duration) ([]logstream.Entry, bool, error) {
+			page, err := c.GetLogs(ctx, id.String(), &LogsQuery{
+				Pod:    opts.Pod,
+				Filter: opts.Filter,
+				Match:  opts.Match,
+				Since:  roundUpSeconds(since),
+				Limit:  gapFillLimit,
+			})
+			if err != nil {
+				return nil, false, err
+			}
+			return page.Entries, len(page.Entries) >= gapFillLimit, nil
+		},
+		Seen: opts.Seen,
+		Warn: opts.Warn,
+	}
+	if opts.Cursor != "" {
+		if ts, err := time.Parse(time.RFC3339Nano, opts.Cursor); err == nil {
+			cfg.Last = ts
+		}
+	}
+	return logstream.Follow(ctx, cfg, emit)
+}
+
+// roundUpSeconds formats d as a whole number of seconds, rounded up, in
+// the Go duration syntax the logs API takes for `since`.
+func roundUpSeconds(d time.Duration) string {
+	s := (d + time.Second - 1) / time.Second
+	if s < 1 {
+		s = 1
+	}
+	return strconv.FormatInt(int64(s), 10) + "s"
+}
+
 // MetricsQuery selects a metric series. Metric is required; empty optional
 // fields use the server's defaults.
 type MetricsQuery struct {
diff --git a/internal/cli/logs/logs.go b/internal/cli/logs/logs.go
new file mode 100644
index 0000000..f14e92d
--- /dev/null
+++ b/internal/cli/logs/logs.go
@@ -0,0 +1,153 @@
+// Package logscmd implements `devlyft logs`.
+package logscmd
+
+import (
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"github.com/devlyft/cli/internal/api"
+	"github.com/devlyft/cli/internal/clierr"
+	"github.com/devlyft/cli/internal/cmdutil"
+	"github.com/devlyft/cli/internal/output"
+)
+
+// maxLimit is the most lines the logs API returns in one read.
+const maxLimit = 5000
+
+type options struct {
+	follow     bool
+	since      string
+	pod        string
+	filter     string
+	match      []string
+	limit      int
+	timestamps bool
+}
+
+// NewCmd returns the `logs` command.
+func NewCmd(f *cmdutil.Factory) *cobra.Command {
+	var o options
+	cmd := &cobra.Command{
+		Use:   "logs <deploy-id>",
+		Short: "Print or follow a deploy's logs",
+		Long: `Print a deploy's recent log lines, oldest first, and with --follow keep
+printing new lines as they are logged until interrupted.
+
+--follow survives the server closing the stream (it does so at least every
+30 minutes to re-check access) and network drops: devlyft reconnects with
+a fresh token and fills in any lines logged while it was disconnected,
+without repeating lines it has already printed.
+
+With --format json each line is printed as one JSON object per line
+(newline-delimited JSON), and --jq is applied to each line:
+
+  devlyft logs <deploy-id> --follow --format json
+  devlyft logs <deploy-id> --jq 'select(.line | test("error"; "i")) | .timestamp'`,
+		Args: cobra.ExactArgs(1),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return run(cmd, f, &o, args[0])
+		},
+	}
+	fl := cmd.Flags()
+	fl.BoolVarP(&o.follow, "follow", "f", false, "Keep printing new lines until interrupted")
+	fl.StringVar(&o.since, "since", "", `How far back to read, as a duration such as "15m" or "2h" (server default 1h)`)
+	fl.StringVar(&o.pod, "pod", "", "Only lines from this pod (by name)")
+	fl.StringVar(&o.filter, "filter", "", "Only lines containing this text (case-insensitive)")
+	fl.StringArrayVar(&o.match, "match", nil, `Only JSON lines whose field matches, e.g. "level=error" or "service=~api-.*" (repeatable)`)
+	fl.IntVar(&o.limit, "limit", 0, fmt.Sprintf("Most lines to read from history (server default 500, at most %d)", maxLimit))
+	fl.BoolVar(&o.timestamps, "timestamps", false, "Prefix each line with its timestamp and pod")
+	return cmd
+}
+
+func run(cmd *cobra.Command, f *cmdutil.Factory, o *options, deployID string) error {
+	if err := o.validate(f); err != nil {
+		return err
+	}
+	client, err := f.Client()
+	if err != nil {
+		return err
+	}
+	ctx := cmd.Context()
+
+	page, err := client.GetLogs(ctx, deployID, &api.LogsQuery{
+		Pod:    o.pod,
+		Since:  o.since,
+		Filter: o.filter,
+		Match:  o.match,
+		Limit:  o.limit,
+	})
+	if err != nil {
+		return err
+	}
+	for i := range page.Entries {
+		if err := printEntry(f.Printer, o, &page.Entries[i]); err != nil {
+			return err
+		}
+	}
+	if !o.follow {
+		return nil
+	}
+
+	return client.FollowLogs(ctx, deployID, &api.FollowOptions{
+		Pod:    o.pod,
+		Filter: o.filter,
+		Match:  o.match,
+		Cursor: page.Cursor,
+		Seen:   page.Entries,
+		Warn: func(msg string) {
+			// Under --format json, stderr is reserved for the error object.
+			if f.Printer.Options().IsJSON() {
+				f.Log.Warn(msg)
+				return
+			}
+			_, _ = fmt.Fprintln(f.IO.ErrOut, msg)
+		},
+	}, func(e api.LogEntry) error {
+		return printEntry(f.Printer, o, &e)
+	})
+}
+
+func (o *options) validate(f *cmdutil.Factory) error {
+	if f.Printer.Options().Quiet {
+		return clierr.Usagef("--quiet is not supported by logs: log lines have no IDs")
+	}
+	if o.since != "" {
+		d, err := time.ParseDuration(o.since)
+		if err != nil || d <= 0 {
+			return clierr.Usagef(`invalid --since %q: expected a positive duration such as "15m" or "2h"`, o.since)
+		}
+	}
+	if o.limit < 0 || o.limit > maxLimit {
+		return clierr.Usagef("invalid --limit %d: must be between 1 and %d", o.limit, maxLimit)
+	}
+	return nil
+}
+
+func printEntry(p *output.Printer, o *options, e *api.LogEntry) error {
+	pod := ""
+	if e.Pod != nil {
+		pod = *e.Pod
+	}
+	text := e.Line
+	if o.timestamps {
+		text = strings.Join(nonEmpty(e.Timestamp, pod, e.Line), " ")
+	}
+	return p.PrintRecord(&output.Record{
+		Data: e,
+		Text: text,
+		Row:  []string{e.Timestamp, pod, e.Line},
+	})
+}
+
+func nonEmpty(ss ...string) []string {
+	out := ss[:0:0]
+	for _, s := range ss {
+		if s != "" {
+			out = append(out, s)
+		}
+	}
+	return out
+}
diff --git a/internal/cli/logs_test.go b/internal/cli/logs_test.go
new file mode 100644
index 0000000..1d9968a
--- /dev/null
+++ b/internal/cli/logs_test.go
@@ -0,0 +1,133 @@
+package cli_test
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+
+	"go.uber.org/mock/gomock"
+
+	"github.com/devlyft/cli/internal/api"
+	"github.com/devlyft/cli/internal/api/mock"
+	"github.com/devlyft/cli/internal/clierr"
+	"github.com/devlyft/cli/internal/testutil/clitest"
+	"github.com/devlyft/cli/internal/testutil/golden"
+)
+
+const deployID = "6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b"
+
+func logEntry(ts, pod, line string) api.LogEntry {
+	e := api.LogEntry{Timestamp: ts, Line: line}
+	if pod != "" {
+		e.Pod = &pod
+	}
+	return e
+}
+
+var logPage = &api.LogsResponse{
+	Entries: []api.LogEntry{
+		logEntry("2026-10-25T12:00:01Z", "web-1", "listening on :8080"),
+		logEntry("2026-10-25T12:00:02.5Z", "web-2", `{"level":"error","msg":"db\ttimeout"}`),
+		logEntry("2026-10-25T12:00:03Z", "", "no pod"),
+	},
+	Cursor: "2026-10-25T12:00:03Z",
+}
+
+// followed are the lines FollowLogs delivers after the history page.
+var followed = []api.LogEntry{
+	logEntry("2026-10-25T12:00:04Z", "web-1", "GET / 200"),
+}
+
+func expectHistory(q *api.LogsQuery) func(m *mock.MockClient) {
+	return func(m *mock.MockClient) {
+		m.EXPECT().GetLogs(gomock.Any(), deployID, q).Return(logPage, nil)
+	}
+}
+
+func expectFollow(q *api.LogsQuery, end error) func(m *mock.MockClient) {
+	return func(m *mock.MockClient) {
+		m.EXPECT().GetLogs(gomock.Any(), deployID, q).Return(logPage, nil)
+		m.EXPECT().FollowLogs(gomock.Any(), deployID, gomock.Any(), gomock.Any()).DoAndReturn(
+			func(_ context.Context, _ string, o *api.FollowOptions, emit func(api.LogEntry) error) error {
+				// The command continues from the page it printed.
+				if o.Cursor != logPage.Cursor || len(o.Seen) != len(logPage.Entries) || o.Pod != q.Pod {
+					return clierr.New(clierr.CodeInternal, "unexpected follow options %+v", o)
+				}
+				o.Warn("Log stream interrupted; reconnecting.")
+				for _, e := range followed {
+					if err := emit(e); err != nil {
+						return err
+					}
+				}
+				return end
+			})
+	}
+}
+
+//nolint:paralleltest // uses t.Setenv
+func TestLogsGolden(t *testing.T) {
+	t.Setenv("NO_COLOR", "")
+	t.Setenv("DEVLYFT_DEBUG", "")
+	t.Setenv("DEVLYFT_API_TOKEN", "")
+
+	all := &api.LogsQuery{}
+	tests := []struct {
+		name string
+		args []string
+		mock func(m *mock.MockClient)
+		ctx  func(context.Context) context.Context
+	}{
+		{name: "logs_default", args: []string{"logs", deployID}, mock: expectHistory(all)},
+		{name: "logs_timestamps", args: []string{"logs", deployID, "--timestamps"}, mock: expectHistory(all)},
+		{name: "logs_json", args: []string{"logs", deployID, "--format", "json"}, mock: expectHistory(all)},
+		{name: "logs_jq", args: []string{"logs", deployID, "--jq", "select(.pod == \"web-1\") | .line"}, mock: expectHistory(all)},
+		{name: "logs_yaml", args: []string{"logs", deployID, "--format", "yaml"}, mock: expectHistory(all)},
+		{name: "logs_tsv", args: []string{"logs", deployID, "--format", "tsv"}, mock: expectHistory(all)},
+		{
+			name: "logs_filters",
+			args: []string{"logs", deployID, "--pod", "web-1", "--since", "15m", "--filter", "err", "--match", "level=error", "--match", "svc=~api", "--limit", "100"},
+			mock: expectHistory(&api.LogsQuery{Pod: "web-1", Since: "15m", Filter: "err", Match: []string{"level=error", "svc=~api"}, Limit: 100}),
+		},
+		{name: "logs_follow", args: []string{"logs", deployID, "-f"}, mock: expectFollow(all, nil)},
+		{name: "logs_follow_json", args: []string{"logs", deployID, "-f", "--format", "json"}, mock: expectFollow(all, nil)},
+		{name: "logs_follow_pod", args: []string{"logs", deployID, "-f", "--pod", "web-1"}, mock: expectFollow(&api.LogsQuery{Pod: "web-1"}, nil)},
+		{
+			name: "logs_follow_interrupted",
+			args: []string{"logs", deployID, "--follow"},
+			mock: expectFollow(all, context.Canceled),
+			ctx: func(ctx context.Context) context.Context {
+				ctx, cancel := context.WithCancel(ctx)
+				cancel()
+				return ctx
+			},
+		},
+		{name: "logs_follow_stream_error", args: []string{"logs", deployID, "-f"}, mock: expectFollow(all, clierr.FromHTTP(403, "log stream: forbidden", "req-3"))},
+		{name: "logs_not_found", args: []string{"logs", deployID}, mock: func(m *mock.MockClient) {
+			m.EXPECT().GetLogs(gomock.Any(), deployID, all).Return(nil, clierr.FromHTTP(404, "deploy not found", "req-1"))
+		}},
+		{name: "logs_usage_no_deploy", args: []string{"logs"}},
+		{name: "logs_usage_bad_since", args: []string{"logs", deployID, "--since", "yesterday"}},
+		{name: "logs_usage_negative_since", args: []string{"logs", deployID, "--since", "-5m"}},
+		{name: "logs_usage_limit", args: []string{"logs", deployID, "--limit", "6000"}},
+		{name: "logs_usage_quiet", args: []string{"logs", deployID, "-q"}},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var client api.Client
+			if tt.mock != nil {
+				m := mock.NewMockClient(gomock.NewController(t))
+				tt.mock(m)
+				client = m
+			} else {
+				client = mock.NewMockClient(gomock.NewController(t)) // no calls expected
+			}
+			ctx := t.Context()
+			if tt.ctx != nil {
+				ctx = tt.ctx(ctx)
+			}
+			res := clitest.Run(t, &clitest.Options{Args: tt.args, API: client, Context: ctx})
+			golden.Assert(t, filepath.Join("testdata", tt.name+".golden"), res.Transcript())
+		})
+	}
+}
diff --git a/internal/cli/root.go b/internal/cli/root.go
index 0a27e72..40af4ad 100644
--- a/internal/cli/root.go
+++ b/internal/cli/root.go
@@ -19,6 +19,7 @@ import (
 	"github.com/devlyft/cli/internal/auth/credstore"
 	authcmd "github.com/devlyft/cli/internal/cli/auth"
 	configcmd "github.com/devlyft/cli/internal/cli/config"
+	logscmd "github.com/devlyft/cli/internal/cli/logs"
 	"github.com/devlyft/cli/internal/clierr"
 	"github.com/devlyft/cli/internal/cmdutil"
 	"github.com/devlyft/cli/internal/config"
@@ -78,6 +79,7 @@ func NewRootCmd(f *cmdutil.Factory) *cobra.Command {
 	root.AddCommand(authcmd.NewCmd(f))
 	root.AddCommand(configcmd.NewConfigCmd(f))
 	root.AddCommand(configcmd.NewContextCmd(f))
+	root.AddCommand(logscmd.NewCmd(f))
 
 	return root
 }
diff --git a/internal/cli/testdata/logs_default.golden b/internal/cli/testdata/logs_default.golden
new file mode 100644
index 0000000..6b4f9b2
--- /dev/null
+++ b/internal/cli/testdata/logs_default.golden
@@ -0,0 +1,7 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b
+--- exit 0
+--- stdout
+listening on :8080
+{"level":"error","msg":"db\ttimeout"}
+no pod
+--- stderr
diff --git a/internal/cli/testdata/logs_filters.golden b/internal/cli/testdata/logs_filters.golden
new file mode 100644
index 0000000..b21fbb1
--- /dev/null
+++ b/internal/cli/testdata/logs_filters.golden
@@ -0,0 +1,7 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b --pod web-1 --since 15m --filter err --match level=error --match svc=~api --limit 100
+--- exit 0
+--- stdout
+listening on :8080
+{"level":"error","msg":"db\ttimeout"}
+no pod
+--- stderr
diff --git a/internal/cli/testdata/logs_follow.golden b/internal/cli/testdata/logs_follow.golden
new file mode 100644
index 0000000..19818c1
--- /dev/null
+++ b/internal/cli/testdata/logs_follow.golden
@@ -0,0 +1,9 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b -f
+--- exit 0
+--- stdout
+listening on :8080
+{"level":"error","msg":"db\ttimeout"}
+no pod
+GET / 200
+--- stderr
+Log stream interrupted; reconnecting.
diff --git a/internal/cli/testdata/logs_follow_interrupted.golden b/internal/cli/testdata/logs_follow_interrupted.golden
new file mode 100644
index 0000000..9f7bdc3
--- /dev/null
+++ b/internal/cli/testdata/logs_follow_interrupted.golden
@@ -0,0 +1,10 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b --follow
+--- exit 130
+--- stdout
+listening on :8080
+{"level":"error","msg":"db\ttimeout"}
+no pod
+GET / 200
+--- stderr
+Log stream interrupted; reconnecting.
+Interrupted.
diff --git a/internal/cli/testdata/logs_follow_json.golden b/internal/cli/testdata/logs_follow_json.golden
new file mode 100644
index 0000000..94e2861
--- /dev/null
+++ b/internal/cli/testdata/logs_follow_json.golden
@@ -0,0 +1,9 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b -f --format json
+--- exit 0
+--- stdout
+{"line":"listening on :8080","pod":"web-1","timestamp":"2026-10-25T12:00:01Z"}
+{"line":"{\"level\":\"error\",\"msg\":\"db\\ttimeout\"}","pod":"web-2","timestamp":"2026-10-25T12:00:02.5Z"}
+{"line":"no pod","timestamp":"2026-10-25T12:00:03Z"}
+{"line":"GET / 200","pod":"web-1","timestamp":"2026-10-25T12:00:04Z"}
+--- stderr
+level=WARN msg="Log stream interrupted; reconnecting."
diff --git a/internal/cli/testdata/logs_follow_pod.golden b/internal/cli/testdata/logs_follow_pod.golden
new file mode 100644
index 0000000..a2f406e
--- /dev/null
+++ b/internal/cli/testdata/logs_follow_pod.golden
@@ -0,0 +1,9 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b -f --pod web-1
+--- exit 0
+--- stdout
+listening on :8080
+{"level":"error","msg":"db\ttimeout"}
+no pod
+GET / 200
+--- stderr
+Log stream interrupted; reconnecting.
diff --git a/internal/cli/testdata/logs_follow_stream_error.golden b/internal/cli/testdata/logs_follow_stream_error.golden
new file mode 100644
index 0000000..8d3db1a
--- /dev/null
+++ b/internal/cli/testdata/logs_follow_stream_error.golden
@@ -0,0 +1,11 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b -f
+--- exit 3
+--- stdout
+listening on :8080
+{"level":"error","msg":"db\ttimeout"}
+no pod
+GET / 200
+--- stderr
+Log stream interrupted; reconnecting.
+Error: log stream: forbidden
+Request ID: req-3
diff --git a/internal/cli/testdata/logs_jq.golden b/internal/cli/testdata/logs_jq.golden
new file mode 100644
index 0000000..852ac2a
--- /dev/null
+++ b/internal/cli/testdata/logs_jq.golden
@@ -0,0 +1,5 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b --jq select(.pod == "web-1") | .line
+--- exit 0
+--- stdout
+listening on :8080
+--- stderr
diff --git a/internal/cli/testdata/logs_json.golden b/internal/cli/testdata/logs_json.golden
new file mode 100644
index 0000000..7a34c91
--- /dev/null
+++ b/internal/cli/testdata/logs_json.golden
@@ -0,0 +1,7 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b --format json
+--- exit 0
+--- stdout
+{"line":"listening on :8080","pod":"web-1","timestamp":"2026-10-25T12:00:01Z"}
+{"line":"{\"level\":\"error\",\"msg\":\"db\\ttimeout\"}","pod":"web-2","timestamp":"2026-10-25T12:00:02.5Z"}
+{"line":"no pod","timestamp":"2026-10-25T12:00:03Z"}
+--- stderr
diff --git a/internal/cli/testdata/logs_not_found.golden b/internal/cli/testdata/logs_not_found.golden
new file mode 100644
index 0000000..3a48fbf
--- /dev/null
+++ b/internal/cli/testdata/logs_not_found.golden
@@ -0,0 +1,6 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b
+--- exit 4
+--- stdout
+--- stderr
+Error: deploy not found
+Request ID: req-1
diff --git a/internal/cli/testdata/logs_timestamps.golden b/internal/cli/testdata/logs_timestamps.golden
new file mode 100644
index 0000000..bc12c48
--- /dev/null
+++ b/internal/cli/testdata/logs_timestamps.golden
@@ -0,0 +1,7 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b --timestamps
+--- exit 0
+--- stdout
+2026-10-25T12:00:01Z web-1 listening on :8080
+2026-10-25T12:00:02.5Z web-2 {"level":"error","msg":"db\ttimeout"}
+2026-10-25T12:00:03Z no pod
+--- stderr
diff --git a/internal/cli/testdata/logs_tsv.golden b/internal/cli/testdata/logs_tsv.golden
new file mode 100644
index 0000000..c002fe8
--- /dev/null
+++ b/internal/cli/testdata/logs_tsv.golden
@@ -0,0 +1,7 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b --format tsv
+--- exit 0
+--- stdout
+2026-10-25T12:00:01Z	web-1	listening on :8080
+2026-10-25T12:00:02.5Z	web-2	{"level":"error","msg":"db\ttimeout"}
+2026-10-25T12:00:03Z		no pod
+--- stderr
diff --git a/internal/cli/testdata/logs_usage_bad_since.golden b/internal/cli/testdata/logs_usage_bad_since.golden
new file mode 100644
index 0000000..bb54489
--- /dev/null
+++ b/internal/cli/testdata/logs_usage_bad_since.golden
@@ -0,0 +1,6 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b --since yesterday
+--- exit 2
+--- stdout
+--- stderr
+Error: invalid --since "yesterday": expected a positive duration such as "15m" or "2h"
+Hint: Run 'devlyft logs --help' for usage.
diff --git a/internal/cli/testdata/logs_usage_limit.golden b/internal/cli/testdata/logs_usage_limit.golden
new file mode 100644
index 0000000..428f3bd
--- /dev/null
+++ b/internal/cli/testdata/logs_usage_limit.golden
@@ -0,0 +1,6 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b --limit 6000
+--- exit 2
+--- stdout
+--- stderr
+Error: invalid --limit 6000: must be between 1 and 5000
+Hint: Run 'devlyft logs --help' for usage.
diff --git a/internal/cli/testdata/logs_usage_negative_since.golden b/internal/cli/testdata/logs_usage_negative_since.golden
new file mode 100644
index 0000000..445de46
--- /dev/null
+++ b/internal/cli/testdata/logs_usage_negative_since.golden
@@ -0,0 +1,6 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b --since -5m
+--- exit 2
+--- stdout
+--- stderr
+Error: invalid --since "-5m": expected a positive duration such as "15m" or "2h"
+Hint: Run 'devlyft logs --help' for usage.
diff --git a/internal/cli/testdata/logs_usage_no_deploy.golden b/internal/cli/testdata/logs_usage_no_deploy.golden
new file mode 100644
index 0000000..008599a
--- /dev/null
+++ b/internal/cli/testdata/logs_usage_no_deploy.golden
@@ -0,0 +1,6 @@
+$ devlyft logs
+--- exit 2
+--- stdout
+--- stderr
+Error: accepts 1 arg(s), received 0
+Hint: Run 'devlyft logs --help' for usage.
diff --git a/internal/cli/testdata/logs_usage_quiet.golden b/internal/cli/testdata/logs_usage_quiet.golden
new file mode 100644
index 0000000..ceb03b0
--- /dev/null
+++ b/internal/cli/testdata/logs_usage_quiet.golden
@@ -0,0 +1,6 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b -q
+--- exit 2
+--- stdout
+--- stderr
+Error: --quiet is not supported by logs: log lines have no IDs
+Hint: Run 'devlyft logs --help' for usage.
diff --git a/internal/cli/testdata/logs_yaml.golden b/internal/cli/testdata/logs_yaml.golden
new file mode 100644
index 0000000..348239f
--- /dev/null
+++ b/internal/cli/testdata/logs_yaml.golden
@@ -0,0 +1,15 @@
+$ devlyft logs 6f1d3c2a-8a4b-4f7e-9c11-2b3d4e5f6a7b --format yaml
+--- exit 0
+--- stdout
+---
+line: listening on :8080
+pod: web-1
+timestamp: "2026-10-25T12:00:01Z"
+---
+line: '{"level":"error","msg":"db\ttimeout"}'
+pod: web-2
+timestamp: "2026-10-25T12:00:02.5Z"
+---
+line: no pod
+timestamp: "2026-10-25T12:00:03Z"
+--- stderr
diff --git a/internal/output/output.go b/internal/output/output.go
index 43f793d..c17c3c1 100644
--- a/internal/output/output.go
+++ b/internal/output/output.go
@@ -137,6 +137,45 @@ func (p *Printer) Print(r Result) error {
 	return fmt.Errorf("unhandled format %q", p.opts.Format)
 }
 
+// Record is one item of a stream, such as a log line, printed as soon as
+// it arrives rather than collected into a list.
+type Record struct {
+	// Data is printed by JSON (one compact object per line), YAML (one
+	// document each) and --jq (applied to each record).
+	Data any
+	// Text is the human form, printed for the default and table formats.
+	Text string
+	// Row is printed tab-separated for TSV.
+	Row []string
+}
+
+// PrintRecord writes one record of a stream to stdout. --quiet has no IDs
+// to print for records, so it prints Text.
+func (p *Printer) PrintRecord(r *Record) error {
+	w := p.ios.Out
+	if p.opts.JQ != "" {
+		return writeJQ(w, p.opts.JQ, r.Data)
+	}
+	switch p.opts.Format {
+	case FormatJSON:
+		enc := json.NewEncoder(w)
+		enc.SetEscapeHTML(false)
+		return enc.Encode(r.Data)
+	case FormatYAML:
+		if _, err := io.WriteString(w, "---\n"); err != nil {
+			return err
+		}
+		return writeYAML(w, r.Data)
+	case FormatTSV:
+		_, err := fmt.Fprintln(w, strings.Join(sanitizeRow(r.Row), "\t"))
+		return err
+	case FormatTable, FormatDefault:
+		_, err := fmt.Fprintln(w, r.Text)
+		return err
+	}
+	return fmt.Errorf("unhandled format %q", p.opts.Format)
+}
+
 func (p *Printer) writeTable(w io.Writer, r Result, aligned bool) error {
 	if r.Table == nil {
 		return writeYAML(w, r.Data)