feat: consolidate pipeline validation

README.md

@@ -291,9 +291,11 @@ uv run tangle sdk pipeline-runs annotations set RUN_ID key value
 uv run tangle sdk pipeline-runs export RUN_ID --output pipeline.yaml
 ```
 
+`pipelines validate` checks local graph shape, the packaged Tangle pipeline JSON schema, and component input wiring when component specs are present. It does not hydrate refs; validate a hydrated file when you need fully resolved component-input checks for remote refs.
+
 `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.
+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; missing required component inputs and invalid task output references are rejected when component specs are available.
 
 ## Programmatic client
 

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

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

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

@@ -954,13 +954,14 @@ 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)
+        if not skip_validation:
+            validation_errors = self.hooks.validate_pipeline_for_run(
+                prepared_spec,
+                pipeline_path=pipeline_path,
+                effective_path=None,
+                skip_validation=False,
+            )
+            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"]

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

@@ -529,7 +529,7 @@ def body_factory(
                 pipeline_path=pipeline_path,
                 run_as=run_as,
                 hydrate=False,
-                skip_validation=skip_validation,
+                skip_validation=True,
             )
             submit_payloads[attempt] = submit_payload
             return submit_payload.to_body()

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

@@ -0,0 +1,21 @@
+"""Shared helpers for traversing Tangle pipeline spec dictionaries."""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+
+def _extract_task_output_refs(value: Any) -> set[str]:
+    """Return task ids referenced by nested taskOutput argument values."""
+
+    refs: set[str] = set()
+    if isinstance(value, Mapping):
+        task_output = value.get("taskOutput")
+        if isinstance(task_output, Mapping) and isinstance(task_output.get("taskId"), str):
+            refs.add(task_output["taskId"])
+        for nested in value.values():
+            refs.update(_extract_task_output_refs(nested))
+    elif isinstance(value, list):
+        for item in value:
+            refs.update(_extract_task_output_refs(item))
+    return refs