feat: validate pipeline specs during submit

Run the full pipelines validate authoring validator during pipeline-runs submit and --dry-run after hydration, and route PipelineRunner through the same base hook. Root graph requirements are enforced by the authoring validator, with README and tests updated. Bump tangle-cli patch version to 0.1.3. This is based on master and intentionally excludes PR TangleML#27 skip-validation CLI flag and separate submit-root check.

Assisted-By: devx/5490f7cb-c7e4-4cf5-b525-a8c4e2af55b1

Author: Volv-G

README.md

@@ -293,6 +293,8 @@ uv run tangle sdk pipeline-runs export RUN_ID --output pipeline.yaml
 
 `submit` hydrates refs by default and builds an API submit payload with `root_task.componentRef.spec`. Use `--no-hydrate` to submit the local YAML structure as-is. Use `--dry-run` to print the payload without creating a run.
 
+Before creating a run—or printing a `--dry-run` payload—`submit` runs the same authoring validation as `tangle sdk pipelines validate` on the hydrated/resolved pipeline spec (or on the as-is spec when `--no-hydrate` is used). Invalid specs fail locally with `Pipeline validation failed` errors before the run-submission API call. For example, the pipeline root must be a graph (`implementation.graph`), so a bare `implementation.container` root is rejected before the run is submitted.
+
 ## Programmatic client
 
 The stable public wrapper for downstream Python tools is:

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

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

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

@@ -28,6 +28,7 @@
 from .pipeline_dehydrator import DehydrateChoice, PipelineDehydrator
 from .pipeline_hydrator import HydrationError, PipelineHydrator
 from .pipeline_run_details import PipelineRunDetails
+from .pipelines import collect_pipeline_spec_errors
 from .pipeline_run_search import PipelineRunSearch
 from .utils import dump_yaml
 
