Add pipelines decompile: YAML -> Python pipeline source (v0.1.31)

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

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

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

@@ -235,6 +235,11 @@ def __init__(
             "header": header,
             "include_env_credentials": include_env_credentials,
         }
+        # Only an EXPLICIT logger is forwarded to an internally created client.
+        # Passing the default one would override the client's own verbosity
+        # gating and make ordinary non-verbose runs start logging.
+        if logger is not None:
+            self._client_options["logger"] = logger
         self.cache: dict[str, Any] = {}
         self.upgrade_deprecated = upgrade_deprecated
         self.verbose = verbose

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

@@ -23,6 +23,7 @@
     TokenOption,
 )
 from .logger import logger_for_log_type
+from .pipeline_decompiler import DecompileError, decompile_pipeline_file
 from .pipelines import (
     PipelineValidationError,
     compile_pipeline_file,
@@ -393,6 +394,135 @@ def pipelines_compile(
         print(f"warning: {warning}")
 
 
+@app.command(name="decompile")
+def pipelines_decompile(
+    pipeline_path: pathlib.Path,
+    *,
+    output: Annotated[
+        pathlib.Path,
+        Parameter(
+            name="--output",
+            alias="-o",
+            help=(
+                "Output path for the generated Python pipeline source. "
+                "Always replaced, atomically."
+            ),
+        ),
+    ],
+    verify: Annotated[
+        bool,
+        Parameter(
+            name="--verify",
+            help=(
+                "Recompile the generated source and require semantic "
+                "equality with the input. Disabling this removes the only "
+                "proof the output is faithful."
+            ),
+        ),
+    ] = True,
+    python_tasks: Annotated[
+        bool,
+        Parameter(
+            name="--python-tasks",
+            help=(
+                "Write components the Python component generator made, whose "
+                "source is recorded in full, as @task functions in "
+                "<stem>_tasks/. The source is copied, never run; compiling the "
+                "script later imports it. --no-python-tasks keeps every "
+                "component a digest-pinned YAML leaf."
+            ),
+        ),
+    ] = True,
+    local_fallbacks: Annotated[
+        bool,
+        Parameter(
+            name="--local-fallbacks",
+            negative="",
+            help=(
+                "Also keep a local copy of every component the library "
+                "verified, behind its published digest, so the output "
+                "compiles and hydrates with the library unreachable. By "
+                "default those components are referenced by digest alone."
+            ),
+        ),
+    ] = False,
+) -> None:
+    """Generate Python pipeline source from a pipeline YAML.
+
+    Accepts hydrated or dehydrated YAML. Components are always resolved and
+    re-pinned before code generation.
+    """
+
+    try:
+        result = decompile_pipeline_file(
+            pipeline_path, output, verify=verify, python_tasks=python_tasks, include_local_fallbacks=local_fallbacks
+        )
+    except DecompileError as exc:
+        raise SystemExit(str(exc)) from exc
+
+    print(
+        f"Decompiled {pipeline_path} -> {result.output_path} "
+        f"({result.task_count} task(s))."
+    )
+    if result.subgraph_module is not None:
+        print(f"Nested graphs are @pipeline functions in {result.subgraph_module}; keep it beside the script.")
+    if result.resolve_config is not None:
+        print(
+            f"Components resolve through {result.resolve_config}, with "
+            f"{len(result.component_files)} local copy(ies) in "
+            f"{result.components_dir}. Keep both beside the script: the "
+            "compiler resolves relative component URLs against the compiled "
+            "YAML's directory, so compile the script in place."
+        )
+    if result.digest_refs:
+        print(
+            f"{result.digest_refs} component(s) are referenced by their published digest "
+            "alone: hydrating the output needs the component library. Pass "
+            "--local-fallbacks to also keep local copies for offline use."
+        )
+    if result.tasks_package is not None:
+        print(
+            f"{result.python_tasks} Python component(s) are @task functions in "
+            f"{result.tasks_package}, holding source recovered verbatim from the "
+            "input. It was not run here: `compile` imports it, and hydrate/submit/"
+            "run execute it to rebuild each component (from the current "
+            "directory tree or a --trusted-source), so review it first. Those "
+            "components are no longer digest-pinned, and their rebuilt specs "
+            "carry new provenance annotations."
+        )
+    if result.image_id_tasks:
+        print(
+            f"{result.image_id_tasks} of them set @task(image_id=...): compile resolves it "
+            "in your environment (--image ID=REF, then registered defaults) and may "
+            "pick a different image than the input recorded. Use --no-python-tasks "
+            "for the digest-exact components."
+        )
+    if result.unwrap_tasks:
+        print(
+            f"{result.unwrap_tasks} of them use @task(unwrap=...): the flattened inputs' "
+            "names were checked, their types are derived again at compile."
+        )
+    kept = result.python_candidates - result.python_tasks
+    if kept > 0:
+        print(f"{kept} component(s) with recorded Python source stayed YAML leaves.")
+    if result.verified:
+        print(
+            "Verified: recompiling the output reproduces the input's "
+            "semantic digest. This is not byte identity \u2014 comments, key "
+            "order and task order are not recoverable."
+            + (
+                " For the Python components this proves the pipeline around "
+                "them, not the components Tangle will rebuild from source."
+                if result.tasks_package is not None
+                else ""
+            )
+        )
+    else:
+        print("warning: --no-verify was passed; the output is unproven.")
+    for note in result.normalizations:
+        print(f"normalization applied when comparing: {note}")
+
+
 @app.command(name="layout")
 def pipelines_layout(
     pipeline_path: pathlib.Path,

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

@@ -60,9 +60,15 @@ class SchemaValidationError(ValueError):
     """Raised when a dehydrated pipeline fails schema/contract validation.
 
     The compiler wraps this in a ``CompileError`` so the CLI exits 1 with
-    a friendly message; tests may assert on it directly.
+    a friendly message; tests may assert on it directly. ``location`` carries
+    the structural path when one is known, so a caller that must not echo the
+    rejected value has an alternative to parsing the message.
     """
 
+    def __init__(self, *args: Any, location: str | None = None) -> None:
+        super().__init__(*args)
+        self.location = location
+
 
 def _schema_path() -> Path:
     """Locate the packaged dehydrated schema JSON.
@@ -127,7 +133,8 @@ def validate_dehydrated_data(data: Mapping[str, Any]) -> None:
     )
     raise SchemaValidationError(
         f"dehydrated pipeline failed schema validation at {location}: "
-        f"{best.message}"
+        f"{best.message}",
+        location=location,
     )
 
 

pyproject.toml

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

tests/test_packaging.py

@@ -183,7 +183,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.30" in metadata
+    assert "Version: 0.1.31" 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