Describe generated components by logical path, not write location (v0.1.16)

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

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

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

@@ -2020,6 +2020,15 @@ def build_component_dict(
 # ============================================================================
 
 
+def _within(path: Path, root: Path) -> bool:
+    """Whether *path* lies inside *root*."""
+    try:
+        path.resolve().relative_to(root.resolve())
+    except ValueError:
+        return False
+    return True
+
+
 def generate_component_yaml(
     file_path: Path,
     output_path: Path,
@@ -2035,12 +2044,21 @@ def generate_component_yaml(
     emit_generation_annotations: bool = True,
     path_annotation_mode: Literal["oss", "td_legacy"] = "oss",
     unwrapped_inputs: dict[str, Any] | None = None,
+    logical_output_path: Path | None = None,
 ) -> bool:
     """Generate a component YAML file from a Python function.
 
     Args:
         file_path: Path to the Python source file
         output_path: Where to write the generated YAML
+        logical_output_path: Where the component is to be UNDERSTOOD to live,
+            for provenance only. Defaults to ``output_path``. Pass this when
+            the file is written somewhere incidental -- a private staging
+            directory, a scratch area -- so the recorded provenance describes
+            the component rather than the accident of where bytes landed.
+            Provenance is otherwise derived from the physical path, which
+            would embed that location and make the emitted YAML, and therefore
+            its digest, differ between runs of identical source.
         container_image: Docker image reference
         function_name: Function to extract (auto-detected if None)
         dependencies_from: Path to pyproject.toml with pip dependencies
@@ -2116,7 +2134,6 @@ def generate_component_yaml(
             deps = read_dependencies(dependencies_from)
 
         # 4. Build annotations
-        directory = file_path.parent.resolve()
         module_code = file_path.read_text()
 
         annotations: dict[str, str] = {
@@ -2148,9 +2165,18 @@ def generate_component_yaml(
         # basename-only paths outside a git checkout to preserve historical
         # snapshots.
         resolved_source = file_path.resolve()
-        resolved_output = output_path.resolve()
+        # Provenance describes where the component LIVES, which is not always
+        # where this call happens to write it. Kept UNRESOLVED: ``td_legacy``
+        # annotates the lexical basename, so resolving here would rewrite the
+        # recorded name whenever the output is a symlink.
+        annotation_output = logical_output_path if logical_output_path is not None else output_path
+        resolved_output = annotation_output.resolve()
         common_dir = Path(os.path.commonpath([resolved_source, resolved_output]))
-        git_root = get_git_root(directory)
+        # Discover from the SOURCE's real directory: a symlinked source file
+        # whose link lives outside the checkout would otherwise find no repo
+        # and drop every git annotation.
+        source_dir = resolved_source.parent
+        git_root = get_git_root(source_dir)
         use_common_paths = path_annotation_mode == "oss" or git_root is not None
 
         def _path_annotation(path: Path) -> str:
@@ -2163,7 +2189,7 @@ def _path_annotation(path: Path) -> str:
 
         if not strip_source_path:
             annotations["python_original_code_path"] = _path_annotation(file_path)
-        annotations["component_yaml_path"] = _path_annotation(output_path)
+        annotations["component_yaml_path"] = _path_annotation(annotation_output)
         if emit_generation_annotations:
             if dependencies_from:
                 annotations["tangle_cli_generation_dependencies_from"] = _path_annotation(dependencies_from)
@@ -2172,7 +2198,12 @@ def _path_annotation(path: Path) -> str:
 
         # Git info — use the same common ancestor as git_relative_dir when common paths are active.
         if git_root:
-            git_info = get_git_info(common_dir)
+            # Read the repository from a directory KNOWN to be inside it. The
+            # common ancestor of source and output need not be: any output
+            # outside the checkout (a configured output_folder, a staging dir)
+            # drags it out, and reading git there returns nothing, silently
+            # publishing a component with no repository of origin.
+            git_info = get_git_info(common_dir if _within(common_dir, git_root) else source_dir)
             git_info.pop("_git_root", None)
             # Override git_relative_dir to be the common ancestor
             try:
@@ -2181,7 +2212,7 @@ def _path_annotation(path: Path) -> str:
                 pass
             annotations.update(git_info)
         else:
-            git_info = get_git_info(directory)
+            git_info = get_git_info(source_dir)
             git_info.pop("_git_root", None)
             annotations.update(git_info)
 

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

@@ -128,6 +128,7 @@ def generate_component_yaml(
         resolve_root: Path | None = None,
         emit_generation_annotations: bool = True,
         unwrapped_inputs: dict[str, Any] | None = None,
+        logical_output_path: Path | None = None,
     ) -> bool:
         """Generate component YAML from a Python function source file.
 
@@ -147,13 +148,21 @@ def generate_component_yaml(
             unwrapped_inputs: Optional persisted unwrap schema. Hydrate forwards
                 this ``local_from_python.unwrapped_inputs`` payload so component
                 generation expands dict parameters exactly as compile did.
+            logical_output_path: Where the component is to be understood to
+                live, for provenance only. Defaults to ``output_path``.
 
         Returns:
             True when generation succeeds, otherwise False.
         """
 
         from tangle_cli.component_from_func import generate_component_yaml
 
+        # Forwarded only when set. These hops are overridable, and an older
+        # override written against the previous signature must keep working
+        # for every caller that does not use the seam.
+        seam: dict[str, Any] = (
+            {} if logical_output_path is None else {"logical_output_path": logical_output_path}
+        )
         return generate_component_yaml(
             file_path=file_path,
             output_path=output_path,
@@ -168,6 +177,7 @@ def generate_component_yaml(
             resolve_root=resolve_root,
             emit_generation_annotations=emit_generation_annotations,
             unwrapped_inputs=unwrapped_inputs,
+            **seam,
         )
 
     def regenerate_yaml(
@@ -184,6 +194,7 @@ def regenerate_yaml(
         resolve_root: Path | None = None,
         emit_generation_annotations: bool = True,
         unwrapped_inputs: dict[str, Any] | None = None,
+        logical_output_path: Path | None = None,
     ) -> bool:
         """Regenerate a YAML component from a Python function source file.
 
@@ -202,6 +213,10 @@ def regenerate_yaml(
             emit_generation_annotations: Whether to emit regeneration metadata.
             unwrapped_inputs: Optional ``local_from_python.unwrapped_inputs``
                 schema to preserve compile-time unwrap expansion at hydrate time.
+            logical_output_path: Where the component is to be understood to
+                live, for provenance only. Defaults to ``output_path``. Note
+                that the image is still read back from the PHYSICAL output, so
+                a caller writing to a fresh location supplies ``image``.
 
         Returns:
             True when regeneration succeeds, otherwise False.
@@ -218,6 +233,9 @@ def regenerate_yaml(
             self._log(f"  Found dependencies: {deps_file}")
 
         final_output.parent.mkdir(parents=True, exist_ok=True)
+        seam: dict[str, Any] = (
+            {} if logical_output_path is None else {"logical_output_path": logical_output_path}
+        )
         return self.run_generation(
             python_file=python_file,
             final_output=final_output,
@@ -231,6 +249,7 @@ def regenerate_yaml(
             resolve_root=resolve_root,
             emit_generation_annotations=emit_generation_annotations,
             unwrapped_inputs=unwrapped_inputs,
+            **seam,
         )
 
     def run_generation(
@@ -248,6 +267,7 @@ def run_generation(
         resolve_root: Path | None = None,
         emit_generation_annotations: bool = True,
         unwrapped_inputs: dict[str, Any] | None = None,
+        logical_output_path: Path | None = None,
     ) -> bool:
         """Execute component generation and clean up partial output on failure.
 
@@ -265,6 +285,8 @@ def run_generation(
             emit_generation_annotations: Whether to emit regeneration metadata.
             unwrapped_inputs: Optional persisted unwrap schema forwarded to the
                 low-level generator for compile/hydrate interface parity.
+            logical_output_path: Where the component is to be understood to
+                live, for provenance only. Defaults to ``final_output``.
 
         Returns:
             True when generation succeeds, otherwise False. On failure, any
@@ -274,6 +296,9 @@ def run_generation(
         try:
             function_detail = f" function {func_name!r}" if func_name else ""
             self._log(f"  Generating component from {python_file.name}{function_detail}...")
+            seam: dict[str, Any] = (
+                {} if logical_output_path is None else {"logical_output_path": logical_output_path}
+            )
             success = self.generate_component_yaml(
                 file_path=python_file,
                 output_path=final_output,
@@ -287,6 +312,7 @@ def run_generation(
                 resolve_root=resolve_root,
                 emit_generation_annotations=emit_generation_annotations,
                 unwrapped_inputs=unwrapped_inputs,
+                **seam,
             )
             if not success:
                 self._log("  ❌ Failed to generate component", err=True)
@@ -340,6 +366,7 @@ def regenerate_yaml(
     resolve_root: Path | None = None,
     logger: Any | None = None,
     unwrapped_inputs: dict[str, Any] | None = None,
+    logical_output_path: Path | None = None,
 ) -> bool:
     """Regenerate component YAML through the default generator.
 
@@ -358,11 +385,16 @@ def regenerate_yaml(
         logger: Optional logger object used by the generator.
         unwrapped_inputs: Optional persisted unwrap schema from
             ``local_from_python.unwrapped_inputs``.
+        logical_output_path: Where the component is to be understood to live,
+            for provenance only. Defaults to ``output_path``.
 
     Returns:
         True when regeneration succeeds, otherwise False.
     """
 
+    seam: dict[str, Any] = (
+        {} if logical_output_path is None else {"logical_output_path": logical_output_path}
+    )
     return ComponentGenerator(logger=logger, verbose=verbose).regenerate_yaml(
         python_file=python_file,
         output_path=output_path,
@@ -375,6 +407,7 @@ def regenerate_yaml(
         mode=mode,
         resolve_root=resolve_root,
         unwrapped_inputs=unwrapped_inputs,
+        **seam,
     )
 
 

pyproject.toml

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