feat(pipeline-run): surface API HTTP failures as clean errors

packages/tangle-cli/src/tangle_cli/cli_helpers.py

@@ -4,10 +4,64 @@
 
 import json
 import pathlib
+from collections.abc import Iterator
+from contextlib import contextmanager
 from typing import Any
 
+import requests
+
 from .args_container import ArgsContainer, ConfigFileError
 
+_HTTP_ERROR_BODY_LIMIT = 2000
+
+
+def format_http_error(exc: requests.HTTPError) -> str:
+    """Render an HTTP status failure as a concise CLI message for SDK commands.
+
+    SDK/static client calls raise ``requests.HTTPError`` for non-2xx responses
+    (via ``raise_for_status``). Client-internal helpers handle the statuses they
+    can recover from (e.g. the 404 run-id -> execution-id fallback) and re-raise
+    the rest, so the command layer surfaces the remaining errors here instead of
+    letting a raw traceback reach the interpreter. The response status, reason,
+    attempted method/URL, and body are preserved as that context is what a caller
+    needs to act on; the body is collapsed to a single line and truncated so the
+    message stays one line. Only HTTP status failures are formatted here;
+    connection/timeout errors carry no response and propagate unchanged.
+    """
+
+    response = exc.response
+    if response is None:
+        return f"Tangle API request failed: {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 ""
+    summary = f"Tangle API request failed ({response.status_code}{reason}) for {target}"
+    body = " ".join(response.text.split())
+    if not body:
+        return summary
+    if len(body) > _HTTP_ERROR_BODY_LIMIT:
+        body = f"{body[:_HTTP_ERROR_BODY_LIMIT]}... (truncated)"
+    return f"{summary}: {body}"
+
+
+@contextmanager
+def surface_http_errors(error_type: type[Exception]) -> Iterator[None]:
+    """Re-raise ``requests.HTTPError`` as ``error_type`` with a formatted message.
+
+    Client-internal recovery runs first and re-raises only the statuses it
+    cannot handle, so errors reaching here are the ones the command layer must
+    surface as a clean nonzero exit rather than a raw traceback.
+    """
+
+    try:
+        yield
+    except requests.HTTPError as exc:
+        raise error_type(format_http_error(exc)) from exc
+
 
 def load_args_or_exit(config: str | None, **kwargs: Any) -> list[ArgsContainer]:
     """Load ArgsContainer values from CLI/config specs, exiting with CLI errors."""

packages/tangle-cli/src/tangle_cli/handler.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 from collections.abc import Callable, Mapping
+from contextlib import AbstractContextManager
 from typing import Any
 
 from .api_transport import default_base_url
+from .cli_helpers import surface_http_errors
 from .logger import Logger, get_default_logger
 
 
@@ -94,3 +96,13 @@ def _require_client(self) -> Any:
         if client is None:
             raise self._required_client_error_type(self._required_client_error_message)
         return client
+
+    def _http_error_type(self) -> type[Exception]:
+        """Exception type used to re-raise a formatted API HTTP failure."""
+
+        return self._required_client_error_type
+
+    def _surface_http_errors(self) -> AbstractContextManager[None]:
+        """Context manager re-raising API ``HTTPError`` as concise, formatted failures."""
+
+        return surface_http_errors(self._http_error_type())

packages/tangle-cli/src/tangle_cli/pipeline_run_annotations.py

@@ -5,11 +5,14 @@
 from typing import Any
 
 from .handler import TangleCliHandler
+from .pipeline_run_manager import PipelineRunError
 
 
 class AnnotationManager(TangleCliHandler):
     """Manage annotations on Tangle pipeline runs."""
 
+    _required_client_error_type = PipelineRunError
+
     @staticmethod
     def to_plain(value: Any) -> Any:
         if hasattr(value, "to_dict"):
@@ -19,7 +22,8 @@ def to_plain(value: Any) -> Any:
         return value
 
     def list_annotations(self, run_id: str) -> dict[str, Any]:
-        annotations = self.to_plain(self._require_client().pipeline_runs_annotations(run_id)) or {}
+        with self._surface_http_errors():
+            annotations = self.to_plain(self._require_client().pipeline_runs_annotations(run_id)) or {}
         if not isinstance(annotations, dict):
             annotations = dict(annotations)
         return {
@@ -30,11 +34,13 @@ def list_annotations(self, run_id: str) -> dict[str, Any]:
         }
 
     def set_annotation(self, run_id: str, key: str, value: Any = None) -> dict[str, Any]:
