From 2e33c7c6e81a1a7eb176bcc965529845fba5f28f Mon Sep 17 00:00:00 2001 From: Arseniy Dugin Date: Mon, 13 Jul 2026 13:43:27 +0000 Subject: [PATCH] feat(pipeline-run): add task-status and task-wait commands task-status prints the task-name to status map for a root execution by walking its child executions; container state falls back to graph-state reduction for nested pipelines. task-wait polls until every task is terminal. SUCCEEDED and SKIPPED are non-failure terminals; failures exit 2 with a stderr summary and the full status map on stdout. Wait bounds are validated up front and the deadline applies while resolving children. Transient 5xx on the per-poll root details fetch are retried within the wait deadline instead of aborting the wait; unbounded waits give up after a few consecutive failures. Also adjusts the shared status reducer used by the existing run-level status/wait: WAITING_FOR_UPSTREAM and UNINITIALIZED now count as active so a run is not reported terminal while a task still waits upstream, mixed terminal aggregates reduce to the failure terminal ahead of SKIPPED/SUCCEEDED, and wait bounds reject NaN/inf. --- .../src/tangle_cli/pipeline_run_manager.py | 313 +++++++++- .../src/tangle_cli/pipeline_runs_cli.py | 79 +++ tests/test_pipeline_runs_cli.py | 558 ++++++++++++++++++ 3 files changed, 940 insertions(+), 10 deletions(-) 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..4cab2d9 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py @@ -12,6 +12,7 @@ import copy import inspect import json +import math import re import time import uuid @@ -21,6 +22,7 @@ from pathlib import Path from typing import Any, Mapping +import requests import yaml from .handler import TangleCliHandler @@ -32,9 +34,42 @@ from .pipeline_run_search import PipelineRunSearch from .utils import dump_yaml -_TERMINAL_STATUSES = ("FAILED", "SYSTEM_ERROR", "CANCELLED", "CANCELED", "SKIPPED", "SUCCEEDED", "INVALID") -_ACTIVE_STATUSES = ("RUNNING", "CANCELLING", "CANCELING", "PENDING", "QUEUED") +# Precedence order for reducing a mixed terminal aggregate to a single status: +# failure terminals must precede the non-failure terminals SKIPPED and SUCCEEDED +# so a mixed aggregate (e.g. {SKIPPED, INVALID}) reduces to the failure instead +# of masking it (a fully-SKIPPED map still reduces to SKIPPED). +_TERMINAL_PRECEDENCE = ("FAILED", "SYSTEM_ERROR", "CANCELLED", "CANCELED", "INVALID", "SKIPPED", "SUCCEEDED") +# Membership set for is_terminal_status. +_TERMINAL_STATUSES = frozenset(_TERMINAL_PRECEDENCE) +# Nonterminal statuses outrank any terminal count when reducing an aggregate; +# WAITING_FOR_UPSTREAM/UNINITIALIZED are backend enum members and nonterminal. +_ACTIVE_STATUSES = ( + "RUNNING", + "CANCELLING", + "CANCELING", + "PENDING", + "QUEUED", + "WAITING_FOR_UPSTREAM", + "UNINITIALIZED", +) _FAILURE_EARLY_EXIT_STATUSES = ("FAILED", "SYSTEM_ERROR") +# The only terminals that are not per-task failures; every other terminal makes +# a root-execution `task-wait` exit non-zero. A single SKIPPED task (e.g. a +# conditional branch) is not a failure, unlike a run whose reduced status is +# SKIPPED at the run level. +_NON_FAILURE_TERMINAL_STATUSES = frozenset({"SKIPPED", "SUCCEEDED"}) +_TASK_FAILURE_STATUSES = _TERMINAL_STATUSES - _NON_FAILURE_TERMINAL_STATUSES +# A child execution's status is unknown until its state row is written; 404 +# means "wrong endpoint type" (container vs graph) and transient 5xx can be +# returned while the orchestrator is still settling. Both fall through to the +# next endpoint / report UNKNOWN so the wait loop polls again. +_RETRYABLE_EXEC_STATE_STATUSES = frozenset({404, 500, 502, 503, 504}) +# An unbounded task-wait (max_wait=None) has no deadline to stop retrying a +# failing root details fetch, so give up after this many consecutive server +# errors instead of hiding a persistent outage behind endless retries. +_MAX_UNBOUNDED_ROOT_FETCH_FAILURES = 3 +_UNKNOWN_TASK_STATUS = "UNKNOWN" +_ROOT_TASK_NAME = "root" _EXECUTION_STATE_TIMINGS_METADATA_KEY = "execution_state_timings" _EXECUTION_STATE_TIMING_MONOTONIC_METADATA_KEY = "_execution_state_timing_monotonic" _SUBMISSION_ID_ANNOTATION_KEY = "tangle-cli/submission-id" @@ -54,6 +89,51 @@ class AmbiguousPipelineRunRecoveryError(PipelineRunError): """Raised when submit recovery finds multiple runs for one submission id.""" +class TransientServerError(PipelineRunError): + """An HTTP 5xx failure that a deadline-bounded polling loop may retry. + + One-shot callers (e.g. ``task-status``) still fail fast on it like any + other PipelineRunError. + """ + + +def _transport_error(context: str, exc: requests.RequestException) -> PipelineRunError: + """Wrap a transport/HTTP failure as a clean CLI-surfaced PipelineRunError. + + Root-execution status resolution talks to the API directly, so connection + resets/timeouts and non-retryable 4xx/5xx (including 429) would otherwise + escape task-status/task-wait as raw ``requests`` tracebacks. The CLI maps + PipelineRunError to a non-zero exit with the message instead. Server errors + are typed TransientServerError so polling loops can retry them. + """ + + response = getattr(exc, "response", None) + status_code = getattr(response, "status_code", None) + if status_code is not None: + message = f"{context}: request failed with HTTP {status_code}" + if status_code >= 500: + return TransientServerError(message) + return PipelineRunError(message) + return PipelineRunError(f"{context}: {exc}") + + +class TaskStatusesFailed(PipelineRunError): + """Raised when a root execution reaches terminal state with failed tasks. + + Carries the full final ``{task_name: status}`` map alongside the ``failures`` + subset so callers can still print the complete map while reporting failures. + """ + + def __init__(self, root_execution_id: str, statuses: dict[str, str], failures: dict[str, str]): + self.root_execution_id = root_execution_id + self.statuses = statuses + self.failures = failures + summary = ", ".join(f"{name}={status}" for name, status in sorted(failures.items())) + super().__init__( + f"Root execution {root_execution_id} had {len(failures)} failed task(s): {summary}" + ) + + @dataclass class PipelineSubmitPayload: """Prepared submit payload state before calling ``pipeline_runs_create``. @@ -817,11 +897,30 @@ def status_from_counts(status_counts: Mapping[str, int]) -> str | None: for status in _ACTIVE_STATUSES: if int(status_counts.get(status, 0) or 0) > 0: return status - for status in _TERMINAL_STATUSES: + for status in _TERMINAL_PRECEDENCE: if int(status_counts.get(status, 0) or 0) > 0: return status return None + @staticmethod + def _status_from_container_state(state: Mapping[str, Any] | Any) -> str | None: + plain = PipelineRunManager.to_plain(state) + if isinstance(plain, Mapping): + status = plain.get("status") + if isinstance(status, str) and status: + return status + return None + + @classmethod + def _status_from_graph_state(cls, state: Mapping[str, Any] | Any) -> str | None: + # A nested graph/subpipeline execution carries no direct ``status``; its + # status is derived from child execution counts the same way run-level + # graph polling does. + direct = cls._status_from_container_state(state) + if direct is not None: + return direct + return cls.status_from_counts(cls.status_counts_from_graph_state(cls.to_plain(state))) + @staticmethod def status_from_run(run: Mapping[str, Any]) -> str | None: summary = run.get("execution_summary") @@ -839,7 +938,7 @@ def status_from_run(run: Mapping[str, Any]) -> str | None: for status in _ACTIVE_STATUSES: if int(stats.get(status, 0) or 0) > 0: return status - for status in _TERMINAL_STATUSES: + for status in _TERMINAL_PRECEDENCE: if int(stats.get(status, 0) or 0) > 0: return status return None @@ -1261,6 +1360,174 @@ 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 _exec_status(self, execution_id: str, *, deadline: float | None = None) -> str | None: + """Resolve a single execution's status, trying container then graph state. + + 404 means the execution is the other kind (container vs graph); a + transient 5xx means the state row is still being written. Both cases + fall through to the next endpoint and, if neither resolves, return + ``None`` so the caller reports UNKNOWN. Once past ``deadline`` (a + ``time.monotonic()`` timestamp) no further requests are issued, so one + poll cannot overrun the caller's wait budget. + """ + + endpoints = ( + (self.client.executions_container_state, self._status_from_container_state), + (self.client.executions_graph_execution_state, self._status_from_graph_state), + ) + context = f"Resolving status for execution {execution_id}" + last_server_error: int | None = None + for endpoint, extract_status in endpoints: + if deadline is not None and time.monotonic() >= deadline: + return None + try: + state = endpoint(execution_id) + except requests.HTTPError as exc: + response = exc.response + if response is not None and response.status_code in _RETRYABLE_EXEC_STATE_STATUSES: + if response.status_code >= 500: + last_server_error = response.status_code + continue + raise _transport_error(context, exc) from exc + except requests.RequestException as exc: + raise _transport_error(context, exc) from exc + status = extract_status(state) + if status is not None: + return status + if last_server_error is not None: + # Without this a wait that times out reports UNKNOWN with no hint + # that the state API was failing the whole time. + self.logger.warn( + f"Status for execution {execution_id} is UNKNOWN: state endpoints " + f"kept failing with HTTP {last_server_error}" + ) + return None + + def task_statuses( + self, root_execution_id: str, *, deadline: float | None = None + ) -> dict[str, str]: + """Return a ``{task_name: status}`` map for a root execution. + + Walks ``child_task_execution_ids`` from the root execution details and + resolves each child's status. A leaf/root-only execution (no children) + reports a single ``{"root": status}`` entry; unresolved child statuses + are reported as ``UNKNOWN``. A missing root id (404) is a clean + not-found error; other transport/HTTP failures surface cleanly too. + Past ``deadline`` (a ``time.monotonic()`` timestamp) remaining statuses + are reported as ``UNKNOWN`` without further requests. + """ + + try: + details = self.to_plain(self.client.executions_details(root_execution_id)) + except requests.HTTPError as exc: + response = exc.response + if response is not None and response.status_code == 404: + raise PipelineRunError(f"Root execution {root_execution_id} not found") from exc + raise _transport_error(f"Fetching root execution {root_execution_id}", exc) from exc + except requests.RequestException as exc: + raise _transport_error(f"Fetching root execution {root_execution_id}", exc) from exc + children = details.get("child_task_execution_ids") if isinstance(details, Mapping) else None + if not isinstance(children, Mapping) or not children: + return { + _ROOT_TASK_NAME: self._exec_status(root_execution_id, deadline=deadline) + or _UNKNOWN_TASK_STATUS + } + return { + str(task_name): self._child_status(execution_id, deadline=deadline) + for task_name, execution_id in children.items() + } + + def _child_status(self, execution_id: Any, *, deadline: float | None = None) -> str: + # A None/empty/whitespace-only child id has no execution to query; + # report UNKNOWN instead of polling a literal blank id forever. + child_id = str(execution_id).strip() if execution_id else "" + if not child_id: + return _UNKNOWN_TASK_STATUS + return self._exec_status(child_id, deadline=deadline) or _UNKNOWN_TASK_STATUS + + def wait_for_task_statuses( + self, + root_execution_id: str, + *, + max_wait: float | None = 1800.0, + poll_interval: float = 5.0, + allow_failure: bool = False, + ) -> dict[str, str]: + """Poll ``task_statuses`` until every task is terminal. + + Returns the final ``{task_name: status}`` map. Raises + ``TaskStatusesFailed`` when any task ends in a failure status unless + ``allow_failure`` is set, and ``PipelineRunError`` on timeout. + ``max_wait=None`` waits indefinitely. + """ + + self._validate_wait_params( + max_wait=max_wait, + poll_interval=poll_interval, + timeout_clock="monotonic", + allow_zero_poll_interval=False, + ) + deadline = None if max_wait is None else time.monotonic() + max_wait + root_fetch_failures = 0 + while True: + try: + statuses = self.task_statuses(root_execution_id, deadline=deadline) + except TransientServerError as exc: + # A 5xx on the root details fetch is tolerated like child-state + # 5xx (retried next poll) so one blip cannot abort a long wait. + # The deadline still bounds the retries; an unbounded wait gives + # up after a few consecutive failures instead of masking a + # persistent outage forever. + root_fetch_failures += 1 + if deadline is not None and time.monotonic() >= deadline: + raise + if deadline is None and root_fetch_failures >= _MAX_UNBOUNDED_ROOT_FETCH_FAILURES: + raise + if root_fetch_failures == 1: + # Warn once per outage, not once per poll, so a long outage + # does not flood the logs. + self.logger.warn(f"{exc}; retrying on the next poll") + self._sleep_before_next_poll(poll_interval, deadline) + continue + root_fetch_failures = 0 + if statuses and all(self.is_terminal_status(status) for status in statuses.values()): + return self._handle_terminal_task_statuses(root_execution_id, statuses, allow_failure) + if deadline is not None and time.monotonic() >= deadline: + pending = ", ".join( + f"{name}={status}" + for name, status in sorted(statuses.items()) + if not self.is_terminal_status(status) + ) + raise PipelineRunError( + f"Root execution {root_execution_id} still has non-terminal tasks " + f"after {max_wait:g}s: {pending}" + ) + self._sleep_before_next_poll(poll_interval, deadline) + + @staticmethod + def _sleep_before_next_poll(poll_interval: float, deadline: float | None) -> None: + if deadline is None: + time.sleep(poll_interval) + return + # Clamp the final sleep to the remaining budget so the loop times out + # at max_wait instead of overshooting by up to a poll interval. + time.sleep(min(poll_interval, max(0.0, deadline - time.monotonic()))) + + def _handle_terminal_task_statuses( + self, + root_execution_id: str, + statuses: dict[str, str], + allow_failure: bool, + ) -> dict[str, str]: + failures = { + name: status + for name, status in statuses.items() + if status.upper() in _TASK_FAILURE_STATUSES + } + if failures and not allow_failure: + raise TaskStatusesFailed(root_execution_id, statuses, failures) + return statuses + def search_runs( self, *, @@ -1452,6 +1719,32 @@ def _poll_run_status( execution_state_timings=execution_state_timings, ) + @staticmethod + def _validate_wait_params( + *, + max_wait: float | None, + poll_interval: float, + timeout_clock: str, + allow_zero_poll_interval: bool, + ) -> None: + """Validate wait/poll parameters, raising before any run is submitted.""" + + # NaN/inf pass the sign checks below (NaN compares False, inf is + # "positive") and would become a never-firing deadline or a raw + # ValueError from time.sleep; reject them up front. An unbounded wait is + # spelled max_wait=None, not an explicit inf. + if max_wait is not None: + if not math.isfinite(max_wait): + raise PipelineRunError("--max-wait must be a finite number") + if max_wait < 0: + raise PipelineRunError("--max-wait must be non-negative") + if not math.isfinite(poll_interval): + raise PipelineRunError("--poll-interval must be a finite number") + if poll_interval < 0 or (poll_interval == 0 and not allow_zero_poll_interval): + raise PipelineRunError("--poll-interval must be positive") + if timeout_clock not in {"monotonic", "wall"}: + raise PipelineRunError("timeout_clock must be 'monotonic' or 'wall'") + def wait_for_completion( self, run_id: str, @@ -1467,12 +1760,12 @@ def wait_for_completion( wait_context = context or PipelineRunContext(run_id=run_id, start_time=time.time()) if exit_on_first_failure: wait_context.metadata["exit_on_first_failure"] = True - if max_wait is not None and max_wait < 0: - raise PipelineRunError("--max-wait must be non-negative") - if poll_interval < 0 or (poll_interval == 0 and not allow_zero_poll_interval): - raise PipelineRunError("--poll-interval must be positive") - if timeout_clock not in {"monotonic", "wall"}: - raise PipelineRunError("timeout_clock must be 'monotonic' or 'wall'") + self._validate_wait_params( + max_wait=max_wait, + poll_interval=poll_interval, + timeout_clock=timeout_clock, + allow_zero_poll_interval=allow_zero_poll_interval, + ) enforce_max_wait = max_wait is not None and self.hooks.should_enforce_max_wait(wait_context) poll_started_at = time.monotonic() deadline_now: Callable[[], float] = time.time if timeout_clock == "wall" else time.monotonic 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..33d50c7 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_runs_cli.py @@ -4,6 +4,7 @@ import json import pathlib +import sys from typing import Annotated, Any from cyclopts import App, Parameter @@ -32,6 +33,7 @@ PipelineRunError, PipelineRunHooks, PipelineRunManager, + TaskStatusesFailed, parse_json_or_key_values, parse_key_value_entries, ) @@ -355,6 +357,83 @@ def pipeline_runs_wait( ) +@app.command(name="task-status") +def pipeline_runs_task_status( + root_execution_id: str | None = None, + *, + base_url: BaseUrlOption = None, + token: TokenOption = None, + auth_header: AuthHeaderOption = None, + header: HeaderOption = None, + config: ConfigOption = None, + log_type: LogTypeOption = "console", +) -> None: + """Print the {task_name: status} map for a root execution id, without polling. + + Walks the root execution's child task executions. A leaf/root-only execution + reports a single ``{"root": status}`` entry; unresolved children are UNKNOWN. + """ + specs = { + "root_execution_id": (root_execution_id,), + "log_type": (log_type, "console"), + **api_arg_specs(base_url=base_url, token=token, auth_header=auth_header, header=header), + } + _run_manager_action( + config, + base_url, + specs, + lambda manager, args: manager.task_statuses(args.root_execution_id), + ) + + +@app.command(name="task-wait") +def pipeline_runs_task_wait( + root_execution_id: str | None = None, + *, + max_wait: float = 1800.0, + poll_interval: float = 5.0, + allow_failure: bool = False, + base_url: BaseUrlOption = None, + token: TokenOption = None, + auth_header: AuthHeaderOption = None, + header: HeaderOption = None, + config: ConfigOption = None, + log_type: LogTypeOption = "console", +) -> None: + """Poll a root execution's task-status map until all tasks are terminal. + + Prints the final {task_name: status} map. Exit codes: 0 when every task + ends in a non-failure terminal status, 2 when any task fails and + --allow-failure is not set (the map is still printed), and non-zero on + timeout or API errors. + """ + specs = { + "root_execution_id": (root_execution_id,), + "max_wait": (max_wait, 1800.0), + "poll_interval": (poll_interval, 5.0), + "allow_failure": (allow_failure, False), + "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) -> dict[str, Any] | None: + try: + return manager.wait_for_task_statuses( + args.root_execution_id, + max_wait=float(args.max_wait), + poll_interval=float(args.poll_interval), + allow_failure=bool(args.allow_failure), + ) + except TaskStatusesFailed as exc: + # Print the full final map (failures retained on the exception drive + # the non-zero exit code) so succeeded/skipped tasks are not dropped. + print(str(exc), file=sys.stderr) + print_json(exc.statuses) + raise SystemExit(2) from exc + + _run_manager_action(config, base_url, specs, action) + + @app.command(name="logs") def pipeline_runs_logs( execution_id: str | None = None, diff --git a/tests/test_pipeline_runs_cli.py b/tests/test_pipeline_runs_cli.py index b74b0ad..bb348b3 100644 --- a/tests/test_pipeline_runs_cli.py +++ b/tests/test_pipeline_runs_cli.py @@ -2,16 +2,22 @@ import copy import json +import math +import time 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 +from tangle_cli.logger import CaptureLogger from tangle_cli.pipeline_run_manager import ( + _TASK_FAILURE_STATUSES, + _TERMINAL_STATUSES, AmbiguousPipelineRunRecoveryError, PipelineRunContext, PipelineRunError, @@ -19,6 +25,8 @@ PipelineRunManager, PipelineWaitOutcome, PipelineWaitPoll, + TaskStatusesFailed, + TransientServerError, ) from tangle_cli.pipeline_runner import PipelineRunner, PipelineRunnerHooks @@ -2790,3 +2798,553 @@ 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 _http_error(status_code: int) -> requests.HTTPError: + response = requests.Response() + response.status_code = status_code + return requests.HTTPError(response=response) + + +class ExecutionStateClient: + """Configurable fake for the root-execution task-map endpoints. + + ``details`` maps a root execution id to its child_task_execution_ids; each + child/leaf id is resolved against ``container_state`` then + ``graph_execution_state``. Endpoint values may be a status dict, an HTTP + error to raise, or absent (KeyError-style 404). + """ + + def __init__( + self, + *, + details: dict[str, Any], + container_state: dict[str, Any] | None = None, + graph_state: dict[str, Any] | None = None, + ) -> None: + self.base_url = "https://tangle.example" + self._details = details + self._container_state = container_state or {} + self._graph_state = graph_state or {} + self.details_calls: list[str] = [] + self.container_calls: list[str] = [] + self.graph_calls: list[str] = [] + + def executions_details(self, id: str) -> dict[str, Any]: + return self._resolve(self._details, self.details_calls, id) + + def _resolve(self, store: dict[str, Any], calls: list[str], id: str) -> dict[str, Any]: + calls.append(id) + if id not in store: + raise _http_error(404) + value = store[id] + if isinstance(value, requests.RequestException): + raise value + return value + + def executions_container_state(self, id: str, **_: Any) -> dict[str, Any]: + return self._resolve(self._container_state, self.container_calls, id) + + def executions_graph_execution_state(self, id: str) -> dict[str, Any]: + return self._resolve(self._graph_state, self.graph_calls, id) + + +def test_task_statuses_walks_children_via_container_state() -> None: + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"a": "exec-a", "b": "exec-b"}}}, + container_state={ + "exec-a": {"status": "SUCCEEDED"}, + "exec-b": {"status": "RUNNING"}, + }, + ) + manager = PipelineRunManager(client=client) + + assert manager.task_statuses("root-1") == {"a": "SUCCEEDED", "b": "RUNNING"} + # Container state resolves first, so graph state is never queried. + assert client.graph_calls == [] + + +def test_task_statuses_falsy_child_id_is_unknown_without_api_call() -> None: + # A None/empty/whitespace-only child execution id maps straight to UNKNOWN; + # it must not be stringified to "None"/blank and queried against the state + # endpoints. + client = ExecutionStateClient( + details={ + "root-1": {"child_task_execution_ids": {"a": None, "b": "", "c": "exec-c", "d": " "}} + }, + container_state={"exec-c": {"status": "RUNNING"}}, + ) + manager = PipelineRunManager(client=client) + + assert manager.task_statuses("root-1") == { + "a": "UNKNOWN", + "b": "UNKNOWN", + "c": "RUNNING", + "d": "UNKNOWN", + } + # Only the real child id reaches the state API. + assert client.container_calls == ["exec-c"] + assert client.graph_calls == [] + + +def test_task_statuses_root_only_execution_reports_root_key() -> None: + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {}}}, + container_state={"root-1": {"status": "SUCCEEDED"}}, + ) + manager = PipelineRunManager(client=client) + + assert manager.task_statuses("root-1") == {"root": "SUCCEEDED"} + + +def test_task_statuses_falls_back_to_graph_state_on_404() -> None: + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"a": "exec-a"}}}, + container_state={}, # exec-a missing -> 404 from container state + graph_state={"exec-a": {"status": "RUNNING"}}, + ) + manager = PipelineRunManager(client=client) + + assert manager.task_statuses("root-1") == {"a": "RUNNING"} + assert client.container_calls == ["exec-a"] + assert client.graph_calls == ["exec-a"] + + +def test_task_statuses_persistent_5xx_reported_as_unknown() -> None: + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"a": "exec-a"}}}, + container_state={"exec-a": _http_error(503)}, + graph_state={"exec-a": _http_error(503)}, + ) + logger = CaptureLogger() + manager = PipelineRunManager(client=client, logger=logger) + + assert manager.task_statuses("root-1") == {"a": "UNKNOWN"} + # The persistent 5xx is announced so an UNKNOWN result is diagnosable. + logs = logger.get_logs() or "" + assert "exec-a" in logs + assert "503" in logs + + +def test_task_statuses_past_deadline_reports_unknown_without_fetching() -> None: + # Past the caller's deadline no further state requests are issued, so one + # task-statuses poll cannot overrun the wait budget of task-wait. + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"a": "exec-a"}}}, + container_state={"exec-a": {"status": "SUCCEEDED"}}, + ) + manager = PipelineRunManager(client=client) + + assert manager.task_statuses("root-1", deadline=time.monotonic() - 1) == {"a": "UNKNOWN"} + assert client.container_calls == [] + assert client.graph_calls == [] + + +def test_failure_statuses_are_the_non_skipped_non_succeeded_terminals() -> None: + # _TASK_FAILURE_STATUSES is derived from _TERMINAL_STATUSES; guard the intended + # membership so a new failure terminal can't be silently treated as success. + assert _TASK_FAILURE_STATUSES == {"FAILED", "SYSTEM_ERROR", "CANCELLED", "CANCELED", "INVALID"} + assert _TASK_FAILURE_STATUSES == frozenset(_TERMINAL_STATUSES) - {"SKIPPED", "SUCCEEDED"} + + +def test_task_statuses_wraps_non_retryable_http_error() -> None: + # A non-retryable state-endpoint status (429/403/etc.) becomes a clean + # PipelineRunError instead of a raw requests traceback out of task-status. + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"a": "exec-a"}}}, + container_state={"exec-a": _http_error(429)}, + ) + manager = PipelineRunManager(client=client) + + with pytest.raises(PipelineRunError, match="HTTP 429"): + manager.task_statuses("root-1") + + +def test_task_statuses_wraps_connection_error() -> None: + # A transport failure (no HTTP response) is surfaced cleanly rather than + # escaping as a raw requests.ConnectionError. + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"a": "exec-a"}}}, + container_state={"exec-a": requests.ConnectionError("connection refused")}, + ) + manager = PipelineRunManager(client=client) + + with pytest.raises(PipelineRunError, match="connection refused"): + manager.task_statuses("root-1") + + +def test_task_statuses_missing_root_is_not_found() -> None: + # A 404 on the root execution details is a clean not-found error, not a + # masked {"root": "UNKNOWN"} map. + client = ExecutionStateClient(details={"root-1": _http_error(404)}) + manager = PipelineRunManager(client=client) + + with pytest.raises(PipelineRunError, match="not found"): + manager.task_statuses("root-1") + + +def test_task_statuses_root_details_transport_error_surfaces_cleanly() -> None: + client = ExecutionStateClient(details={"root-1": requests.ConnectionError("boom")}) + manager = PipelineRunManager(client=client) + + with pytest.raises(PipelineRunError, match="boom"): + manager.task_statuses("root-1") + + +def test_task_statuses_root_details_5xx_is_transient_error() -> None: + # A one-shot task-status still fails fast on a root details 5xx, but the + # error is typed so the task-wait polling loop can retry it. + client = ExecutionStateClient(details={"root-1": _http_error(503)}) + manager = PipelineRunManager(client=client) + + with pytest.raises(TransientServerError, match="HTTP 503"): + manager.task_statuses("root-1") + + +def test_task_statuses_derives_graph_status_from_child_stats() -> None: + # A child that is itself a graph/subpipeline has no direct `status`; derive + # it from child_execution_status_stats counts rather than reporting UNKNOWN. + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"sub": "exec-sub"}}}, + container_state={}, # 404 -> fall back to graph state + graph_state={"exec-sub": {"child_execution_status_stats": {"leaf": {"SUCCEEDED": 1}}}}, + ) + manager = PipelineRunManager(client=client) + + assert manager.task_statuses("root-1") == {"sub": "SUCCEEDED"} + + +def test_task_statuses_derives_graph_status_from_status_totals() -> None: + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"sub": "exec-sub"}}}, + container_state={}, + graph_state={"exec-sub": {"status_totals": {"RUNNING": 1}}}, + ) + manager = PipelineRunManager(client=client) + + # An active count outranks terminal counts, matching status_from_counts. + assert manager.task_statuses("root-1") == {"sub": "RUNNING"} + + +def test_task_statuses_prefers_direct_graph_status_over_counts() -> None: + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"sub": "exec-sub"}}}, + container_state={}, + graph_state={"exec-sub": {"status": "FAILED", "status_totals": {"SUCCEEDED": 1}}}, + ) + manager = PipelineRunManager(client=client) + + assert manager.task_statuses("root-1") == {"sub": "FAILED"} + + +@pytest.mark.parametrize( + "counts, expected", + [ + # An active/nonterminal status must outrank terminal success. + ({"SUCCEEDED": 1, "WAITING_FOR_UPSTREAM": 1}, "WAITING_FOR_UPSTREAM"), + ({"SUCCEEDED": 1, "UNINITIALIZED": 1}, "UNINITIALIZED"), + # Failure/invalid terminal statuses must outrank SUCCEEDED. + ({"SUCCEEDED": 1, "INVALID": 1}, "INVALID"), + ({"SUCCEEDED": 1, "FAILED": 1}, "FAILED"), + # INVALID must also outrank the other non-failure terminal, SKIPPED, so a + # mixed graph aggregate does not mask an invalid child behind SKIPPED. + ({"SKIPPED": 1, "INVALID": 1}, "INVALID"), + ({"SKIPPED": 1, "FAILED": 1}, "FAILED"), + ], +) +def test_status_from_counts_nonterminal_and_failure_outrank_success(counts, expected) -> None: + assert PipelineRunManager.status_from_counts(counts) == expected + + +def test_task_statuses_nested_graph_waiting_is_nonterminal_and_blocks_wait() -> None: + # {"SUCCEEDED": 1, "WAITING_FOR_UPSTREAM": 1} must not collapse to SUCCEEDED + # and must not be treated as terminal by task-wait. + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"sub": "exec-sub"}}}, + container_state={}, + graph_state={ + "exec-sub": {"child_execution_status_stats": {"leaf": {"SUCCEEDED": 1, "WAITING_FOR_UPSTREAM": 1}}} + }, + ) + manager = PipelineRunManager(client=client) + + statuses = manager.task_statuses("root-1") + assert statuses == {"sub": "WAITING_FOR_UPSTREAM"} + assert not PipelineRunManager.is_terminal_status(statuses["sub"]) + + +def test_task_statuses_nested_graph_invalid_is_failure(monkeypatch) -> None: + # {"SUCCEEDED": 1, "INVALID": 1} reduces to INVALID, so task-wait fails. + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"sub": "exec-sub"}}}, + container_state={}, + graph_state={"exec-sub": {"status_totals": {"SUCCEEDED": 1, "INVALID": 1}}}, + ) + manager = PipelineRunManager(client=client) + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda value: None) + + with pytest.raises(TaskStatusesFailed) as exc_info: + manager.wait_for_task_statuses("root-1", poll_interval=1) + assert exc_info.value.statuses == {"sub": "INVALID"} + + +def test_task_statuses_nested_graph_skipped_and_invalid_collapses_to_invalid(monkeypatch) -> None: + # A nested-graph child whose counts mix SKIPPED and INVALID must reduce to + # INVALID (the failure), not SKIPPED, so task-wait surfaces the invalid child. + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"sub": "exec-sub"}}}, + container_state={}, + graph_state={"exec-sub": {"status_totals": {"SKIPPED": 1, "INVALID": 1}}}, + ) + manager = PipelineRunManager(client=client) + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda value: None) + + assert manager.task_statuses("root-1") == {"sub": "INVALID"} + with pytest.raises(TaskStatusesFailed) as exc_info: + manager.wait_for_task_statuses("root-1", poll_interval=1) + assert exc_info.value.failures == {"sub": "INVALID"} + + +def test_wait_for_task_statuses_skipped_is_successful_terminal() -> None: + # SKIPPED is a non-failure terminal (a skipped task, e.g. a conditional + # branch, is not a failure), so a fully SKIPPED task map returns + # successfully (exit 0), not as a failure. + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + manager.task_statuses = lambda root_id, deadline=None: {"a": "SKIPPED"} # type: ignore[method-assign] + + assert manager.wait_for_task_statuses("root-1") == {"a": "SKIPPED"} + + +def test_wait_for_task_statuses_returns_terminal_map(monkeypatch) -> None: + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda value: sleeps.append(value)) + statuses = iter([{"a": "RUNNING"}, {"a": "SUCCEEDED"}]) + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + monkeypatch.setattr(manager, "task_statuses", lambda root_id, deadline=None: next(statuses)) + + assert manager.wait_for_task_statuses("root-1", poll_interval=1) == {"a": "SUCCEEDED"} + assert sleeps == [1] + + +def test_wait_for_task_statuses_max_wait_none_waits_until_terminal(monkeypatch) -> None: + # max_wait=None waits indefinitely: no deadline is computed and the plain + # poll interval is slept between checks. + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda value: sleeps.append(value)) + statuses = iter([{"a": "RUNNING"}, {"a": "RUNNING"}, {"a": "SUCCEEDED"}]) + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + monkeypatch.setattr(manager, "task_statuses", lambda root_id, deadline=None: next(statuses)) + + result = manager.wait_for_task_statuses("root-1", max_wait=None, poll_interval=3) + + assert result == {"a": "SUCCEEDED"} + assert sleeps == [3, 3] + + +def test_wait_for_task_statuses_raises_on_failure() -> None: + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + manager.task_statuses = lambda root_id, deadline=None: {"a": "SUCCEEDED", "b": "FAILED"} # type: ignore[method-assign] + + with pytest.raises(TaskStatusesFailed) as exc_info: + manager.wait_for_task_statuses("root-1") + # Failures drive the exit code, but the full final map is retained too. + assert exc_info.value.failures == {"b": "FAILED"} + assert exc_info.value.statuses == {"a": "SUCCEEDED", "b": "FAILED"} + + +def test_wait_for_task_statuses_allow_failure_returns_map() -> None: + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + final = {"a": "SUCCEEDED", "b": "CANCELLED"} + manager.task_statuses = lambda root_id, deadline=None: final # type: ignore[method-assign] + + assert manager.wait_for_task_statuses("root-1", allow_failure=True) == final + + +def test_wait_for_task_statuses_timeout(monkeypatch) -> None: + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda value: None) + # deadline calc, first check, sleep-clamp read, then second check past deadline. + ticks = iter([0.0, 0.0, 0.0, 100.0]) + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.monotonic", lambda: next(ticks)) + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + manager.task_statuses = lambda root_id, deadline=None: {"a": "RUNNING"} # type: ignore[method-assign] + + with pytest.raises(PipelineRunError, match="non-terminal"): + manager.wait_for_task_statuses("root-1", max_wait=10, poll_interval=1) + + +def test_wait_for_task_statuses_clamps_final_sleep_to_remaining(monkeypatch) -> None: + # Near the deadline the final sleep must shrink to the remaining budget so the + # loop times out at max_wait instead of overshooting by a full poll interval. + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda value: sleeps.append(value)) + # deadline calc -> 0 (deadline=10); check -> 8 (not past); sleep clamp -> 8 + # (10-8=2 < poll_interval 5); next check -> 10 (timeout). + ticks = iter([0.0, 8.0, 8.0, 10.0]) + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.monotonic", lambda: next(ticks)) + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + manager.task_statuses = lambda root_id, deadline=None: {"a": "RUNNING"} # type: ignore[method-assign] + + with pytest.raises(PipelineRunError, match="non-terminal"): + manager.wait_for_task_statuses("root-1", max_wait=10, poll_interval=5) + assert sleeps == [2.0] + + +def test_wait_for_task_statuses_retries_transient_root_fetch_5xx(monkeypatch) -> None: + # A single 5xx blip on the per-poll root details fetch must not abort a + # long wait; it is tolerated like child-state 5xx and retried next poll. + sleeps: list[float] = [] + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda value: sleeps.append(value)) + outcomes = iter([TransientServerError("Fetching root execution root-1: request failed with HTTP 503"), {"a": "SUCCEEDED"}]) + + def fetch(root_id: str, deadline: float | None = None) -> dict[str, str]: + value = next(outcomes) + if isinstance(value, Exception): + raise value + return value + + logger = CaptureLogger() + manager = PipelineRunManager(client=ExecutionStateClient(details={}), logger=logger) + monkeypatch.setattr(manager, "task_statuses", fetch) + + assert manager.wait_for_task_statuses("root-1", poll_interval=1) == {"a": "SUCCEEDED"} + assert sleeps == [1] + # The tolerated blip is announced so a later timeout is diagnosable. + assert "HTTP 503" in (logger.get_logs() or "") + + +def test_wait_for_task_statuses_persistent_root_fetch_5xx_raises_at_deadline(monkeypatch) -> None: + # Retrying the root fetch is bounded by the wait deadline: a persistent + # outage surfaces as the underlying server error once max_wait elapses, + # and the warning is logged once per outage rather than once per poll. + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda value: None) + # deadline calc, first-failure deadline check, sleep-clamp read, then + # second-failure deadline check past the deadline. + ticks = iter([0.0, 0.0, 0.0, 100.0]) + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.monotonic", lambda: next(ticks)) + logger = CaptureLogger() + manager = PipelineRunManager(client=ExecutionStateClient(details={}), logger=logger) + + def fetch(root_id: str, deadline: float | None = None) -> dict[str, str]: + raise TransientServerError("Fetching root execution root-1: request failed with HTTP 502") + + monkeypatch.setattr(manager, "task_statuses", fetch) + + with pytest.raises(TransientServerError, match="HTTP 502"): + manager.wait_for_task_statuses("root-1", max_wait=10, poll_interval=1) + assert (logger.get_logs() or "").count("HTTP 502") == 1 + + +def test_wait_for_task_statuses_unbounded_wait_caps_root_fetch_retries(monkeypatch) -> None: + # With max_wait=None there is no deadline to stop the retries, so a + # persistent root fetch outage must still surface after a bounded number + # of consecutive failures instead of being retried forever. + monkeypatch.setattr("tangle_cli.pipeline_run_manager.time.sleep", lambda value: None) + calls = 0 + + def fetch(root_id: str, deadline: float | None = None) -> dict[str, str]: + nonlocal calls + calls += 1 + raise TransientServerError("Fetching root execution root-1: request failed with HTTP 500") + + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + monkeypatch.setattr(manager, "task_statuses", fetch) + + with pytest.raises(TransientServerError, match="HTTP 500"): + manager.wait_for_task_statuses("root-1", max_wait=None, poll_interval=1) + assert calls == pipeline_run_manager._MAX_UNBOUNDED_ROOT_FETCH_FAILURES + + +def test_wait_for_task_statuses_rejects_invalid_bounds() -> None: + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + with pytest.raises(PipelineRunError, match="max-wait"): + manager.wait_for_task_statuses("root-1", max_wait=-1) + with pytest.raises(PipelineRunError, match="poll-interval"): + manager.wait_for_task_statuses("root-1", poll_interval=-1) + + +def test_wait_for_task_statuses_rejects_non_finite_bounds() -> None: + # NaN/inf bounds slip past the sign checks; they must reject cleanly instead + # of reaching time.sleep (raw ValueError) or creating an unbounded deadline. + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + for bad in (math.nan, math.inf): + with pytest.raises(PipelineRunError, match="max-wait must be a finite number"): + manager.wait_for_task_statuses("root-1", max_wait=bad) + with pytest.raises(PipelineRunError, match="poll-interval must be a finite number"): + manager.wait_for_task_statuses("root-1", max_wait=1, poll_interval=bad) + + +def test_wait_for_task_statuses_rejects_zero_poll_interval_without_spinning(monkeypatch) -> None: + # A zero interval must be rejected before any polling, not busy-loop. + calls = 0 + + def counting_statuses(root_id: str, deadline: float | None = None) -> dict[str, str]: + nonlocal calls + calls += 1 + return {"a": "RUNNING"} + + manager = PipelineRunManager(client=ExecutionStateClient(details={})) + monkeypatch.setattr(manager, "task_statuses", counting_statuses) + + with pytest.raises(PipelineRunError, match="poll-interval must be positive"): + manager.wait_for_task_statuses("root-1", max_wait=0.02, poll_interval=0) + assert calls == 0 + + +def test_cli_task_status_outputs_json(monkeypatch, capsys) -> None: + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"a": "exec-a"}}}, + container_state={"exec-a": {"status": "SUCCEEDED"}}, + ) + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: client) + + run_app(cli.build_app(), ["sdk", "pipeline-runs", "task-status", "root-1"]) + + assert json.loads(capsys.readouterr().out) == {"a": "SUCCEEDED"} + + +def test_cli_task_wait_failure_exits_nonzero_and_prints_full_map(monkeypatch, capsys) -> None: + # Mixed outcome: the failure exit code must not drop the succeeded task. + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"a": "exec-a", "b": "exec-b"}}}, + container_state={ + "exec-a": {"status": "SUCCEEDED"}, + "exec-b": {"status": "FAILED"}, + }, + ) + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: client) + + with pytest.raises(SystemExit) as exc_info: + # All tasks are already terminal, so the first poll returns without sleeping. + run_app(cli.build_app(), ["sdk", "pipeline-runs", "task-wait", "root-1", "--poll-interval", "1"]) + + assert exc_info.value.code == 2 + captured = capsys.readouterr() + assert json.loads(captured.out) == {"a": "SUCCEEDED", "b": "FAILED"} + # A human-readable failure summary goes to stderr; stdout stays pure JSON. + assert "failed task" in captured.err + + +def test_cli_task_wait_success_outputs_map(monkeypatch, capsys) -> None: + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"a": "exec-a"}}}, + container_state={"exec-a": {"status": "SUCCEEDED"}}, + ) + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: client) + + run_app(cli.build_app(), ["sdk", "pipeline-runs", "task-wait", "root-1", "--poll-interval", "1"]) + + assert json.loads(capsys.readouterr().out) == {"a": "SUCCEEDED"} + + +def test_cli_task_wait_skipped_exits_zero(monkeypatch, capsys) -> None: + # SKIPPED is a non-failure terminal, so task-wait must exit 0 rather than + # raise SystemExit. + client = ExecutionStateClient( + details={"root-1": {"child_task_execution_ids": {"a": "exec-a"}}}, + container_state={"exec-a": {"status": "SKIPPED"}}, + ) + monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: client) + + run_app(cli.build_app(), ["sdk", "pipeline-runs", "task-wait", "root-1", "--poll-interval", "1"]) + + assert json.loads(capsys.readouterr().out) == {"a": "SKIPPED"}