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)