TangleML · arseniy-pplx · Jul 13, 2026
@@ -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."""

@@ -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,
         *,

@@ -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")