fix(tangle-cli): defer submit failure hooks during recovery

Assisted-By: devx/0c001c3a-79d6-4fd4-9ba8-4012e253d1e2

Author: Volv-G

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

@@ -1049,6 +1049,7 @@ def submit_prepared_body(
         pipeline_path: str | Path | None = None,
         attempt: int = 1,
         context: PipelineRunContext | None = None,
+        notify_submit_error: bool = True,
     ) -> dict[str, Any]:
         self.normalize_submit_body_in_place(body)
         pipeline_spec = body["root_task"]["componentRef"]["spec"]
@@ -1068,7 +1069,8 @@ def submit_prepared_body(
         try:
             response = self.to_plain(client.pipeline_runs_create(body=body))
         except Exception as exc:
-            self.hooks.on_submit_error(exc, context=submit_context)
+            if notify_submit_error:
+                self.hooks.on_submit_error(exc, context=submit_context)
             raise
         if not isinstance(response, dict):
             response = {}
@@ -1737,6 +1739,7 @@ def _run_body_factory(
                                     pipeline_path=pipeline_path,
                                     attempt=attempt,
                                     context=context,
+                                    notify_submit_error=False,
                                 )
                             except Exception as submit_exc:
                                 if context.run_id is not None:
@@ -1751,6 +1754,7 @@ def _run_body_factory(
                                     submission_id=submission_id_for_recovery,
                                 )
                                 if recovered_response is None:
+                                    self.hooks.on_submit_error(submit_exc, context=context)
                                     raise
                                 response = self._adopt_submitted_run(
                                     response=recovered_response,

tests/test_pipeline_runs_cli.py

@@ -1679,6 +1679,11 @@ class DynamicTimeHooks(PipelineRunnerHooks):
         def __init__(self) -> None:
             super().__init__()
             self.prepare_run_arguments_calls = 0
+            self.submit_errors: list[str] = []
+
+        def on_submit_error(self, error, *, context):
+            del context
+            self.submit_errors.append(str(error))
 
         def read_pipeline_yaml(self, pipeline_path):
             return {
@@ -1720,6 +1725,7 @@ def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]:
     assert result["context"].root_execution_id == "exec-created"
     assert result["context"].metadata["recovered_after_submit_error"] is True
     assert hooks.prepare_run_arguments_calls == 1
+    assert hooks.submit_errors == []
     assert len(client.created) == 1
     submission_id = client.created[0]["annotations"]["tangle-cli/submission-id"]
     assert submission_id
@@ -1739,6 +1745,11 @@ class DynamicTimeHooks(PipelineRunnerHooks):
         def __init__(self) -> None:
             super().__init__()
             self.prepare_run_arguments_calls = 0
+            self.submit_errors: list[str] = []
+
+        def on_submit_error(self, error, *, context):
+            del context
+            self.submit_errors.append(str(error))
 
         def read_pipeline_yaml(self, pipeline_path):
             return {
@@ -1779,6 +1790,7 @@ def pipeline_runs_list(self, **kwargs: Any) -> dict[str, Any]:
 
     assert result["response"]["id"] == "run-2"
     assert hooks.prepare_run_arguments_calls == 1
+    assert hooks.submit_errors == ["submit timed out"]
     assert [body["root_task"]["arguments"]["exec_time"] for body in client.created] == ["time-1", "time-1"]
     assert [body["root_task"]["componentRef"]["spec"]["name"] for body in client.created] == [
         "run-time-1",