-        self._require_client().pipeline_runs_put_annotations(run_id, key, value=value)
+        with self._surface_http_errors():
+            self._require_client().pipeline_runs_put_annotations(run_id, key, value=value)
         return {"status": "success", "run_id": run_id, "key": key, "value": value}
 
     def delete_annotation(self, run_id: str, key: str) -> dict[str, Any]:
-        self._require_client().pipeline_runs_delete_annotations(run_id, key)
+        with self._surface_http_errors():
+            self._require_client().pipeline_runs_delete_annotations(run_id, key)
         return {"status": "success", "run_id": run_id, "key": key}
 
 

packages/tangle-cli/src/tangle_cli/pipeline_run_details.py

@@ -11,6 +11,9 @@
 from concurrent.futures import TimeoutError as FutureTimeoutError
 from typing import Any
 
+import requests
+
+from .cli_helpers import format_http_error
 from .handler import TangleCliHandler
 
 
@@ -164,6 +167,8 @@ def get_graph_state_output(
                     results.append(future.result(timeout=timeout))
                 except FutureTimeoutError:
                     results.append(_error_result(run_id, f"timeout after {timeout}s"))
+                except requests.HTTPError as exc:
+                    results.append(_error_result(run_id, format_http_error(exc)))
                 except Exception as exc:
                     results.append(_error_result(run_id, str(exc)))
             finally:

packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py

@@ -617,6 +617,9 @@ def __post_init__(self) -> None:
         if self.hooks is not self:
             setattr(self.hooks, "client", self.client)
 
+    def _http_error_type(self) -> type[Exception]:
+        return PipelineRunError
+
     @staticmethod
     def to_plain(value: Any) -> Any:
         if isinstance(value, Mapping):
@@ -1137,7 +1140,8 @@ def submit_prepared_body(
         self.hooks.before_submit_context(submit_context)
         client = self._require_client()
         try:
-            response = self.to_plain(client.pipeline_runs_create(body=body))
+            with self._surface_http_errors():
+                response = self.to_plain(client.pipeline_runs_create(body=body))
         except Exception as exc:
             if notify_submit_error:
                 self.hooks.on_submit_error(exc, context=submit_context)
@@ -1224,12 +1228,13 @@ def submit_pipeline(
         return self.submit_prepared_payload(payload, pipeline_path=pipeline_path, attempt=attempt)
 
     def get_run(self, run_id: str, *, include_execution_stats: bool = True) -> dict[str, Any]:
-        return self.to_plain(
-            self.client.pipeline_runs_get(
-                run_id,
-                include_execution_stats=include_execution_stats,
+        with self._surface_http_errors():
+            return self.to_plain(
+                self.client.pipeline_runs_get(
+                    run_id,
+                    include_execution_stats=include_execution_stats,
+                )
             )
-        )
 
     def get_run_details(
         self,
@@ -1240,26 +1245,33 @@ def get_run_details(
         include_implementations: bool = False,
         execution_id: str | None = None,
     ) -> dict[str, Any]:
-        return PipelineRunDetails(client=self.client).get_run_details_output(
-            run_id,
-            include_implementations=include_implementations,
-            include_annotations=include_annotations,
-            include_execution_state=include_execution_state,
-            execution_id=execution_id,
-        )
+        with self._surface_http_errors():
+            return PipelineRunDetails(client=self.client).get_run_details_output(
+                run_id,
+                include_implementations=include_implementations,
+                include_annotations=include_annotations,
+                include_execution_state=include_execution_state,
+                execution_id=execution_id,
+            )
 
     def cancel_run(self, run_id: str) -> dict[str, Any]:
-        return self.to_plain(self.client.pipeline_runs_cancel(run_id)) or {"id": run_id, "cancelled": True}
+        with self._surface_http_errors():
+            cancelled = self.to_plain(self.client.pipeline_runs_cancel(run_id))
+        return cancelled or {"id": run_id, "cancelled": True}
 
     def graph_state(self, execution_id: str) -> Mapping[str, Any] | Any:
-        graph_state = self.client.executions_graph_execution_state(execution_id)
+        with self._surface_http_errors():
+            graph_state = self.client.executions_graph_execution_state(execution_id)
         return self.to_plain(graph_state)
 
     def graph_state_output(self, run_ids: list[str], *, timeout: float = 30.0) -> dict[str, Any]:
+        # Per-run failures are reported in each result's "error" field rather
+        # than raised, so no HTTP-error surfacing is needed at this boundary.
         return PipelineRunDetails(client=self.client).get_graph_state_output(run_ids, timeout=timeout)
 
     def logs(self, execution_id: str) -> dict[str, Any]:
-        return self.to_plain(self.hooks.fetch_logs(self.client, execution_id))
+        with self._surface_http_errors():
+            return self.to_plain(self.hooks.fetch_logs(self.client, execution_id))
 
     def search_runs(
         self,
@@ -1270,15 +1282,16 @@ def search_runs(
         include_pipeline_names: bool | None = None,
         include_execution_stats: bool | None = True,
     ) -> dict[str, Any]:
-        return self.to_plain(
-            self.client.pipeline_runs_list(
-                page_token=page_token,
-                filter=filter,
-                filter_query=filter_query,
-                include_pipeline_names=include_pipeline_names,
-                include_execution_stats=include_execution_stats,
+        with self._surface_http_errors():
+            return self.to_plain(
+                self.client.pipeline_runs_list(
+                    page_token=page_token,
+                    filter=filter,
+                    filter_query=filter_query,
+                    include_pipeline_names=include_pipeline_names,
+                    include_execution_stats=include_execution_stats,
+                )
             )
-        )
 
     def search_pipeline_runs(
         self,
@@ -1293,17 +1306,18 @@ def search_pipeline_runs(
         limit: int = 10,
         page_token: str | None = None,
     ) -> dict[str, Any]:
-        return PipelineRunSearch(client=self.client, logger=self.logger).search(
-            name=name,
-            created_by=created_by,
-            annotations=annotations,
-            start_date=start_date,
-            end_date=end_date,
-            local_time=local_time,
-            query=query,
-            limit=limit,
-            page_token=page_token,
-        )
+        with self._surface_http_errors():
+            return PipelineRunSearch(client=self.client, logger=self.logger).search(
+                name=name,
+                created_by=created_by,
+                annotations=annotations,
+                start_date=start_date,
+                end_date=end_date,
+                local_time=local_time,
+                query=query,
+                limit=limit,
+                page_token=page_token,
+            )
 
     def export_run(
         self,
@@ -1312,7 +1326,8 @@ def export_run(
         *,
         dehydrate: bool = False,
     ) -> dict[str, Any]:
-        task_spec = self.client.get_run_pipeline_spec(run_id)
+        with self._surface_http_errors():
+            task_spec = self.client.get_run_pipeline_spec(run_id)
         if task_spec is None:
             raise PipelineRunError(f"No pipeline spec found for run {run_id}")
         raw = getattr(task_spec, "raw", None)
@@ -1328,12 +1343,13 @@ def export_run(
         if dehydrate and output is None:
             raise PipelineRunError("--dehydrate requires --output")
         if dehydrate:
-            spec = PipelineDehydrator(
-                remembered_choices={"": DehydrateChoice.AUTO},
-                output_file=output,
-                client=self.client,
-                logger=self.logger,
-            ).dehydrate(spec)
+            with self._surface_http_errors():
+                spec = PipelineDehydrator(
+                    remembered_choices={"": DehydrateChoice.AUTO},
+                    output_file=output,
+                    client=self.client,
+                    logger=self.logger,
+                ).dehydrate(spec)
         content = dump_yaml(spec)
         if output is None:
             return {"run_id": run_id, "pipeline": spec, "yaml": content, "dehydrated": dehydrate}

packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py

@@ -120,7 +120,10 @@ def _run_annotation_action(config: str | None, cli_base_url: str | None, specs:
                 client=_api_client(args, cli_base_url=cli_base_url, command_name="pipeline-run annotation commands"),
                 logger=logger,
             )
-            print_json(fn(manager, args))
+            try:
+                print_json(fn(manager, args))
+            except PipelineRunError as exc:
+                raise SystemExit(str(exc)) from exc
         finally:
             finalize_logs()