diff --git a/packages/tangle-cli/src/tangle_cli/client.py b/packages/tangle-cli/src/tangle_cli/client.py
index 060c960..e1dac0f 100644
--- a/packages/tangle-cli/src/tangle_cli/client.py
+++ b/packages/tangle-cli/src/tangle_cli/client.py
@@ -9,7 +9,7 @@
 from __future__ import annotations
 
 import time
-from collections.abc import Iterable, Mapping
+from collections.abc import Iterable, Iterator, Mapping
 from dataclasses import asdict, is_dataclass
 from email.utils import parsedate_to_datetime
 from typing import Any
@@ -53,6 +53,14 @@ class TangleApiClient(GeneratedTangleApiOperations):
     _MAX_RATE_LIMIT_RETRIES = 3
     _RATE_LIMIT_BACKOFF_SECONDS = 1.0
     _MAX_RETRY_AFTER_SECONDS = 60.0
+    # Opening a log stream retries transient failures (transport-open errors
+    # and retryable 5xx) with doubling backoff, spent entirely before any line
+    # is yielded. An already-open stream that drops is the caller's to handle;
+    # never re-opening an established stream means lines cannot be duplicated.
+    _RETRYABLE_STREAM_STATUSES = frozenset({500, 502, 503, 504})
+    _MAX_STREAM_OPEN_ATTEMPTS = 7
+    _STREAM_OPEN_BACKOFF_SECONDS = 1.0
+    _MAX_STREAM_OPEN_BACKOFF_SECONDS = 30.0
 
     def __init__(
         self,
@@ -132,6 +140,12 @@ def _make_request(
             request_kwargs=kwargs,
         )
         if response.status_code == 401:
+            # The 401 response is discarded by the auth-refresh retry. For a
+            # streamed request it is an open streamed connection, so close it
+            # before refreshing auth and issuing the second request to avoid
+            # leaking it. ``response.headers`` stays available after close.
+            if kwargs.get("stream"):
+                response.close()
             self._refresh_auth()
             response = self._request_with_rate_limit_retries(
                 request_method,
@@ -152,7 +166,7 @@ def _request_with_rate_limit_retries(
         params: Mapping[str, Any] | None,
         json_data: Any,
         extra_headers: Mapping[str, str] | None,
-        timeout: float,
+        timeout: float | tuple[float, float | None],
         request_kwargs: Mapping[str, Any],
     ) -> requests.Response:
         response: requests.Response | None = None
@@ -168,6 +182,13 @@ def _request_with_rate_limit_retries(
             )
             if response.status_code != 429 or attempt == self._MAX_RATE_LIMIT_RETRIES:
                 return response
+            # The 429 response is discarded by the retry. For a streamed request
+            # it is an open streamed connection, so close it before the sleep to
+            # avoid holding it open while we wait. ``response.headers`` stays
+            # available after close, so ``_sleep_for_rate_limit`` can still read
+            # Retry-After.
+            if request_kwargs.get("stream"):
+                response.close()
             self._sleep_for_rate_limit(response, attempt)
         return response
 
@@ -181,6 +202,14 @@ def _sleep_for_rate_limit(self, response: requests.Response, attempt: int) -> No
             self.logger.info(f"429 rate limited; retrying in {delay:.1f}s")
         time.sleep(delay)
 
+    def _sleep_for_stream_open_retry(self, backoff: float, next_attempt: int, reason: str) -> None:
+        delay = min(backoff, self._MAX_STREAM_OPEN_BACKOFF_SECONDS)
+        self.logger.warn(
+            f"transient {reason} opening log stream; retrying in {delay:.1f}s "
+            f"(attempt {next_attempt}/{self._MAX_STREAM_OPEN_ATTEMPTS})"
+        )
+        time.sleep(delay)
+
     @staticmethod
     def _retry_after_delay(value: str | None) -> float | None:
         if not value:
@@ -205,7 +234,7 @@ def _request_with_same_origin_redirects(
         params: Mapping[str, Any] | None,
         json_data: Any,
         extra_headers: Mapping[str, str] | None,
-        timeout: float,
+        timeout: float | tuple[float, float | None],
         request_kwargs: Mapping[str, Any],
     ) -> requests.Response:
         """Send one request, following only same-origin redirects.
@@ -235,6 +264,16 @@ def _request_with_same_origin_redirects(
                 **request_kwargs,
             )
             if self.verbose:
+                # For streamed responses, reading ``response.text`` would buffer
+                # the entire body here, defeating callers that stream via
+                # ``iter_content``/``iter_lines``; a followed container-log
+                # stream may never terminate. Log a placeholder and leave the
+                # body unread.
+                response_body = (
+                    "<streaming body omitted>"
+                    if request_kwargs.get("stream")
+                    else response.text
+                )
                 log_http_exchange(
                     self.logger,
                     method=current_method,
@@ -243,7 +282,7 @@ def _request_with_same_origin_redirects(
                     request_body=current_json,
                     response_status=response.status_code,
                     response_headers=dict(response.headers),
-                    response_body=response.text,
+                    response_body=response_body,
                 )
             if response.status_code not in self._REDIRECT_STATUSES:
                 return response
@@ -254,6 +293,13 @@ def _request_with_same_origin_redirects(
 
             next_url = urljoin(response.url, location)
             if not self._same_origin(response.url, next_url):
+                # Close before raising so callers that catch this and fall back
+                # to another route (or a streamed open) do not leak the open
+                # streamed redirect response and its pooled connection.
+                try:
+                    response.close()
+                except Exception:
+                    pass
                 raise requests.HTTPError(
                     f"Refusing to follow cross-origin redirect from {response.url} to {next_url}",
                     response=response,
@@ -358,16 +404,108 @@ def get_execution_details(self, execution_id: str) -> GetExecutionInfoResponse:
         return details
 
     def stream_execution_container_log(self, execution_id: str) -> requests.Response:
-        response = self._make_request(
-            "GET",
-            self._format_path(
-                "/api/executions/{id}/stream_container_log",
-                {"id": execution_id},
-            ),
-            stream=True,
+        """Open the streaming container-log response for ``execution_id``.
+
+        The endpoint delivers raw log lines over a long-lived chunked HTTP
+        response; the client streams those lines as-is and does no
+        event-protocol parsing.
+
+        Establishing the stream (open + status check) follows a transient-error
+        retry budget: transport-open errors (connection/timeout) and retryable
+        5xx responses are retried with exponential backoff before any line is
+        read. Same-origin redirect protection errors (cross-origin ``HTTPError``
+        / ``TooManyRedirects``) are not transport blips and propagate
+        immediately. Once the stream is open the caller owns the response and
+        must close it; :meth:`iter_execution_container_log_lines` does that.
+
+        The request uses ``(connect, read)`` timeouts of ``(self.timeout,
+        None)``: the connect timeout still bounds opening the stream, but there
+        is no per-read timeout, because a healthy follow stream stays silent for
+        as long as the container emits no output.
+        """
+
+        path = self._format_path(
+            "/api/executions/{id}/stream_container_log",
+            {"id": execution_id},
         )
-        response.raise_for_status()
-        return response
+        backoff = self._STREAM_OPEN_BACKOFF_SECONDS
+        last_exc: requests.RequestException | None = None
+        last_error_response: requests.Response | None = None
+        for attempt in range(1, self._MAX_STREAM_OPEN_ATTEMPTS + 1):
+            try:
+                response = self._make_request(
+                    "GET", path, stream=True, timeout=(self.timeout, None)
+                )
+            except (requests.HTTPError, requests.TooManyRedirects) as exc:
+                # Same-origin redirect guard errors carry the rejected streamed
+                # response and are intentionally not retried. No iterator ever
+                # receives that response, so close it before re-raising.
+                if exc.response is not None:
+                    exc.response.close()
+                raise
+            except (requests.ConnectionError, requests.Timeout) as exc:
+                last_exc = exc
+                last_error_response = None
+                if attempt == self._MAX_STREAM_OPEN_ATTEMPTS:
+                    break
+                self._sleep_for_stream_open_retry(backoff, attempt + 1, type(exc).__name__)
+                backoff *= 2.0
+                continue
+            if response.status_code in self._RETRYABLE_STREAM_STATUSES:
+                response.close()
+                last_error_response = response
+                last_exc = None
+                if attempt == self._MAX_STREAM_OPEN_ATTEMPTS:
+                    break
+                self._sleep_for_stream_open_retry(
+                    backoff, attempt + 1, f"HTTP {response.status_code}"
+                )
+                backoff *= 2.0
+                continue
+            try:
+                response.raise_for_status()
+            except requests.HTTPError:
+                # Non-retryable status (e.g. 400/403/404): close the open
+                # streamed response before propagating so it is not leaked.
+                response.close()
+                raise
+            return response
+        if last_exc is not None:
+            raise last_exc
+        if last_error_response is not None:
+            last_error_response.raise_for_status()
+        # Defensive: every exhausted attempt records either a transport error
+        # (re-raised above) or a retryable-status response (raise_for_status
+        # always raises for those), so this cannot be reached.
+        raise RuntimeError(  # pragma: no cover
+            "log stream open retries exhausted without a failure to re-raise"
+        )
+
+    def iter_execution_container_log_lines(self, execution_id: str) -> Iterable[str]:
+        """Return an iterator of decoded container-log lines for ``execution_id``.
+
+        The stream is opened eagerly, so open failures (HTTP status or transport
+        errors) raise from this call rather than on first iteration; anything
+        raised while iterating is a drop of an already-open stream. The
+        underlying streaming response is always closed when iteration finishes
+        or the consumer stops early.
+        """
+
+        response = self.stream_execution_container_log(execution_id)
+
+        def lines() -> Iterator[str]:
+            try:
+                # Decode whole ``bytes`` lines as UTF-8 explicitly rather than
+                # via ``decode_unicode=True``: requests' charset guessing falls
+                # back to latin-1 when the response declares no charset and
+                # would mojibake non-ASCII output. Whole-line decoding also
+                # reassembles multibyte sequences split across stream chunks.
+                for raw in response.iter_lines():
+                    yield raw.decode("utf-8", "replace")
+            finally:
+                response.close()
+
+        return lines()
 
     def get_component_spec(self, digest: str) -> ComponentSpec:
         """Return a parsed domain component spec from the generated component endpoint."""
diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py b/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py
index 24742fa..66e17a1 100644
--- a/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py
+++ b/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py
@@ -15,12 +15,13 @@
 import re
 import time
 import uuid
-from collections.abc import Callable
+from collections.abc import Callable, Iterable
 from contextlib import AbstractContextManager, nullcontext
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Mapping
 
+import requests
 import yaml
 
 from .handler import TangleCliHandler
@@ -54,6 +55,27 @@ class AmbiguousPipelineRunRecoveryError(PipelineRunError):
     """Raised when submit recovery finds multiple runs for one submission id."""
 
 
+def _describe_http_error(exc: requests.HTTPError) -> str:
+    """Summarize an HTTP status failure with the attempted request target.
+
+    ``str(exc)`` alone can omit the URL (e.g. errors constructed by client
+    helpers), so read the status, reason, and method/URL off the attached
+    response when present.
+    """
+
+    response = exc.response
+    if response is None:
+        return str(exc)
+    request = response.request
+    target = (
+        f"{request.method} {request.url}"
+        if request is not None and request.url
+        else (response.url or "Tangle API")
+    )
+    reason = f" {response.reason}" if response.reason else ""
+    return f"HTTP {response.status_code}{reason} for {target}"
+
+
 @dataclass
 class PipelineSubmitPayload:
     """Prepared submit payload state before calling ``pipeline_runs_create``.
@@ -599,6 +621,14 @@ def fetch_logs(self, client: Any, execution_id: str) -> Any:
         """Hook for alternate TD log providers; OSS uses the Tangle API only."""
         return client.executions_container_log(execution_id)
 
+    def stream_logs(self, client: Any, execution_id: str) -> Iterable[str]:
+        """Hook for alternate TD log providers; OSS streams via the Tangle API.
+
+        Failures to open the stream must raise from this call; exceptions raised
+        while iterating are reported as mid-stream interruptions.
+        """
+        return client.iter_execution_container_log_lines(execution_id)
+
 
 @dataclass
 class PipelineRunManager(TangleCliHandler):
@@ -1261,6 +1291,38 @@ def graph_state_output(self, run_ids: list[str], *, timeout: float = 30.0) -> di
     def logs(self, execution_id: str) -> dict[str, Any]:
         return self.to_plain(self.hooks.fetch_logs(self.client, execution_id))
 
+    def stream_logs(self, execution_id: str) -> Iterable[str]:
+        try:
+            lines = self.hooks.stream_logs(self.client, execution_id)
+        except requests.HTTPError as exc:
+            # A definitive non-2xx answer to the stream-open request (404 for a
+            # missing execution or endpoint, 403, ...): keep the status and the
+            # attempted target visible, since that is what the caller acts on.
+            raise PipelineRunError(
+                f"Failed to open log stream for execution {execution_id}: "
+                f"{_describe_http_error(exc)}"
+            ) from exc
+        except requests.RequestException as exc:
+            # Non-HTTP transport failures (connection refused, timeout) raise
+            # from the open call itself; surface them as a clean open failure.
+            raise PipelineRunError(
+                f"Failed to open log stream for execution {execution_id}: {exc}"
+            ) from exc
+        return self._relabel_stream_drops(lines, execution_id)
+
+    @staticmethod
+    def _relabel_stream_drops(lines: Iterable[str], execution_id: str) -> Iterable[str]:
+        try:
+            yield from lines
+        except requests.RequestException as exc:
+            # The stream opened (the hook call succeeded), so a transport
+            # failure here is a drop of the live follow, possibly before the
+            # first line arrived. Surface it as an interruption rather than a
+            # fetch failure, which would wrongly imply the initial open failed.
+            raise PipelineRunError(
+                f"Log stream for execution {execution_id} was interrupted: {exc}"
+            ) from exc
+
     def search_runs(
         self,
         *,
diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py
index 2da0eb5..bb9052b 100644
--- a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py
+++ b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py
@@ -3,7 +3,9 @@
 from __future__ import annotations
 
 import json
+import os
 import pathlib
+import sys
 from typing import Annotated, Any
 
 from cyclopts import App, Parameter
@@ -359,6 +361,14 @@ def pipeline_runs_wait(
 def pipeline_runs_logs(
     execution_id: str | None = None,
     *,
+    stream: Annotated[
+        bool | None,
+        Parameter(
+            help="Follow the live log stream instead of fetching a one-shot snapshot. "
+            "The follow has no read timeout and stays open silently while the "
+            "container emits no output."
+        ),
+    ] = None,
     base_url: BaseUrlOption = None,
     token: TokenOption = None,
     auth_header: AuthHeaderOption = None,
@@ -369,11 +379,24 @@ def pipeline_runs_logs(
     """Print Tangle API container logs for an execution id."""
     specs = {
         "execution_id": (execution_id,),
+        "stream": (stream, None),
         "log_type": (log_type, "console"),
         **api_arg_specs(base_url=base_url, token=token, auth_header=auth_header, header=header),
     }
 
     def action(manager: PipelineRunManager, args: ArgsContainer) -> object:
+        if args.stream:
+            try:
+                for line in manager.stream_logs(args.execution_id):
+                    print(line, flush=True)
+            except BrokenPipeError:
+                # The downstream reader closed the pipe (e.g. `... | head`).
+                # Point stdout at devnull so the interpreter's exit-time flush
+                # of the closed pipe cannot raise a second BrokenPipeError.
+                devnull_fd = os.open(os.devnull, os.O_WRONLY)
+                os.dup2(devnull_fd, sys.stdout.fileno())
+                os.close(devnull_fd)
+            return None
         result = manager.logs(args.execution_id)
         if isinstance(result, dict) and isinstance(result.get("log_text"), str):
             print(result["log_text"], end="" if result["log_text"].endswith("\n") else "\n")
diff --git a/tests/test_client.py b/tests/test_client.py
index 5456ac0..881ac0f 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -1,13 +1,47 @@
 from __future__ import annotations
 
+import io
+import json
 from types import SimpleNamespace
+from typing import Any
 from unittest.mock import MagicMock
 
+import pytest
+import requests
+
 import tangle_cli.client as client_module
 from tangle_cli.client import TangleApiClient
+from tangle_cli.logger import CaptureLogger
 from tangle_cli.models import ComponentInfo
 
 
+def _response(payload: Any = None, status_code: int = 200) -> requests.Response:
+    r = requests.Response()
+    r.status_code = status_code
+    if payload is None:
+        r._content = b""
+    else:
+        r._content = json.dumps(payload).encode("utf-8")
+        r.headers["Content-Type"] = "application/json"
+    r.request = requests.Request("GET", "https://api.test").prepare()
+    return r
+
+
+class _FakeSession:
+    def __init__(self, responses: list[requests.Response | Exception] | None = None) -> None:
+        self.calls: list[dict[str, Any]] = []
+        self.responses = responses or []
+
+    def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
+        self.calls.append({"method": method, "url": url, **kwargs})
+        if self.responses:
+            next_response = self.responses.pop(0)
+            if isinstance(next_response, Exception):
+                raise next_response
+            return next_response
+        return _response({})
+
+
 def test_find_existing_components_matches_exact_names_case_insensitively() -> None:
     client = TangleApiClient("https://api.test")
     client.list_published_component_infos = MagicMock(
@@ -53,3 +87,395 @@ def fail_from_dict(*args, **kwargs):
 
     assert client.get_run_pipeline_spec("run-1") is task_spec
     client.executions_details.assert_called_once_with("root-exec-1")
+
+
+def _tracked_stream_response(raw: Any, status_code: int = 200) -> requests.Response:
+    """A streaming-style response reading from ``raw`` that records ``close()`` in ``_closed``."""
+
+    r = requests.Response()
+    r.status_code = status_code
+    r.raw = raw
+    r.headers["Content-Type"] = "text/event-stream"
+    r.request = requests.Request("GET", "https://api.test").prepare()
+    r._closed = False
+    original_close = r.close
+
+    def tracked_close() -> None:
+        r._closed = True
+        original_close()
+
+    r.close = tracked_close  # type: ignore[method-assign]
+    return r
+
+
+def _stream_response(lines: list[bytes] | None = None, status_code: int = 200) -> requests.Response:
+    body = b"\n".join(lines) if lines else b""
+    return _tracked_stream_response(io.BytesIO(body), status_code)
+
+
+def test_stream_execution_container_log_yields_lines_and_closes() -> None:
+    stream = _stream_response([b"line-1", b"line-2", b"line-3"])
+    session = _FakeSession([stream])
+    client = TangleApiClient("https://api.test", session=session)
+
+    lines = list(client.iter_execution_container_log_lines("exec-1"))
+
+    assert lines == ["line-1", "line-2", "line-3"]
+    assert stream._closed is True
+    assert session.calls[0]["url"] == "https://api.test/api/executions/exec-1/stream_container_log"
+    assert session.calls[0]["stream"] is True
+    # The follow stream keeps the connect timeout but has no per-read timeout,
+    # so a healthy stream that is idle (container quiet) is never killed.
+    assert session.calls[0]["timeout"] == (client.timeout, None)
+
+
+def test_stream_execution_container_log_closes_on_early_break() -> None:
+    stream = _stream_response([b"a", b"b", b"c"])
+    session = _FakeSession([stream])
+    client = TangleApiClient("https://api.test", session=session)
+
+    gen = client.iter_execution_container_log_lines("exec-1")
+    assert next(iter(gen)) == "a"
+    gen.close()  # type: ignore[union-attr]
+
+    assert stream._closed is True
+
+
+def test_stream_open_retries_transient_status_then_succeeds(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    bad = _stream_response(status_code=503)
+    ok = _stream_response([b"recovered"])
+    session = _FakeSession([bad, ok])
+    logger = CaptureLogger()
+    client = TangleApiClient("https://api.test", session=session, logger=logger)
+
+    lines = list(client.iter_execution_container_log_lines("exec-1"))
+
+    assert lines == ["recovered"]
+    assert bad._closed is True
+    assert sleeps == [1.0]
+    assert len(session.calls) == 2
+    # Every stream-open retry sleep is announced through the client logger.
+    assert "transient HTTP 503 opening log stream; retrying in 1.0s (attempt 2/7)" in (
+        logger.get_logs() or ""
+    )
+
+
+def test_stream_open_retries_transport_error_then_succeeds(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    ok = _stream_response([b"after-blip"])
+    calls = {"n": 0}
+
+    class FlakySession(_FakeSession):
+        def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
+            calls["n"] += 1
+            if calls["n"] == 1:
+                raise requests.ConnectionError("transient transport blip")
+            return ok
+
+    client = TangleApiClient("https://api.test", session=FlakySession())
+
+    lines = list(client.iter_execution_container_log_lines("exec-1"))
+
+    assert lines == ["after-blip"]
+    assert calls["n"] == 2
+    assert sleeps == [1.0]
+
+
+def test_stream_open_backoff_doubles_and_is_capped(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    attempts = TangleApiClient._MAX_STREAM_OPEN_ATTEMPTS
+    session = _FakeSession([_stream_response(status_code=503) for _ in range(attempts)])
+    logger = CaptureLogger()
+    client = TangleApiClient("https://api.test", session=session, logger=logger)
+
+    with pytest.raises(requests.HTTPError):
+        client.stream_execution_container_log("exec-1")
+
+    assert sleeps == [1.0, 2.0, 4.0, 8.0, 16.0, 30.0]
+    # The last retry announces the final attempt; the exhausted 7th attempt
+    # raises without announcing an 8th.
+    logs = logger.get_logs() or ""
+    assert "(attempt 7/7)" in logs
+    assert "8/7" not in logs
+
+
+def test_stream_open_raises_non_retryable_status_immediately(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    bad = _stream_response(status_code=404)
+    session = _FakeSession([bad])
+    client = TangleApiClient("https://api.test", session=session)
+
+    with pytest.raises(requests.HTTPError):
+        client.stream_execution_container_log("exec-1")
+
+    # The streamed response must be closed before the non-retryable error
+    # propagates so the open connection is not leaked.
+    assert bad._closed is True
+    assert sleeps == []
+    assert len(session.calls) == 1
+
+
+def test_stream_open_exhausts_retries_and_raises_last_status(monkeypatch) -> None:
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _seconds: None)
+    attempts = TangleApiClient._MAX_STREAM_OPEN_ATTEMPTS
+    session = _FakeSession([_stream_response(status_code=502) for _ in range(attempts)])
+    client = TangleApiClient("https://api.test", session=session)
+
+    with pytest.raises(requests.HTTPError) as exc_info:
+        client.stream_execution_container_log("exec-1")
+
+    assert exc_info.value.response.status_code == 502
+    assert len(session.calls) == attempts
+
+
+def test_stream_open_exhausts_retries_and_raises_last_transport_error(monkeypatch) -> None:
+    monkeypatch.setattr("tangle_cli.client.time.sleep", lambda _seconds: None)
+
+    class AlwaysFailingSession(_FakeSession):
+        def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
+            raise requests.ConnectionError("permanent transport failure")
+
+    client = TangleApiClient("https://api.test", session=AlwaysFailingSession())
+
+    with pytest.raises(requests.ConnectionError, match="permanent transport failure"):
+        client.stream_execution_container_log("exec-1")
+
+
+def test_stream_open_cross_origin_redirect_is_not_retried(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    redirect = _stream_response(status_code=307)
+    redirect.url = "https://api.test/api/executions/exec-1/stream_container_log"
+    redirect.headers["Location"] = "https://attacker.example/leak"
+    session = _FakeSession([redirect])
+    client = TangleApiClient("https://api.test", session=session)
+
+    with pytest.raises(requests.HTTPError, match="cross-origin redirect") as exc_info:
+        client.stream_execution_container_log("exec-1")
+
+    # Same-origin redirect protection must propagate immediately, not be retried.
+    assert sleeps == []
+    assert len(session.calls) == 1
+    # The rejected streamed response is attached to the guard error and no
+    # iterator ever receives it, so it must be closed before the error
+    # propagates to avoid leaking the open connection.
+    assert exc_info.value.response is redirect
+    assert redirect._closed is True
+
+
+def test_stream_open_too_many_redirects_is_not_retried(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+
+    def same_origin_redirect() -> requests.Response:
+        r = _stream_response(status_code=307)
+        r.url = "https://api.test/api/executions/exec-1/stream_container_log"
+        r.headers["Location"] = "/api/executions/exec-1/stream_container_log"
+        return r
+
+    redirect_calls = TangleApiClient._MAX_REDIRECTS + 1
+    responses = [same_origin_redirect() for _ in range(redirect_calls)]
+    session = _FakeSession(list(responses))
+    client = TangleApiClient("https://api.test", session=session)
+
+    with pytest.raises(requests.TooManyRedirects) as exc_info:
+        client.stream_execution_container_log("exec-1")
+
+    # One stream-open attempt that exhausts redirects; no retry of the open.
+    assert sleeps == []
+    assert len(session.calls) == redirect_calls
+    # Every streamed redirect response must be closed; the final one is
+    # attached to the guard error and must not leak.
+    assert all(r._closed is True for r in responses)
+    assert exc_info.value.response is responses[-1]
+
+
+def test_stream_open_verbose_does_not_read_streamed_body(monkeypatch) -> None:
+    monkeypatch.setenv("TANGLE_VERBOSE", "1")
+    stream = _stream_response([b"line-1", b"line-2"])
+    text_reads: list[int] = []
+    original_text = type(stream).text
+
+    def tracked_text(self: requests.Response) -> str:
+        text_reads.append(1)
+        return original_text.fget(self)  # type: ignore[attr-defined]
+
+    monkeypatch.setattr(type(stream), "text", property(tracked_text))
+    logger = CaptureLogger()
+    session = _FakeSession([stream])
+    client = TangleApiClient("https://api.test", session=session, logger=logger)
+
+    response = client.stream_execution_container_log("exec-1")
+
+    # Verbose logging must not drain the streamed body before the caller can
+    # iterate it; the log stream stays readable.
+    assert text_reads == []
+    assert response._closed is False
+    assert list(response.iter_lines()) == [b"line-1", b"line-2"]
+    logs = logger.get_logs() or ""
+    assert "<streaming body omitted>" in logs
+
+
+def test_rate_limit_retry_closes_streamed_response_before_sleep(monkeypatch) -> None:
+    closed_at_sleep: list[bool] = []
+    rate_limited = _stream_response(status_code=429)
+    rate_limited.headers["Retry-After"] = "0"
+    ok = _stream_response([b"recovered"])
+
+    def tracking_sleep(_seconds: float) -> None:
+        # Record whether the 429 stream is already closed when the rate-limit
+        # sleep runs; it must not be held open during the sleep.
+        closed_at_sleep.append(rate_limited._closed)
+
+    monkeypatch.setattr("tangle_cli.client.time.sleep", tracking_sleep)
+    closed_at_retry: list[bool] = []
+
+    class TrackingSession(_FakeSession):
+        def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
+            # Record whether the prior 429 stream was already closed by the
+            # time the successful retry is issued.
+            if self.calls:
+                closed_at_retry.append(rate_limited._closed)
+            return super().request(method, url, **kwargs)
+
+    session = TrackingSession([rate_limited, ok])
+    client = TangleApiClient("https://api.test", session=session)
+
+    response = client.stream_execution_container_log("exec-1")
+
+    assert response is ok
+    assert len(session.calls) == 2
+    # The intermediate 429 streamed response must be closed before sleeping and
+    # before the retry is issued.
+    assert closed_at_sleep == [True]
+    assert closed_at_retry == [True]
+    assert rate_limited._closed is True
+    assert list(response.iter_lines()) == [b"recovered"]
+
+
+def test_auth_refresh_closes_streamed_response_before_retry() -> None:
+    unauthorized = _stream_response(status_code=401)
+    ok = _stream_response([b"authorized"])
+    closed_at_refresh: list[bool] = []
+    closed_at_retry: list[bool] = []
+
+    class RefreshingClient(TangleApiClient):
+        def __init__(self, *args: Any, **kwargs: Any) -> None:
+            super().__init__(*args, **kwargs)
+            self.refreshes = 0
+
+        def _refresh_auth(self) -> None:
+            self.refreshes += 1
+            # On the auth-refresh triggered by the 401, the streamed 401
+            # response must already be closed (not held open during refresh).
+            if self.refreshes == 2:
+                closed_at_refresh.append(unauthorized._closed)
+            self.headers["Authorization"] = f"Bearer refreshed-{self.refreshes}"
+
+    class TrackingSession(_FakeSession):
+        def request(self, method: str, url: str, **kwargs: Any) -> requests.Response:
+            # Record whether the prior 401 stream was already closed by the
+            # time the successful retry is issued.
+            if self.calls:
+                closed_at_retry.append(unauthorized._closed)
+            return super().request(method, url, **kwargs)
+
+    session = TrackingSession([unauthorized, ok])
+    client = RefreshingClient("https://api.test", session=session)
+
+    response = client._make_request("GET", "/api/users/me", stream=True)
+
+    assert response is ok
+    assert client.refreshes == 2
+    assert len(session.calls) == 2
+    # The intermediate 401 streamed response must be closed before the auth
+    # refresh and before the retry is issued.
+    assert closed_at_refresh == [True]
+    assert closed_at_retry == [True]
+    assert unauthorized._closed is True
+    # The successful retry stream remains open and readable for the caller.
+    assert response._closed is False
+    assert list(response.iter_lines()) == [b"authorized"]
+
+
+def test_stream_open_synthetic_http_error_from_make_request_is_not_retried(monkeypatch) -> None:
+    sleeps: list[float] = []
+    monkeypatch.setattr("tangle_cli.client.time.sleep", sleeps.append)
+    client = TangleApiClient("https://api.test", session=_FakeSession())
+    calls = {"n": 0}
+
+    def fake_make_request(*args: Any, **kwargs: Any) -> requests.Response:
+        calls["n"] += 1
+        raise requests.HTTPError("redirect guard tripped")
+
+    monkeypatch.setattr(client, "_make_request", fake_make_request)
+
+    with pytest.raises(requests.HTTPError, match="redirect guard tripped"):
+        client.stream_execution_container_log("exec-1")
+
+    assert sleeps == []
+    assert calls["n"] == 1
+
+
+class _ScriptedRaw:
+    """A raw stream whose ``read`` replays scripted byte chunks/exceptions.
+
+    Each ``read`` returns the next queued ``bytes`` chunk verbatim (ignoring the
+    requested size, so a multi-byte char can be split across reads) or raises a
+    queued exception, modelling a mid-stream transport failure.
+    """
+
+    def __init__(self, chunks: list[bytes | Exception]) -> None:
+        self._chunks = list(chunks)
+
+    def read(self, _size: int = -1) -> bytes:
+        if not self._chunks:
+            return b""
+        item = self._chunks.pop(0)
+        if isinstance(item, Exception):
+            raise item
+        return item
+
+    def close(self) -> None:
+        self._chunks.clear()
+
+
+def _scripted_stream_response(chunks: list[bytes | Exception]) -> requests.Response:
+    return _tracked_stream_response(_ScriptedRaw(chunks))
+
+
+def test_stream_decodes_multibyte_char_split_across_chunks() -> None:
+    # "café" and "日本語" each contain multi-byte UTF-8 sequences; feeding the
+    # stream one byte at a time splits those sequences across chunk reads.
+    # Decoding whole lines as UTF-8 must reassemble them rather than yield
+    # replacement characters or mojibake.
+    payload = "café\n日本語\n".encode("utf-8")
+    stream = _scripted_stream_response([payload[i : i + 1] for i in range(len(payload))])
+    client = TangleApiClient("https://api.test", session=_FakeSession([stream]))
+
+    lines = list(client.iter_execution_container_log_lines("exec-1"))
+
+    assert lines == ["café", "日本語"]
+    assert stream._closed is True
+
+
+def test_stream_read_error_mid_iteration_propagates_and_closes() -> None:
+    # Once the stream is open the retry budget is spent; a transport failure
+    # during iteration must propagate (not be retried or swallowed) and the
+    # streamed response must still be closed by the iterator's finally block.
+    stream = _scripted_stream_response(
+        [b"line-1\n", requests.exceptions.ChunkedEncodingError("connection broken mid-stream")]
+    )
+    client = TangleApiClient("https://api.test", session=_FakeSession([stream]))
+
+    gen = iter(client.iter_execution_container_log_lines("exec-1"))
+    assert next(gen) == "line-1"
+    with pytest.raises(requests.exceptions.ChunkedEncodingError, match="connection broken mid-stream"):
+        next(gen)
+
+    assert stream._closed is True
diff --git a/tests/test_pipeline_runs_cli.py b/tests/test_pipeline_runs_cli.py
index b74b0ad..6d642df 100644
--- a/tests/test_pipeline_runs_cli.py
+++ b/tests/test_pipeline_runs_cli.py
@@ -1,13 +1,17 @@
 from __future__ import annotations
 
 import copy
+import io
 import json
+import os
+import sys
 from contextlib import nullcontext
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any
 
 import pytest
+import requests
 import yaml
 
 from tangle_cli import cli, pipeline_run_manager, pipeline_runs_cli
@@ -114,6 +118,10 @@ def executions_graph_execution_state(self, id: str) -> dict[str, Any]:
     def executions_container_log(self, id: str) -> dict[str, Any]:
         return {"log_text": f"logs for {id}\n"}
 
+    def iter_execution_container_log_lines(self, id: str):
+        yield f"stream {id} line 1"
+        yield f"stream {id} line 2"
+
     def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]:
         self.list_calls.append(kwargs)
         return {"pipeline_runs": [{"id": "run-1"}], "next_page_token": None}
@@ -2790,3 +2798,197 @@ def cleanup_prepared_pipeline(self, preparation, *, error=None):  # type: ignore
 
     assert cleaned == [(temp_effective_path, "Pipeline validation failed:\n  - boom")]
     assert not temp_effective_path.exists()
+
+
+def test_pipeline_runs_logs_stream_prints_lines(monkeypatch, capsys) -> None:
+    fake_client = FakeClient()
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client)
+    app = cli.build_app()
+
+    run_app(app, ["sdk", "pipeline-runs", "logs", "exec-7", "--stream"])
+
+    assert capsys.readouterr().out == "stream exec-7 line 1\nstream exec-7 line 2\n"
+
+
+def test_pipeline_runs_logs_snapshot_is_default_when_stream_absent(monkeypatch, capsys) -> None:
+    class TrackingClient(FakeClient):
+        def __init__(self) -> None:
+            super().__init__()
+            self.streamed_ids: list[str] = []
+
+        def iter_execution_container_log_lines(self, id: str):
+            self.streamed_ids.append(id)
+            yield from ()
+
+    fake_client = TrackingClient()
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client)
+    app = cli.build_app()
+
+    run_app(app, ["sdk", "pipeline-runs", "logs", "exec-1"])
+
+    assert capsys.readouterr().out == "logs for exec-1\n"
+    assert fake_client.streamed_ids == []
+
+
+def test_stream_logs_open_transport_error_is_clean() -> None:
+    class RefusingClient(FakeClient):
+        def iter_execution_container_log_lines(self, id: str):
+            raise requests.ConnectionError("connection refused")
+
+    manager = pipeline_run_manager.PipelineRunManager(client=RefusingClient())
+
+    with pytest.raises(
+        PipelineRunError, match="Failed to open log stream for execution exec-1"
+    ):
+        manager.stream_logs("exec-1")
+
+
+def test_pipeline_runs_logs_stream_open_failure_exits_cleanly(monkeypatch) -> None:
+    class RefusingClient(FakeClient):
+        def iter_execution_container_log_lines(self, id: str):
+            raise requests.ConnectionError("connection refused")
+
+    fake_client = RefusingClient()
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client)
+    app = cli.build_app()
+
+    with pytest.raises(SystemExit) as exc_info:
+        app(["sdk", "pipeline-runs", "logs", "exec-1", "--stream"])
+
+    assert "Failed to open log stream for execution exec-1" in str(exc_info.value)
+    assert "connection refused" in str(exc_info.value)
+
+
+def test_pipeline_runs_logs_stream_broken_pipe_exits_cleanly(monkeypatch) -> None:
+    # Simulates `... logs --stream | head`: the downstream reader closes the
+    # pipe, so the first stdout write raises BrokenPipeError. The command must
+    # stop following and exit cleanly instead of tracebacking.
+    fake_client = FakeClient()
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client)
+    devnull_fd = os.open(os.devnull, os.O_WRONLY)
+    write_attempts = {"count": 0}
+
+    class ClosedPipeStdout(io.TextIOBase):
+        def write(self, _text: str) -> int:
+            write_attempts["count"] += 1
+            raise BrokenPipeError
+
+        def fileno(self) -> int:
+            return devnull_fd
+
+    monkeypatch.setattr(sys, "stdout", ClosedPipeStdout())
+    app = cli.build_app()
+
+    try:
+        run_app(app, ["sdk", "pipeline-runs", "logs", "exec-7", "--stream"])
+    finally:
+        os.close(devnull_fd)
+
+    assert write_attempts["count"] == 1
+
+
+def test_pipeline_runs_logs_stream_ctrl_c_exits_cleanly(monkeypatch, capsys) -> None:
+    # An interactive Ctrl-C is the normal way to stop a live follow; it must
+    # exit with the conventional interrupt code (128 + SIGINT), not a
+    # KeyboardInterrupt traceback. cyclopts provides this by default
+    # (suppress_keyboard_interrupt); this locks that contract in for --stream.
+    class InterruptedStreamClient(FakeClient):
+        def iter_execution_container_log_lines(self, id: str):
+            yield "line-1"
+            raise KeyboardInterrupt
+
+    fake_client = InterruptedStreamClient()
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client)
+    app = cli.build_app()
+
+    with pytest.raises(SystemExit) as exc_info:
+        app(["sdk", "pipeline-runs", "logs", "exec-1", "--stream"])
+
+    assert exc_info.value.code == 130
+    assert capsys.readouterr().out == "line-1\n"
+
+
+def test_pipeline_runs_logs_help_documents_stream(capsys) -> None:
+    app = cli.build_app()
+    run_app(app, ["sdk", "pipeline-runs", "logs", "--help"])
+    assert "--stream" in capsys.readouterr().out
+
+
+def test_pipeline_runs_stream_logs_uses_client_iterator() -> None:
+    fake_client = FakeClient()
+    manager = PipelineRunManager(client=fake_client)
+
+    assert list(manager.stream_logs("exec-9")) == ["stream exec-9 line 1", "stream exec-9 line 2"]
+
+
+def test_pipeline_runs_stream_logs_midstream_error_is_interruption() -> None:
+    # A transport failure after some lines have flowed must surface as a
+    # mid-stream interruption, not the "failed to open" wording reserved for a
+    # stream that never opened.
+    class MidStreamDropClient(FakeClient):
+        def iter_execution_container_log_lines(self, id: str):
+            yield "line-1"
+            raise requests.exceptions.ChunkedEncodingError("connection broken mid-stream")
+
+    manager = PipelineRunManager(client=MidStreamDropClient())
+
+    gen = iter(manager.stream_logs("exec-9"))
+    assert next(gen) == "line-1"
+    with pytest.raises(PipelineRunError, match="interrupted") as exc_info:
+        next(gen)
+    assert "Failed to open" not in str(exc_info.value)
+
+
+def test_pipeline_runs_stream_logs_zero_line_drop_is_interruption() -> None:
+    # A connection that opens successfully but drops before the first line is
+    # still a mid-stream interruption; only open failures (raised from the
+    # hook call itself) get the "failed to open" wording.
+    class ZeroLineDropClient(FakeClient):
+        def iter_execution_container_log_lines(self, id: str):
+            raise requests.exceptions.ChunkedEncodingError("Response ended prematurely")
+            yield  # pragma: no cover
+
+    manager = PipelineRunManager(client=ZeroLineDropClient())
+
+    with pytest.raises(PipelineRunError, match="interrupted") as exc_info:
+        list(manager.stream_logs("exec-9"))
+    assert "Failed to open" not in str(exc_info.value)
+
+
+def test_pipeline_runs_logs_stream_honors_log_type_none(monkeypatch, capsys) -> None:
+    # --log-type controls the CLI's diagnostic logger and is orthogonal to
+    # --stream; the streamed log content still reaches stdout under any mode.
+    fake_client = FakeClient()
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client)
+    app = cli.build_app()
+
+    run_app(app, ["sdk", "pipeline-runs", "logs", "exec-7", "--stream", "--log-type", "none"])
+
+    assert capsys.readouterr().out == "stream exec-7 line 1\nstream exec-7 line 2\n"
+
+
+def _logs_http_error(status: int, path: str) -> requests.HTTPError:
+    response = requests.Response()
+    response.status_code = status
+    response.request = requests.Request("GET", f"https://api.test{path}").prepare()
+    return requests.HTTPError(f"{status} Client Error", response=response)
+
+
+def test_pipeline_runs_logs_stream_missing_endpoint_is_clean(monkeypatch) -> None:
+    class NoStreamClient(FakeClient):
+        def iter_execution_container_log_lines(self, id: str):
+            # Like the real client, open failures raise from the call itself
+            # rather than on first iteration.
+            raise _logs_http_error(404, f"/api/executions/{id}/stream_container_log")
+
+    fake_client = NoStreamClient()
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client)
+    app = cli.build_app()
+
+    # A clean SystemExit (not a raw requests.HTTPError) means no traceback.
+    with pytest.raises(SystemExit) as exc_info:
+        app(["sdk", "pipeline-runs", "logs", "exec-1", "--stream"])
+
+    message = str(exc_info.value)
+    assert "404" in message
+    assert "/api/executions/exec-1/stream_container_log" in message