@@ -378,6 +379,26 @@ def prepare_run_arguments(
         """Hook for TD JOB_CONFIG time input / scheduled runtime behavior."""
         return run_args
 
+    def validate_pipeline_for_run(
+        self,
+        pipeline_spec: dict[str, Any],
+        *,
+        pipeline_path: str | Path | None,
+        effective_path: str | Path | None,
+        skip_validation: bool,
+    ) -> list[str]:
+        """Return submit-time validation errors for a prepared pipeline spec.
+
+        The OSS default enforces the same local authoring validator used by
+        ``tangle pipeline validate``. Downstreams can override or extend this
+        hook with stricter schema/input validators.
+        """
+
+        del pipeline_path, effective_path
+        if skip_validation:
+            return []
+        return collect_pipeline_spec_errors(pipeline_spec)
+
     def transform_run_name(
         self,
         run_name: str,
@@ -834,6 +855,11 @@ def _accepts_client_keyword(method: Any) -> bool:
             for parameter in parameters.values()
         )
 
+    @staticmethod
+    def _raise_pipeline_validation_error(validation_errors: list[str]) -> None:
+        if validation_errors:
+            raise PipelineRunError("Pipeline validation failed:\n  - " + "\n  - ".join(validation_errors))
+
     def load_pipeline_for_submit(
         self,
         pipeline_path: str | Path,
@@ -909,13 +935,15 @@ def prepare_submit_payload_from_spec(
         pipeline_path: str | Path | None = None,
         run_as: str | None = None,
         hydrate: bool = True,
+        skip_validation: bool = False,
     ) -> PipelineSubmitPayload:
         """Prepare the generic submit payload from a pipeline spec.
 
         The order here is the submit-body contract shared by OSS and TD:
         prepare the spec, prepare runtime arguments, expand run-name templates,
-        convert/sanitize the payload, then merge downstream/default annotations
-        before caller-supplied annotations override them.
+        validate the prepared authoring spec, convert/sanitize the payload, then
+        merge downstream/default annotations before caller-supplied annotations
+        override them.
         """
 
         prepared_spec = self.prepare_pipeline_spec_for_submit(
@@ -926,6 +954,13 @@ def prepare_submit_payload_from_spec(
         )
         prepared_run_args = self.hooks.prepare_run_arguments(prepared_spec, run_args)
         prepared_spec = self.apply_run_name_template(prepared_spec, prepared_run_args)
+        validation_errors = self.hooks.validate_pipeline_for_run(
+            prepared_spec,
+            pipeline_path=pipeline_path,
+            effective_path=None,
+            skip_validation=skip_validation,
+        )
+        self._raise_pipeline_validation_error(validation_errors)
         payload = self.convert_yaml_to_payload(copy.deepcopy(prepared_spec), prepared_run_args)
         payload = self.sanitize_submit_payload(payload)
         root_task = payload["root_task"]
@@ -961,6 +996,7 @@ def build_submit_body_from_spec(
         pipeline_path: str | Path | None = None,
         run_as: str | None = None,
         hydrate: bool = True,
+        skip_validation: bool = False,
     ) -> dict[str, Any]:
         """Build a submit body from an already-prepared pipeline spec."""
 
@@ -971,6 +1007,7 @@ def build_submit_body_from_spec(
             pipeline_path=pipeline_path,
             run_as=run_as,
             hydrate=hydrate,
+            skip_validation=skip_validation,
         ).to_body()
 
     def prepare_submit_payload(
@@ -982,6 +1019,7 @@ def prepare_submit_payload(
         hydrate: bool = True,
         run_as: str | None = None,
         resolution_overrides: dict[str, Any] | None = None,
+        skip_validation: bool = False,
     ) -> PipelineSubmitPayload:
         pipeline_spec = self.load_pipeline_for_submit(
             pipeline_path,
@@ -995,6 +1033,7 @@ def prepare_submit_payload(
             pipeline_path=pipeline_path,
             run_as=run_as,
             hydrate=hydrate,
+            skip_validation=skip_validation,
         )
 
     def build_submit_body(
@@ -1006,6 +1045,7 @@ def build_submit_body(
         hydrate: bool = True,
         run_as: str | None = None,
         resolution_overrides: dict[str, Any] | None = None,
+        skip_validation: bool = False,
     ) -> dict[str, Any]:
         return self.prepare_submit_payload(
             pipeline_path,
@@ -1014,6 +1054,7 @@ def build_submit_body(
             hydrate=hydrate,
             run_as=run_as,
             resolution_overrides=resolution_overrides,
+            skip_validation=skip_validation,
         ).to_body()
 
     @staticmethod
@@ -1117,6 +1158,7 @@ def submit_pipeline_spec(
         run_as: str | None = None,
         hydrate: bool = True,
         attempt: int = 1,
+        skip_validation: bool = False,
     ) -> dict[str, Any]:
         payload = self.prepare_submit_payload_from_spec(
             pipeline_spec,
@@ -1125,6 +1167,7 @@ def submit_pipeline_spec(
             pipeline_path=pipeline_path,
             run_as=run_as,
             hydrate=hydrate,
+            skip_validation=skip_validation,
         )
         return self.submit_prepared_payload(payload, pipeline_path=pipeline_path, attempt=attempt)
 
@@ -1138,6 +1181,7 @@ def submit_pipeline(
         run_as: str | None = None,
         resolution_overrides: dict[str, Any] | None = None,
         attempt: int = 1,
+        skip_validation: bool = False,
     ) -> dict[str, Any]:
         payload = self.prepare_submit_payload(
             pipeline_path,
@@ -1146,6 +1190,7 @@ def submit_pipeline(
             hydrate=hydrate,
             run_as=run_as,
             resolution_overrides=resolution_overrides,
+            skip_validation=skip_validation,
         )
         return self.submit_prepared_payload(payload, pipeline_path=pipeline_path, attempt=attempt)
 
@@ -1909,6 +1954,7 @@ def run_pipeline_spec(
         exit_on_first_failure: bool = False,
         metadata: dict[str, Any] | None = None,
         submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS,
+        skip_validation: bool = False,
     ) -> dict[str, Any]:
         """Submit/wait/retry an already hydrated/validated in-memory spec."""
 
@@ -1924,6 +1970,7 @@ def body_factory(
                 pipeline_path=pipeline_path,
                 run_as=run_as,
                 hydrate=hydrate,
+                skip_validation=skip_validation,
             ).to_body()
 
         return self._run_body_factory(
@@ -1960,6 +2007,7 @@ def run_pipeline(
         exit_on_first_failure: bool = False,
         metadata: dict[str, Any] | None = None,
         submit_recovery_attempts: int = _DEFAULT_SUBMIT_RECOVERY_ATTEMPTS,
+        skip_validation: bool = False,
     ) -> dict[str, Any]:
         """Submit (and optionally wait for) a pipeline with lifecycle hooks.
 
@@ -1980,6 +2028,7 @@ def body_factory(
                 hydrate=hydrate,
                 run_as=run_as,
                 resolution_overrides=resolution_overrides,
+                skip_validation=skip_validation,
             ).to_body()
 
         return self._run_body_factory(

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

@@ -109,24 +109,6 @@ def prepare_loaded_pipeline_spec(
 
         return pipeline_spec
 
-    def validate_pipeline_for_run(
-        self,
-        pipeline_spec: dict[str, Any],
-        *,
-        pipeline_path: str | Path,
-        effective_path: str | Path | None,
-        skip_validation: bool,
-    ) -> list[str]:
-        """Return validation errors for a prepared pipeline spec.
-
-        The OSS default intentionally does not enforce the local authoring
-        validator here: submit-time API validation remains the source of truth,
-        while downstreams can plug in stricter schema/input validators.
-        """
-
-        del pipeline_spec, pipeline_path, effective_path, skip_validation
-        return []
-
     def has_layout(self, pipeline_spec: Mapping[str, Any]) -> bool:
         """Return True when a pipeline graph already has non-zero coordinates."""
 
@@ -547,6 +529,7 @@ def body_factory(
                 pipeline_path=pipeline_path,
                 run_as=run_as,
                 hydrate=False,
+                skip_validation=skip_validation,
             )
             submit_payloads[attempt] = submit_payload
             return submit_payload.to_body()

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

@@ -89,8 +89,8 @@ def validate_pipeline_file(path: str | Path) -> dict[str, Any]:
     return pipeline
 
 
-def validate_pipeline_spec(pipeline: Mapping[str, Any]) -> None:
-    """Validate the OSS-compatible local pipeline shape.
+def collect_pipeline_spec_errors(pipeline: Mapping[str, Any]) -> list[str]:
+    """Return OSS-compatible local pipeline shape validation errors.
 
     This is a pragmatic validator for local authoring workflows. It focuses on
     the graph structure that the CLI commands consume rather than provider-specific
@@ -99,6 +99,13 @@ def validate_pipeline_spec(pipeline: Mapping[str, Any]) -> None:
 
     errors: list[str] = []
     _validate_root_pipeline(pipeline, errors)
+    return errors
+
+
+def validate_pipeline_spec(pipeline: Mapping[str, Any]) -> None:
+    """Validate the OSS-compatible local pipeline shape."""
+
+    errors = collect_pipeline_spec_errors(pipeline)
     if errors:
         details = "\n".join(f"- {error}" for error in errors)
         raise PipelineValidationError(f"Pipeline validation failed:\n{details}")

pyproject.toml

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

tests/test_packaging.py

@@ -178,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.2" in metadata
+    assert "Version: 0.1.3" in metadata
     assert "Requires-Dist: tangle-api==0.1.1" in requires_dist
     assert not any("extra == 'native'" in line for line in requires_dist)
     assert "Provides-Extra: native" in metadata