feat(pipeline-run): configure submit recovery attempts

Assisted-By: devx/12bfa114-4ee7-4929-a37c-98018f6b0f41

Author: Volv-G

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

@@ -14,6 +14,6 @@
 try:
     __version__ = metadata_version("tangle-cli")
 except PackageNotFoundError:
-    __version__ = "0.1.0"
+    __version__ = "0.1.1"
 
 __all__ = ["TangleDynamicDiscoveryClient", "__version__"]

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

@@ -37,7 +37,8 @@
 _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"
-_SUBMIT_RECOVERY_BACKOFF_SECONDS = (0.5, 1.0, 2.0)
+_SUBMIT_RECOVERY_BACKOFF_SECONDS = (0.5, 1.0, 2.0, 4.0, 8.0, 16.0)
+_DEFAULT_SUBMIT_RECOVERY_ATTEMPTS = 2
 
 
 class PipelineRunError(RuntimeError):
@@ -1528,6 +1529,11 @@ def _submission_id_from_body(body: Mapping[str, Any]) -> str | None:
         submission_id = annotations.get(_SUBMISSION_ID_ANNOTATION_KEY)
         return str(submission_id) if submission_id else None
 
+    @staticmethod
+    def _submit_recovery_backoff_seconds(submit_recovery_attempts: int) -> tuple[float, ...]:
+        attempt_count = max(0, min(int(submit_recovery_attempts), len(_SUBMIT_RECOVERY_BACKOFF_SECONDS)))
+        return _SUBMIT_RECOVERY_BACKOFF_SECONDS[:attempt_count]
+
     def _submitted_runs_for_submission_id(self, submission_id: str) -> list[dict[str, Any]]:
         query = {
             "and": [
@@ -1553,11 +1559,21 @@ def _recover_submitted_run_after_submit_error(
         self,
         *,
         submission_id: str | None,
+        submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS,
     ) -> dict[str, Any] | None:
         if not submission_id:
             return None
-        total_lookup_attempts = len(_SUBMIT_RECOVERY_BACKOFF_SECONDS)
-        for lookup_attempt, delay_seconds in enumerate(_SUBMIT_RECOVERY_BACKOFF_SECONDS, start=1):
+        backoff_seconds = self._submit_recovery_backoff_seconds(submit_recovery_attempts)
+        total_lookup_attempts = len(backoff_seconds)
+        if total_lookup_attempts == 0:
+            self.logger.warn(
+                "Submit recovery lookup disabled "
+                f"({_SUBMISSION_ID_ANNOTATION_KEY}={submission_id}, "
+                f"submit_recovery_attempts={submit_recovery_attempts}); "
+                "resubmitting the same frozen body with preserved inputs."
+            )
+            return None
+        for lookup_attempt, delay_seconds in enumerate(backoff_seconds, start=1):
             self.logger.info(
                 "Waiting "
                 f"{delay_seconds:g}s before checking whether failed submit already created a pipeline run "
@@ -1652,6 +1668,7 @@ def _run_body_factory(
         timeout_clock: str = "monotonic",
         exit_on_first_failure: bool = False,
         metadata: dict[str, Any] | None = None,
+        submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS,
         metadata_factory: Callable[
             [int, PipelineRunContext | None, Exception | None], dict[str, Any]
         ] | None = None,
@@ -1728,6 +1745,7 @@ def _run_body_factory(
                         if reused_after_submit_failure:
                             recovered_response = self._recover_submitted_run_after_submit_error(
                                 submission_id=self._submission_id_from_body(body),
+                                submit_recovery_attempts=submit_recovery_attempts,
                             )
                         if recovered_response is not None:
                             response = self._adopt_submitted_run(
@@ -1759,6 +1777,7 @@ def _run_body_factory(
                                 )
                                 recovered_response = self._recover_submitted_run_after_submit_error(
                                     submission_id=submission_id_for_recovery,
+                                    submit_recovery_attempts=submit_recovery_attempts,
                                 )
                                 if recovered_response is None:
                                     self.hooks.on_submit_error(submit_exc, context=context)
@@ -1839,6 +1858,7 @@ def run_prepared_body(
         timeout_clock: str = "monotonic",
         exit_on_first_failure: bool = False,
         metadata: dict[str, Any] | None = None,
+        submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS,
     ) -> dict[str, Any]:
         """Submit/wait/retry an already prepared submit body.
 
@@ -1867,6 +1887,7 @@ def body_factory(
             timeout_clock=timeout_clock,
             exit_on_first_failure=exit_on_first_failure,
             metadata=metadata,
+            submit_recovery_attempts=submit_recovery_attempts,
         )
 
     def run_pipeline_spec(
@@ -1887,6 +1908,7 @@ def run_pipeline_spec(
         timeout_clock: str = "monotonic",
         exit_on_first_failure: bool = False,
         metadata: dict[str, Any] | None = None,
+        submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS,
     ) -> dict[str, Any]:
         """Submit/wait/retry an already hydrated/validated in-memory spec."""
 
@@ -1916,6 +1938,7 @@ def body_factory(
             timeout_clock=timeout_clock,
             exit_on_first_failure=exit_on_first_failure,
             metadata=metadata,
+            submit_recovery_attempts=submit_recovery_attempts,
         )
 
     def run_pipeline(
@@ -1936,6 +1959,7 @@ def run_pipeline(
         timeout_clock: str = "monotonic",
         exit_on_first_failure: bool = False,
         metadata: dict[str, Any] | None = None,
+        submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS,
     ) -> dict[str, Any]:
         """Submit (and optionally wait for) a pipeline with lifecycle hooks.
 
@@ -1970,6 +1994,7 @@ def body_factory(
             timeout_clock=timeout_clock,
             exit_on_first_failure=exit_on_first_failure,
             metadata=metadata,
+            submit_recovery_attempts=submit_recovery_attempts,
         )
 
 

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

@@ -507,6 +507,7 @@ def run_pipeline(
         open_browser: bool = False,
         include_next_steps: bool = False,
         metadata: dict[str, Any] | None = None,
+        submit_recovery_attempts: int = 2,
     ) -> dict[str, Any]:
         """Run a pipeline path through generic preparation + lifecycle hooks.
 
@@ -603,6 +604,7 @@ def metadata_factory(
                 timeout_clock=timeout_clock,
                 exit_on_first_failure=exit_on_first_failure,
                 metadata_factory=metadata_factory,
+                submit_recovery_attempts=submit_recovery_attempts,
             )
             context = result.get("context")
             attempt = context.attempt if isinstance(context, PipelineRunContext) else max(preparations)

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

@@ -28,6 +28,7 @@
 from .logger import Logger, logger_for_log_type
 from .pipeline_run_annotations import AnnotationManager
 from .pipeline_run_manager import (
+    _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS,
     PipelineRunError,
     PipelineRunHooks,
     PipelineRunManager,
@@ -166,6 +167,15 @@ def pipeline_runs_submit(
     auth_header: AuthHeaderOption = None,
     header: HeaderOption = None,
     config: ConfigOption = None,
+    submit_recovery_attempts: Annotated[
+        int,
+        Parameter(
+            help=(
+                "Number of post-failed-submit recovery lookups before resubmitting; "
+                "higher values wait longer for delayed run registration."
+            )
+        ),
+    ] = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS,
     log_type: LogTypeOption = "console",
 ) -> None:
     """Hydrate and submit a local pipeline YAML file as a run."""
@@ -181,6 +191,7 @@ def pipeline_runs_submit(
         "run_as": (run_as, None),
         "trusted_source": (trusted_source, None),
         "trusted_hydration_cli": ("trusted_hydration_cli", trusted_hydration, None, False),
+        "submit_recovery_attempts": (submit_recovery_attempts, _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS),
         "log_type": (log_type, "console"),
         **api_arg_specs(base_url=base_url, token=token, auth_header=auth_header, header=header),
     }
@@ -194,7 +205,12 @@ def action(manager: PipelineRunManager, args: ArgsContainer) -> dict[str, Any]:
         }
         if args.dry_run:
             return manager.build_submit_body(args.pipeline_path, **kwargs)
-        return manager.submit_pipeline(args.pipeline_path, **kwargs)
+        result = manager.run_pipeline(
+            args.pipeline_path,
+            **kwargs,
+            submit_recovery_attempts=args.submit_recovery_attempts,
+        )
+        return result["response"]
 
     _run_manager_action(config, base_url, specs, action)
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "tangle-cli"
-version = "0.1.0"
+version = "0.1.1"
 description = "CLI for Tangle, the open-source ML pipeline orchestration platform"
 readme = "README.md"
 authors = [

tests/test_packaging.py

@@ -14,7 +14,6 @@
 
 from tangle_cli.openapi import codegen
 
-
 _REPO_ROOT = Path(__file__).resolve().parents[1]
 
 
@@ -179,7 +178,7 @@ def test_tangle_cli_wheel_supports_expert_no_deps_import_path_without_tangle_api
     requires_dist = [line for line in metadata.splitlines() if line.startswith("Requires-Dist: ")]
     assert not any(name.startswith("tangle_api/") for name in names)
     assert "tangle_cli/openapi/openapi.json" not in names
-    assert "Version: 0.1.0" in metadata
+    assert "Version: 0.1.1" in metadata
     assert "Requires-Dist: tangle-api==0.1.0" in requires_dist
     assert not any("extra == 'native'" in line for line in requires_dist)
     assert "Provides-Extra: native" in metadata

tests/test_pipeline_runs_cli.py

@@ -135,7 +135,9 @@ def test_pipeline_runs_help_exposes_run_commands_not_local_pipeline_commands(cap
     assert "diagram" not in output
 
     run_app(app, ["sdk", "pipeline-runs", "submit", "--help"])
-    assert "--log-type" in capsys.readouterr().out
+    submit_help = capsys.readouterr().out
+    assert "--log-type" in submit_help
+    assert "--submit-recovery-attempts" in submit_help
 
 
 def test_pipeline_runs_submit_builds_create_payload(monkeypatch, tmp_path: Path, capsys):
@@ -161,12 +163,62 @@ def test_pipeline_runs_submit_builds_create_payload(monkeypatch, tmp_path: Path,
 
     result = json.loads(capsys.readouterr().out)
     assert result == {"id": "run-1", "root_execution_id": "exec-1"}
-    assert fake_client.created[0]["annotations"] == {"team": "oss"}
+    assert fake_client.created[0]["annotations"]["team"] == "oss"
+    assert fake_client.created[0]["annotations"]["tangle-cli/submission-id"]
     root_task = fake_client.created[0]["root_task"]
     assert root_task["componentRef"]["spec"]["name"] == "Demo Pipeline"
     assert root_task["arguments"] == {"query": "default", "required": "value"}
 
 
+def test_pipeline_runs_submit_uses_default_submit_recovery_attempts(monkeypatch, tmp_path: Path, capsys):
+    pipeline_path = _write_pipeline(tmp_path / "pipeline.yaml")
+    captured: dict[str, Any] = {}
+
+    def fake_run_pipeline(self, pipeline_path, **kwargs):
+        del self, pipeline_path
+        captured.update(kwargs)
+        return {"response": {"id": "run-1"}}
+
+    monkeypatch.setattr(PipelineRunManager, "run_pipeline", fake_run_pipeline)
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: FakeClient())
+    app = cli.build_app()
+
+    run_app(app, ["sdk", "pipeline-runs", "submit", str(pipeline_path), "--no-hydrate"])
+
+    assert json.loads(capsys.readouterr().out) == {"id": "run-1"}
+    assert captured["submit_recovery_attempts"] == pipeline_run_manager._DEFAULT_SUBMIT_RECOVERY_ATTEMPTS
+
+
+def test_pipeline_runs_submit_accepts_submit_recovery_attempts(monkeypatch, tmp_path: Path, capsys):
+    pipeline_path = _write_pipeline(tmp_path / "pipeline.yaml")
+    captured: dict[str, Any] = {}
+
+    def fake_run_pipeline(self, pipeline_path, **kwargs):
+        del self, pipeline_path
+        captured.update(kwargs)
+        return {"response": {"id": "run-1"}}
+
+    monkeypatch.setattr(PipelineRunManager, "run_pipeline", fake_run_pipeline)
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: FakeClient())
+    app = cli.build_app()
+
+    run_app(
+        app,
+        [
+            "sdk",
+            "pipeline-runs",
+            "submit",
+            str(pipeline_path),
+            "--no-hydrate",
+            "--submit-recovery-attempts",
+            "6",
+        ],
+    )
+
+    assert json.loads(capsys.readouterr().out) == {"id": "run-1"}
+    assert captured["submit_recovery_attempts"] == 6
+
+
 def test_pipeline_runs_submit_accepts_export_config_args_and_hydrate(monkeypatch, tmp_path: Path):
     pipeline_path = _write_pipeline(tmp_path / "pipeline.yaml")
     config = tmp_path / "pipeline.config.yaml"
@@ -1752,7 +1804,7 @@ def pipeline_runs_create(self, body: Any = None) -> dict[str, Any]:
 
         def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]:
             self.list_calls.append(kwargs)
-            if len(self.list_calls) <= 2:
+            if len(self.list_calls) < len(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS):
                 return {"pipeline_runs": [], "next_page_token": None}
             return {
                 "pipeline_runs": [{"id": "run-created", "root_execution_id": "exec-created"}],
@@ -1763,13 +1815,13 @@ def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]:
     manager = PipelineRunManager(client=client)
     body = {"root_task": {"componentRef": {"spec": {"name": "delayed-recovery"}}}}
 
-    result = manager.run_prepared_body(body)
+    result = manager.run_prepared_body(body, submit_recovery_attempts=6)
 
     assert result["response"]["id"] == "run-created"
     assert result["context"].metadata["recovered_after_submit_error"] is True
     assert len(client.created) == 1
-    assert len(client.list_calls) == 3
-    assert sleeps == [0.5, 1.0, 2.0]
+    assert len(client.list_calls) == len(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS)
+    assert sleeps == list(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS)
 
 
 def test_pipeline_runs_submit_failure_recovery_refuses_ambiguous_matches(monkeypatch) -> None:
@@ -1869,8 +1921,8 @@ def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]:
     assert client.created[0]["annotations"]["tangle-cli/submission-id"] == client.created[1]["annotations"][
         "tangle-cli/submission-id"
     ]
-    assert len(client.list_calls) == 2 * len(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS)
-    expected_sleeps = list(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS)
+    expected_sleeps = list(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS[:2])
+    assert len(client.list_calls) == 2 * len(expected_sleeps)
     assert sleeps == expected_sleeps + expected_sleeps
 
 
@@ -1927,15 +1979,15 @@ def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]:
     client = RecoverOnRetryClient()
     manager = PipelineRunner(client=client, hooks=hooks)
 
-    result = manager.run_pipeline(pipeline_path, hydrate=False, max_attempts=2)
+    result = manager.run_pipeline(pipeline_path, hydrate=False, max_attempts=2, submit_recovery_attempts=6)
 
     assert result["response"]["id"] == "run-created"
     assert result["context"].attempt == 2
     assert result["context"].metadata["recovered_after_submit_error"] is True
     assert hooks.prepare_run_arguments_calls == 1
     assert len(client.created) == 1
     assert len(client.list_calls) == len(pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS) + 1
-    assert sleeps == [0.5, 1.0, 2.0, 0.5]
+    assert sleeps == [*pipeline_run_manager._SUBMIT_RECOVERY_BACKOFF_SECONDS, 0.5]
     assert events == [("before_retry", 2, None), ("after_retry_submit", 2, "run-created")]