From 142ae2a6735103e697bf1f69f6faf5546ccb78f1 Mon Sep 17 00:00:00 2001 From: Volv G Date: Wed, 16 Sep 2026 10:24:40 -0700 Subject: [PATCH 1/2] Dedup @task sidecar entries by component identity (v0.1.14) A pipeline that reaches one shared `@task` function from several call sites with different task-level options silently ran every one of them on the FIRST call site's configuration. `_build_local_from_python_components` keyed the generated `.components.yaml` by the hyphenated FUNCTION NAME, so a shared `run_dbt` decorated once with a slim image and once with a fat image produced a single `run-dbt` entry, and `_rewrite_task_componentref_urls` recomputed that same name per task -- pointing every task at the survivor. Nothing failed: the pipeline compiled, validated, and ran, on the wrong image. The name was never an identity. What hydrate regenerates from an entry is determined by the whole `local_from_python` block -- image (explicit or resolved `image_id`), function, mode, resolve_root, dependencies_from, the persisted unwrap schema, and the source file -- so that block, plus the module-qualified function identity, is what dedup must key on. `_plan_task_sidecar` now computes both halves at once and returns a `TaskSidecarPlan(entries, fragment_by_task)`. Sidecar emission and componentRef rewriting consume the SAME map, so the two cannot drift: a fragment depends on how many distinct components a function generates pipeline-wide, which no per-call-site recomputation can know. Identity is module-qualified, not path-shaped and not `__module__`. The compile driver imports pipeline scripts as `_tangle_user_pipeline_`, so runtime `__module__` is a fresh string every compile for a script-defined task while being a real dotted name for an imported one -- unusable. The namespace is derived from the source LAYOUT instead, which is what an import would have produced anyway: walk up while `__init__.py` exists (`package_a/tasks.py` -> `package_a.tasks`, `pkg/__init__.py` -> `pkg`), and record the first non-package ancestor relative to the pipeline source directory. `__qualname__` rides along to separate same-named functions nested in different scopes. Consequences: * Repeated identical calls still dedup to exactly one entry. * One function with two configurations emits two entries, and each graph task is rewritten to its own fragment. * `package_a.tasks.run` and `package_b.tasks.run` are two components and both are emitted. The old hard `CompileError` for "two distinct @task source files map to the same sidecar fragment" is removed: it existed to prevent one file silently shadowing the other, and distinct fragments satisfy that intent without rejecting legitimate authoring. Fragment naming keeps the readable name where it is unambiguous. A base with exactly ONE identity stays `run-dbt` (or `combine--` for an unwrapped task), so existing single-component pipelines emit byte-identical sidecars. Once a base collides, EVERY variant is suffixed `--` -- no arbitrary first-traced variant keeps the bare name -- and colliding variants are emitted in sorted order so the sidecar text does not depend on trace order. The hash is a SHA-256 prefix over canonical JSON of the identity payload: never `hash()`, never dict or API ordering. Identity paths are anchored at the pipeline's own source directory rather than the output directory, so fragment names survive relocating the project and compiling into a different output directory (the hidden submit bundle compiles elsewhere by design). Emitted `local_from_python` paths stay relative to the sidecar as before. No image, registry, tag, digest, or source-path text reaches a fragment name; the diagnostic for an internal digest collision renders `module::qualname` only. `_build_local_from_python_components` is deleted rather than kept as a wrapper. After this change it had no production caller, and an underscore-private function is not an API worth preserving for downstream tests. `_plan_task_sidecar` requires `identity_root` explicitly for the same reason: defaulting it to the output directory would quietly reintroduce output-dir-dependent fragment names. `examples/python_pipeline/dedup_image_variants/` is a runnable manual check -- one shared helper decorated three times, two images -- whose docstring states the compile command and the exact fragments, task refs, and no-leakage assertions to expect. Co-authored-by: Claude Opus 5 Assisted-By: devx/43e17814-42d0-4ca6-91b1-567962ce0219 --- README.md | 4 + .../dedup_image_variants/__init__.py | 7 + .../dedup_image_variants/pipeline.py | 84 +++ .../dedup_image_variants/shared_dbt.py | 16 + .../tangle-cli/src/tangle_cli/__init__.py | 2 +- .../src/tangle_cli/pipeline_compiler.py | 518 +++++++++++++----- pyproject.toml | 2 +- tests/test_packaging.py | 2 +- tests/test_pipeline_compiler.py | 455 +++++++++++++++ uv.lock | 2 +- 10 files changed, 963 insertions(+), 129 deletions(-) create mode 100644 examples/python_pipeline/dedup_image_variants/__init__.py create mode 100644 examples/python_pipeline/dedup_image_variants/pipeline.py create mode 100644 examples/python_pipeline/dedup_image_variants/shared_dbt.py diff --git a/README.md b/README.md index f8fad0e..af9d42d 100644 --- a/README.md +++ b/README.md @@ -457,6 +457,10 @@ Notes: A minimal graph uses `@pipeline` for the graph and `@task` for local Python components. `@task` functions are not executed at compile time; the compiler records call sites, emits a sibling `.components.yaml` with `local_from_python` entries, and rewrites task component refs to that sidecar. Hydrate later regenerates the same component YAML from the Python source. +Sidecar entries are deduplicated by the **generated component**, not by function name. The dedup key is the module-qualified function identity — a logical module namespace derived from the project-relative source layout (`package_a/tasks.py` -> `package_a.tasks`, `pkg/__init__.py` -> `pkg`), plus `__qualname__` and the function name — together with every generation-affecting option: image (explicit or resolved `image_id`), `mode`, `resolve_root`, `dependencies_from`, the `unwrap` schema, and the source file. Runtime `__module__` is deliberately not used, because pipeline scripts are imported under throwaway UUID module names. + +Call sites sharing that whole identity collapse to one entry keyed by the readable hyphenated function name (`run_dbt` -> `run-dbt`). Otherwise every colliding variant — a shared helper re-decorated with different task-level options, or `package_a.tasks.run` alongside `package_b.tasks.run` — is emitted as `run-dbt--`, where `` is a content digest of the canonical identity; no variant keeps the unsuffixed name, and each task ref points at its own. Digests are anchored at the project source directory, so fragment names are unchanged by relocating the project, compiling into a different output directory, or reordering the pipeline's calls, and they never embed image, path, or credential-bearing values. + ```python from cloud_pipelines import components from tangle_cli.python_pipeline import In, Out, pipeline, task diff --git a/examples/python_pipeline/dedup_image_variants/__init__.py b/examples/python_pipeline/dedup_image_variants/__init__.py new file mode 100644 index 0000000..13fc21f --- /dev/null +++ b/examples/python_pipeline/dedup_image_variants/__init__.py @@ -0,0 +1,7 @@ +"""Package marker so the shared helper has a package-qualified module identity. + +With this file present the compiler derives the logical module namespace +``dedup_image_variants.shared_dbt`` from the source layout (it walks up while +``__init__.py`` exists). Without it the helper would still work, but its module +identity would be the bare ``shared_dbt``. +""" diff --git a/examples/python_pipeline/dedup_image_variants/pipeline.py b/examples/python_pipeline/dedup_image_variants/pipeline.py new file mode 100644 index 0000000..2e5306f --- /dev/null +++ b/examples/python_pipeline/dedup_image_variants/pipeline.py @@ -0,0 +1,84 @@ +"""Manual-test example: per-variant @task sidecar dedup for ONE shared function. + +The SAME module-qualified function -- ``dedup_image_variants.shared_dbt.run_dbt`` +-- is re-decorated three times: twice with distinct task-level images, and once +more with an image identical to the first. Before the dedup fix the sidecar was +keyed by function NAME, so all three collapsed into the first entry and every +task silently resolved to the first call site's image. + +Compile from the repository root with:: + + uv run tangle sdk pipelines compile \ + examples/python_pipeline/dedup_image_variants/pipeline.py \ + --pipeline dedup_image_variants_pipeline \ + --output /tmp/tangle-dedup-demo/pipeline.yaml + +Then inspect the two generated artifacts:: + + cat /tmp/tangle-dedup-demo/pipeline.components.yaml + cat /tmp/tangle-dedup-demo/pipeline.yaml + +Expected assertions +------------------- +1. ``pipeline.components.yaml`` has EXACTLY TWO entries, both hashed: + ``run-dbt--<10 hex chars>``. Neither keeps the bare ``run-dbt`` name -- + once a base collides, every variant is suffixed, so there is no arbitrary + "first variant wins". +2. The two entries differ ONLY in ``local_from_python.image`` + (``python:3.12-slim`` vs ``python:3.12``); both carry + ``function: run_dbt`` and the same ``file: ./shared_dbt.py``. +3. Each graph task's ``componentRef.url`` is + ``resolve://./pipeline.components.yaml#run-dbt--`` pointing at the + fragment whose image matches that task's decorator: + ``daily_orders`` and ``hourly_sessions`` -> the SLIM entry (they share one + identity and therefore one fragment -- repeated identical calls still + dedup), ``backfill_orders`` -> the FAT entry. +4. No raw image, registry, tag, digest, or source-path text appears in any + fragment name: the suffix is a SHA-256 prefix over the canonical identity, + so ``python``, ``slim``, ``3.12``, ``shared_dbt`` and ``.py`` must NOT be + substrings of any sidecar key. +5. Fragment names are stable: recompiling to a different ``--output`` + directory, or moving this example tree elsewhere, produces the SAME two + fragment names (identity is anchored at the pipeline source directory, not + the output directory and not an absolute machine path). + +Observed output on this revision (the digests are derived only from +project-relative values, so they reproduce on any checkout):: + + run-dbt--b196ad73a4 image: python:3.12-slim + run-dbt--6024a73044 image: python:3.12 + + daily_orders -> resolve://./pipeline.components.yaml#run-dbt--b196ad73a4 + hourly_sessions -> resolve://./pipeline.components.yaml#run-dbt--b196ad73a4 + backfill_orders -> resolve://./pipeline.components.yaml#run-dbt--6024a73044 + +Note on ``file:``: sidecar paths are relative to the OUTPUT directory, so +compiling into ``/tmp`` (outside the source tree) writes a long ``../../..`` +path to ``shared_dbt.py``. That is expected and is exactly the point of +assertion 5 -- the emitted path tracks the output directory while the fragment +NAMES do not. Compile next to the script if you want short paths. +""" + +from tangle_cli.python_pipeline import Out, pipeline, task + +from shared_dbt import run_dbt + +# Three decorations of ONE function. ``dbt_slim`` and ``dbt_slim_again`` are +# byte-identical configurations, so they share a single generated component; +# ``dbt_fat`` differs in image, so it is a genuinely different component. +dbt_slim = task(image="python:3.12-slim")(run_dbt) +dbt_slim_again = task(image="python:3.12-slim")(run_dbt) +dbt_fat = task(image="python:3.12")(run_dbt) + + +@pipeline("Dedup image variants demo") +def dedup_image_variants_pipeline() -> Out[str]: + # Two call sites, one image -> ONE sidecar entry, two task refs to it. + daily = dbt_slim.named("daily_orders")(model="orders_daily") + hourly = dbt_slim_again.named("hourly_sessions")(model="sessions_hourly") + + # Same function, different image -> its OWN sidecar entry and task ref. + backfill = dbt_fat.named("backfill_orders")(model=daily, target="backfill") + + print(hourly) + return backfill diff --git a/examples/python_pipeline/dedup_image_variants/shared_dbt.py b/examples/python_pipeline/dedup_image_variants/shared_dbt.py new file mode 100644 index 0000000..ef86c1f --- /dev/null +++ b/examples/python_pipeline/dedup_image_variants/shared_dbt.py @@ -0,0 +1,16 @@ +"""One shared dbt-style helper, imported and re-decorated by ``pipeline.py``. + +This module exists so the example exercises the REAL shape of the bug: a single +module-qualified function (``dedup_image_variants.shared_dbt.run_dbt``) reached +from several call sites with different task-level images, rather than two +separate functions that happen to look alike. +""" + + +def run_dbt(model: str, target: str = "prod") -> str: + """Run one dbt model. + + Metadata: + Name: Run Dbt Model + """ + return f"{model}@{target}" diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index f102116..a4243cf 100644 --- a/packages/tangle-cli/src/tangle_cli/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/__init__.py @@ -14,6 +14,6 @@ try: __version__ = metadata_version("tangle-cli") except PackageNotFoundError: - __version__ = "0.1.13" + __version__ = "0.1.14" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py b/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py index 9685856..27f45e0 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py @@ -479,17 +479,25 @@ def _compile_pipeline_fn( ) if task_refs: components_path = output_path.with_name(output_path.stem + ".components.yaml") - components_entries = _build_local_from_python_components( + # ONE plan drives both halves: entries and componentRef fragments must + # agree, and a fragment depends on how many distinct components the + # function generates pipeline-wide (a shared helper called with + # different task-level images emits one entry per configuration). + sidecar_plan = _plan_task_sidecar( task_refs, components_yaml_dir=components_path.parent, + # Identity is anchored at the pipeline SOURCE dir, not the output + # dir, so fragment names survive compiling elsewhere (e.g. the + # hidden submit bundle) and relocating the project. + identity_root=base_dir, image_overrides=ctx.image_overrides, unwrapped_input_keys=builder.task_unwrapped_input_keys, ) + components_entries = sidecar_plan.entries _rewrite_task_componentref_urls( body_dict=body_dict, - task_refs=task_refs, + fragment_by_task=sidecar_plan.fragment_by_task, components_yaml_name=components_path.name, - unwrapped_input_keys=builder.task_unwrapped_input_keys, ) # 4b. A CHILD artifact is written under ``.subgraphs/``, away from @@ -1155,6 +1163,21 @@ def _unwrapped_schema_for_task( raise CompileError(str(exc)) from exc +def _stable_payload_hash(payload: Any) -> str: + """Short deterministic SHA-256 prefix for a JSON-serialisable payload. + + Args: + payload: Any JSON-serialisable structure. Mapping keys are sorted so + the digest never depends on Python dict insertion order, and the + digest is stable across processes (unlike :func:`hash`). + + Returns: + A 10-character lowercase hex digest prefix. + """ + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest()[:10] + + def _unwrapped_schema_hash(schema: Mapping[str, Any]) -> str: """Hash a persisted unwrap schema for sidecar fragment disambiguation. @@ -1167,26 +1190,46 @@ def _unwrapped_schema_hash(schema: Mapping[str, Any]) -> str: ``combine--0123abcd`` so different key sets for the same function do not collide in the generated components sidecar. """ - payload = json.dumps(schema, sort_keys=True, separators=(",", ":"), default=str) - return hashlib.sha256(payload.encode("utf-8")).hexdigest()[:10] + return _stable_payload_hash(schema) + + +def _task_fragment_base(ref: CallableRef) -> str: + """Return the LEGACY, un-disambiguated sidecar fragment for a ``@task``. + + Args: + ref: The ``@task`` callable reference. + + Returns: + The hyphenated function name, matching the ``local_from_python`` + resolver's output-filename convention (``my_task`` -> ``my-task``). + This is the fragment used verbatim when the function generates exactly + ONE component across the whole pipeline; see + :func:`_plan_task_sidecar` for the multi-variant disambiguation. + """ + assert ref._task_function_name is not None # only @task refs reach here + return ref._task_function_name.replace("_", "-") def _fragment_for_task(ref: CallableRef, unwrapped_schema: Mapping[str, Any] | None = None) -> str: - """Return the stable components-sidecar fragment for a ``@task`` call. + """Return the legacy single-variant fragment for a ``@task`` call. Args: ref: The ``@task`` callable reference. unwrapped_schema: Optional persisted unwrap schema for this call site. Returns: - The hyphenated function name for normal tasks, matching the - ``local_from_python`` resolver's output-filename convention - (``my_task`` -> ``my-task``). For unwrapped tasks, returns the function - fragment plus a schema hash so different key sets do not collide into - one component schema. + The hyphenated function name for normal tasks. For unwrapped tasks, + returns the function fragment plus a schema hash so different key sets + do not collide into one component schema. + + Note: + This is only the SINGLE-VARIANT naming. When one function generates + several DIFFERENT components in one pipeline (e.g. a shared ``run_dbt`` + called with different task-level ``image``/``dependencies_from``), + :func:`_plan_task_sidecar` appends a component-identity hash instead so + the variants do not collapse into the first one emitted. """ - assert ref._task_function_name is not None # only @task refs reach here - base = ref._task_function_name.replace("_", "-") + base = _task_fragment_base(ref) if unwrapped_schema: return f"{base}--{_unwrapped_schema_hash(unwrapped_schema)}" return base @@ -1530,163 +1573,388 @@ def _relocate_relative_local_url(url: str, source_dir: Path, sidecar_dir: Path) return None -def _build_local_from_python_components( +@dataclass(frozen=True) +class TaskSidecarPlan: + """Resolved ``@task`` sidecar entries plus the per-task fragment map. + + Attributes: + entries: The ``.components.yaml`` content, ``{fragment: + {local_from_python: {...}}}``, in first-seen task order. + fragment_by_task: Graph ``task_id`` -> the sidecar fragment that task's + ``componentRef`` must point at. Sidecar emission and componentRef + rewriting MUST share this map: fragment naming depends on how many + distinct components one function generates across the whole + pipeline, so it cannot be recomputed per call site in isolation. + """ + + entries: dict[str, Any] + fragment_by_task: dict[str, str] + + +def _local_from_python_payload( + ref: CallableRef, + *, + source: Path, + components_yaml_dir: Path, + image_overrides: Mapping[str, str] | None, + unwrapped_schema: Mapping[str, Any], +) -> dict[str, Any]: + """Build the ``local_from_python`` block for ONE traced ``@task`` call. + + Every field that changes what ``regenerate_yaml`` produces at hydrate time + belongs in this payload — the payload IS the component's identity (see + :func:`_plan_task_sidecar`). + + Args: + ref: The ``@task`` callable reference for this call site. + source: The ``@task`` source file (already existence-checked). + components_yaml_dir: Directory the generated sidecar will live in; + local paths are emitted relative to it. + image_overrides: Optional compile-time ``--image ID=REF`` overrides. + unwrapped_schema: Persisted unwrap schema for this call site, or an + empty mapping when the task does not use ``@task(unwrap=...)``. + + Returns: + The ``local_from_python`` mapping for this call site. + + Raises: + CompileError: when ``image_id`` does not resolve, or a referenced + ``resolve_root`` / ``dependencies_from`` path is unreachable. + """ + local_from_python: dict[str, Any] = {} + if ref._task_image is not None: + local_from_python["image"] = ref._task_image + elif ref._task_image_id is not None: + resolved_image = resolve_image_id(ref._task_image_id, image_overrides) + if resolved_image is None: + raise CompileError( + f"@task image_id={ref._task_image_id!r} on function " + f"{ref._task_function_name!r} did not resolve to an image. " + f"Pass --image {ref._task_image_id}=IMAGE to `tangle sdk pipelines compile`, " + f"or register a default with register_image_id({ref._task_image_id!r}, IMAGE)." + ) + local_from_python["image"] = resolved_image + # Always pin the function name. Without it the hydrator defaults + # to the file stem and extracts the wrong symbol. + assert ref._task_function_name is not None + local_from_python["function"] = ref._task_function_name + if ref._task_mode is not None: + local_from_python["mode"] = ref._task_mode + if ref._task_resolve_root is not None: + resolve_root = ref._task_resolve_root + if not resolve_root.exists(): + raise CompileError( + f"@task resolve_root is unreachable: {resolve_root}. " + "Point resolve_root at an existing directory or drop it." + ) + local_from_python["resolve_root"] = _relpath_posix(resolve_root, components_yaml_dir) + if ref._task_dependencies_from is not None: + deps = ref._task_dependencies_from + if not deps.exists(): + raise CompileError( + f"@task dependencies_from file is unreachable: {deps}. " + "Point dependencies_from at an existing file or drop it." + ) + local_from_python["dependencies_from"] = _relpath_posix(deps, components_yaml_dir) + if unwrapped_schema: + local_from_python["unwrapped_inputs"] = dict(unwrapped_schema) + local_from_python["file"] = _relpath_posix(source, components_yaml_dir) + return local_from_python + + +def _logical_module_id(source: Path, *, identity_root: Path) -> str: + """Derive a STABLE logical module namespace for a ``@task`` source file. + + Runtime ``__module__`` is unusable as an identity: the compile driver + imports the pipeline script under a throwaway + ``_tangle_user_pipeline_`` module name, so a ``@task`` defined in the + script would get a different "module" on every compile while a ``@task`` + imported from a sibling helper would get a real dotted name. The namespace + is therefore derived from the source LAYOUT instead, which is what an + import would have produced anyway: + + * Walk up while the directory is a package (``__init__.py`` present) to + build the dotted package chain — ``pkg_a/tasks.py`` -> ``pkg_a.tasks``, + and ``pkg_a/__init__.py`` -> ``pkg_a`` (the ``__init__`` segment is the + package itself, never a child module). + * The first NON-package ancestor is the import root. Two identically named + modules under different import roots are genuinely different modules, so + that root is recorded as a ``:`` prefix, expressed RELATIVE to + ``identity_root`` (never an absolute machine path). + + Args: + source: Absolute path of the ``@task`` source file. + identity_root: Stable project anchor (the pipeline's own source + directory) that the import root is expressed relative to. + + Returns: + ``":"``, e.g. ``".:pkg_a.tasks"`` or + ``"./src:pipeline"``. Stable when the project is relocated or compiled + into a different output directory. + + Raises: + CompileError: when no relative path to ``identity_root`` can be formed + (see :func:`_relpath_posix`). + """ + parts: list[str] = [] + if source.stem != "__init__": + parts.append(source.stem) + directory = source.parent + while (directory / "__init__.py").exists(): + parts.append(directory.name) + parent = directory.parent + if parent == directory: # filesystem root — stop rather than loop + break + directory = parent + dotted = ".".join(reversed(parts)) + return f"{_relpath_posix(directory, identity_root)}:{dotted}" + + +def _task_component_identity( + ref: CallableRef, + *, + source: Path, + identity_root: Path, + image_overrides: Mapping[str, str] | None, + unwrapped_schema: Mapping[str, Any], +) -> dict[str, Any]: + """Build the canonical dedup identity for one traced ``@task`` call. + + The identity is the MODULE-QUALIFIED FUNCTION IDENTITY plus every + generation-affecting option. Two call sites produce the same component — + and therefore share one sidecar entry — exactly when this payload matches. + + Paths inside the identity are anchored at ``identity_root`` (the pipeline's + own source directory) rather than at the sidecar directory, so the identity + is invariant under relocating the project or compiling into a different + output directory, and never embeds an absolute machine path. + + Args: + ref: The ``@task`` callable reference for this call site. + source: The ``@task`` source file. + identity_root: Stable project anchor for path-derived identity parts. + image_overrides: Optional compile-time ``--image ID=REF`` overrides. + unwrapped_schema: Persisted unwrap schema, or an empty mapping. + + Returns: + A JSON-serialisable identity payload. Hash it with + :func:`_stable_payload_hash`; render it with + :func:`_task_identity_label` for diagnostics. + + Raises: + CompileError: see :func:`_local_from_python_payload` and + :func:`_logical_module_id`. + """ + # ``__qualname__`` is forwarded onto the ref by @task. It distinguishes + # same-named functions nested in different scopes within ONE module; + # it equals the function name for the module-level authoring surface. + qualname = getattr(ref, "__qualname__", None) or ref._task_function_name + return { + "module": _logical_module_id(source, identity_root=identity_root), + "qualname": qualname, + "function": ref._task_function_name, + # Generation config, anchored at the stable project root. This mirrors + # the emitted block field-for-field so a new generation-affecting + # option cannot be added to the sidecar without also splitting dedup. + "generation": _local_from_python_payload( + ref, + source=source, + components_yaml_dir=identity_root, + image_overrides=image_overrides, + unwrapped_schema=unwrapped_schema, + ), + } + + +def _task_identity_label(identity_payload: Mapping[str, Any]) -> str: + """Human-readable ``::`` label for an identity payload. + + Args: + identity_payload: A :func:`_task_component_identity` payload. + + Returns: + A diagnostic label. Only the logical module/function identity is + rendered — never image, path, or credential-bearing generation values. + """ + module = str(identity_payload.get("module", "")) + _root, _sep, dotted = module.partition(":") + qualname = identity_payload.get("qualname") or identity_payload.get("function") + return f"{dotted or module}::{qualname}" + + +def _plan_task_sidecar( task_refs: list[tuple[str, CallableRef]], *, components_yaml_dir: Path, + identity_root: Path, image_overrides: Mapping[str, str] | None = None, unwrapped_input_keys: Mapping[str, dict[str, list[str]]] | None = None, -) -> dict[str, Any]: - """Build the ``.components.yaml`` content for @task refs. +) -> TaskSidecarPlan: + """Plan the ``.components.yaml`` entries and per-task fragments. + + Dedup is by GENERATED COMPONENT IDENTITY — module-qualified function + identity plus every generation-affecting option — never by bare function + name. Two call sites collapse into one entry only when the whole + :func:`_task_component_identity` payload matches: logical module namespace, + ``__qualname__``, function name, image (explicit or resolved ``image_id``), + mode, resolve_root, dependencies_from, the persisted unwrap schema, and the + source file. Consequently: + + * A shared helper such as ``run_dbt`` invoked with different task-level + images or dependency files emits one entry PER distinct configuration, + and each graph task is rewritten to its OWN fragment instead of silently + inheriting the first call site's component. + * ``pkg_a/tasks.py::run`` and ``pkg_b/tasks.py::run`` are different + components and both get emitted. This is NOT an error: the bare function + name is not an identity, so two packages may each define ``run``. + * Repeated identical calls still dedup to exactly one entry. + + Fragment naming: + + * Exactly ONE identity for a function's readable base -> the legacy + fragment (``run_dbt`` -> ``run-dbt``, or ``combine--`` for + an unwrapped task). Existing single-component pipelines keep + byte-identical sidecars. + * SEVERAL identities sharing one base -> ``--`` for EVERY + colliding variant; no arbitrary "first" variant keeps the unsuffixed + name. ``identity`` is a SHA-256 prefix over the canonical JSON identity + payload, so it is content-addressed (never :func:`hash`, never + dict/API/trace ordering), stable across processes, invariant under + project relocation and output-directory changes, and it never leaks + image, path, or credential-bearing values into the fragment name. + Colliding variants are also EMITTED in sorted fragment order so the + sidecar text does not depend on which variant was traced first. Args: task_refs: Traced ``(task_id, CallableRef)`` records for ``@task`` calls in the pipeline graph. components_yaml_dir: Directory where the generated sidecar will live; - local paths are written relative to this directory. + emitted local paths are written relative to this directory. + identity_root: Stable project anchor for path-derived IDENTITY parts — + the pipeline's OWN source directory. Required (not defaulted to + ``components_yaml_dir``) because anchoring identity at the output + directory would make fragment names change when the same pipeline + is compiled elsewhere, e.g. into the hidden submit bundle. image_overrides: Optional compile-time ``--image ID=REF`` overrides. unwrapped_input_keys: Optional trace metadata for ``@task(unwrap=...)`` - calls. When present, this function writes - ``local_from_python.unwrapped_inputs`` so hydrate can regenerate the - exact same flattened input schema without call-site context. + calls. When present, ``local_from_python.unwrapped_inputs`` is + written so hydrate regenerates the same flattened input schema + without call-site context. Returns: - An ordered map ``{fragment: {name?, local_from_python: - {image?, function, mode?, resolve_root?, dependencies_from?, file}}}``, DEDUPED by FUNCTION - (the fragment = hyphenated function name). The SAME @task function - called from multiple task sites collapses to one entry; TWO DISTINCT - @task functions defined in ONE file each get their own entry (they - share the same ``file:`` but carry different ``function:`` keys and - distinct fragments). This matches how ``_rewrite_task_componentref_urls`` - points each task at its OWN function fragment — deduping by source path - instead would drop every function but the first and leave the others' - ``resolve://...#`` refs dangling. - - The ``function`` field is always emitted so hydrate's - ``regenerate_yaml`` extracts the right function: it otherwise defaults - to the file STEM, which is wrong whenever the @task function name - differs from the source filename (the common case). - - Paths in ``local_from_python.{file,dependencies_from}`` are POSIX and - relative to ``components_yaml_dir`` so the sidecar is portable: as - long as the layout under that directory matches at compile- and - hydrate-time, the paths resolve correctly. + A :class:`TaskSidecarPlan`. Paths in + ``local_from_python.{file,dependencies_from,resolve_root}`` are POSIX + and relative to ``components_yaml_dir`` so the bundle stays portable. + + The component ``name`` is NOT emitted. A top-level ``name`` on a + resolve entry means "resolve a published component by this name" to the + hydrator (``PipelineHydrator._resolve_primary``), which would let a + same-named library component silently win over this local ``@task``. + The component's name comes from its source docstring + (``Metadata: Name:``) at hydrate time, read by ``regenerate_yaml``. Raises: - CompileError: when two distinct source files map to the same - fragment (function-name collision), when a referenced local - file (the @task source or its ``dependencies_from``) is - unreachable, or when a relative path cannot be formed (see - :func:`_relpath_posix`). + CompileError: when a referenced local file (the ``@task`` source or its + ``dependencies_from`` / ``resolve_root``) is unreachable, or when a + relative path cannot be formed (see :func:`_relpath_posix`). """ - seen_fragments: dict[str, Path] = {} - entries: dict[str, Any] = {} + checked_sources: set[Path] = set() + # base fragment -> identity digest -> (emitted payload, identity payload). + variants: dict[str, dict[str, tuple[dict[str, Any], dict[str, Any]]]] = {} + # (base, identity digest) -> the legacy single-variant fragment. + legacy_fragments: dict[tuple[str, str], str] = {} + task_identities: list[tuple[str, str, str]] = [] + for task_id, ref in task_refs: source = ref._task_source_path if source is None: # defensive — only @task refs are recorded here continue - unwrapped_schema = _unwrapped_schema_for_task(ref, task_id, unwrapped_input_keys) - fragment = _fragment_for_task(ref, unwrapped_schema) - - prior_source = seen_fragments.get(fragment) - if prior_source is not None: - # Already emitted this fragment. Fine when it is the SAME source - # (the same @task called from multiple sites). A DIFFERENT - # source sharing the function name would silently collide on the - # resolve:// fragment, so reject it loudly. - if prior_source != source: + if source not in checked_sources: + if not source.exists(): raise CompileError( - "two distinct @task source files map to the same sidecar " - f"fragment {fragment!r}: {prior_source} and {source}. " - "Rename one of the @task functions so each has a unique " - "name (the function name becomes the resolve:// fragment)." + f"@task source file is unreachable: {source}. Compile into " + "the pipeline source directory, or reference the component " + "with an absolute file://, gs://, or resolve:// URL." ) - continue + checked_sources.add(source) - if not source.exists(): - raise CompileError( - f"@task source file is unreachable: {source}. Compile into " - "the pipeline source directory, or reference the component " - "with an absolute file://, gs://, or resolve:// URL." - ) + unwrapped_schema = _unwrapped_schema_for_task(ref, task_id, unwrapped_input_keys) + base = _task_fragment_base(ref) + emitted = _local_from_python_payload( + ref, + source=source, + components_yaml_dir=components_yaml_dir, + image_overrides=image_overrides, + unwrapped_schema=unwrapped_schema, + ) + identity_payload = _task_component_identity( + ref, + source=source, + identity_root=identity_root, + image_overrides=image_overrides, + unwrapped_schema=unwrapped_schema, + ) + identity = _stable_payload_hash(identity_payload) + variants.setdefault(base, {}).setdefault(identity, (emitted, identity_payload)) + legacy_fragments.setdefault((base, identity), _fragment_for_task(ref, unwrapped_schema)) + task_identities.append((task_id, base, identity)) - local_from_python: dict[str, Any] = {} - if ref._task_image is not None: - local_from_python["image"] = ref._task_image - elif ref._task_image_id is not None: - resolved_image = resolve_image_id(ref._task_image_id, image_overrides) - if resolved_image is None: - raise CompileError( - f"@task image_id={ref._task_image_id!r} on function " - f"{ref._task_function_name!r} did not resolve to an image. " - f"Pass --image {ref._task_image_id}=IMAGE to `tangle sdk pipelines compile`, " - f"or register a default with register_image_id({ref._task_image_id!r}, IMAGE)." - ) - local_from_python["image"] = resolved_image - # Always pin the function name. Without it the hydrator defaults - # to the file stem and extracts the wrong symbol. - assert ref._task_function_name is not None - local_from_python["function"] = ref._task_function_name - if ref._task_mode is not None: - local_from_python["mode"] = ref._task_mode - if ref._task_resolve_root is not None: - resolve_root = ref._task_resolve_root - if not resolve_root.exists(): - raise CompileError( - f"@task resolve_root is unreachable: {resolve_root}. " - "Point resolve_root at an existing directory or drop it." - ) - local_from_python["resolve_root"] = _relpath_posix(resolve_root, components_yaml_dir) - if ref._task_dependencies_from is not None: - deps = ref._task_dependencies_from - if not deps.exists(): + entries: dict[str, Any] = {} + fragment_by_identity: dict[tuple[str, str], str] = {} + identity_labels: dict[str, str] = {} + for base, by_identity in variants.items(): + if len(by_identity) == 1: + identity, (emitted, _identity_payload) = next(iter(by_identity.items())) + named = [(legacy_fragments[(base, identity)], identity, emitted)] + else: + # EVERY colliding variant is suffixed — the trace-order "first" one + # does not get to keep the bare name — and they are emitted in + # sorted order so the sidecar text is call-order independent. + named = sorted( + (f"{base}--{identity}", identity, emitted) + for identity, (emitted, _identity_payload) in by_identity.items() + ) + for fragment, identity, emitted in named: + if fragment in entries: # defensive — digests are content-addressed raise CompileError( - f"@task dependencies_from file is unreachable: {deps}. " - "Point dependencies_from at an existing file or drop it." + f"internal error: sidecar fragment {fragment!r} was generated twice; " + f"colliding @task identities: {identity_labels[fragment]!r} and " + f"{_task_identity_label(by_identity[identity][1])!r}." ) - local_from_python["dependencies_from"] = _relpath_posix(deps, components_yaml_dir) - if unwrapped_schema: - local_from_python["unwrapped_inputs"] = unwrapped_schema - local_from_python["file"] = _relpath_posix(source, components_yaml_dir) - - seen_fragments[fragment] = source - # The component name is NOT emitted here. A top-level ``name`` on a - # resolve entry means "resolve a published component by this name" to - # the hydrator (PipelineHydrator._resolve_primary), which would let a - # same-named library component silently win over this local @task. The - # component's name comes from its source docstring (``Metadata: Name:``) - # at hydrate time, read by regenerate_yaml. - entry: dict[str, Any] = {"local_from_python": local_from_python} - entries[fragment] = entry - return entries + fragment_by_identity[(base, identity)] = fragment + identity_labels[fragment] = _task_identity_label(by_identity[identity][1]) + entries[fragment] = {"local_from_python": emitted} + + fragment_by_task = { + task_id: fragment_by_identity[(base, identity)] for task_id, base, identity in task_identities + } + return TaskSidecarPlan(entries=entries, fragment_by_task=fragment_by_task) def _rewrite_task_componentref_urls( *, body_dict: dict[str, Any], - task_refs: list[tuple[str, CallableRef]], + fragment_by_task: Mapping[str, str], components_yaml_name: str, - unwrapped_input_keys: Mapping[str, dict[str, list[str]]] | None = None, ) -> None: """Rewrite each @task task's ``componentRef`` to a pure resolve URL. Args: body_dict: Emitted dehydrated pipeline body to mutate in place. - task_refs: Traced ``(task_id, CallableRef)`` records for ``@task`` - calls in the pipeline graph. + fragment_by_task: ``TaskSidecarPlan.fragment_by_task`` — the SAME map + used to emit the sidecar, so a task whose component differs from a + same-function sibling points at its own fragment. components_yaml_name: Filename of the generated components sidecar. - unwrapped_input_keys: Optional trace metadata for ``@task(unwrap=...)`` - calls. The same persisted schema used for sidecar emission is used - here to compute schema-hashed fragments consistently. Returns: None. ``body_dict`` is mutated so each task's componentRef becomes ``{"url": "resolve://./#"}``. + + Raises: + CompileError: when a traced ``@task`` has no matching graph task. """ tasks = body_dict.get("implementation", {}).get("graph", {}).get("tasks", {}) - for task_id, ref in task_refs: - unwrapped_schema = _unwrapped_schema_for_task(ref, task_id, unwrapped_input_keys) - fragment = _fragment_for_task(ref, unwrapped_schema) + for task_id, fragment in fragment_by_task.items(): url = f"resolve://./{components_yaml_name}#{fragment}" if task_id not in tasks: raise CompileError( diff --git a/pyproject.toml b/pyproject.toml index 9c05dcf..30c1dd5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.13" +version = "0.1.14" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 5bcb1a1..21f20bd 100644 --- a/tests/test_packaging.py +++ b/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.13" in metadata + assert "Version: 0.1.14" 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 diff --git a/tests/test_pipeline_compiler.py b/tests/test_pipeline_compiler.py index eacb559..1c78179 100644 --- a/tests/test_pipeline_compiler.py +++ b/tests/test_pipeline_compiler.py @@ -651,6 +651,461 @@ def test_compile_task_image_id_without_default_or_override_fails(tmp_path): IMAGE_IDS.update(original) +# --------------------------------------------------------------------------- +# @task sidecar dedup by generated-component identity. +# +# A shared helper (e.g. ``run_dbt``) re-decorated with different task-level +# options must NOT collapse into the first call site's component: dedup is by +# the generated ``local_from_python`` payload, not by function name. + + +def _write_shared_task_pipeline(project: Path, body: str, *, extra_files: dict[str, str] | None = None) -> Path: + """Write a pipeline that re-decorates one shared helper function. + + Args: + project: Project root; ``src/`` holds ``shared.py`` and ``pipeline.py``. + body: The pipeline module text after the shared-helper import. + extra_files: Optional ``{relative path: contents}`` written under + ``src/`` (e.g. per-variant dependency files). + + Returns: + The written ``pipeline.py`` path. + """ + src = project / "src" + src.mkdir(parents=True) + (src / "shared.py").write_text( + "def run_dbt(model: str) -> str:\n" + ' """Run a dbt model.\n\n' + " Metadata:\n" + " Name: Run Dbt\n" + ' """\n' + " return model\n", + encoding="utf-8", + ) + for rel, contents in (extra_files or {}).items(): + target = src / rel + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(contents, encoding="utf-8") + pipeline_path = src / "pipeline.py" + pipeline_path.write_text(body, encoding="utf-8") + return pipeline_path + + +def _sidecar_and_tasks(pipeline_path: Path, out: Path, **kwargs): + result = compile_pipeline(pipeline_path, out, **kwargs) + sidecar = yaml.safe_load(result.components_path.read_text()) + tasks = yaml.safe_load(out.read_text())["implementation"]["graph"]["tasks"] + fragments = { + task_id: task["componentRef"]["url"].split("#", 1)[1] for task_id, task in tasks.items() + } + return sidecar, fragments + + +def test_compile_task_same_function_same_image_dedupes_to_one_entry(tmp_path): + """Repeated identical calls to one @task still collapse to ONE entry + under the legacy readable fragment, and both tasks share its ref.""" + project = tmp_path / "project" + pipeline_path = _write_shared_task_pipeline( + project, + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from shared import run_dbt\n\n" + "dbt = task(image='registry.example/dbt:1')(run_dbt)\n\n" + "@pipeline('Same Image Pipeline')\n" + "def same_image_pipeline() -> Out[str]:\n" + " first = dbt.named('first')(model='a')\n" + " second = dbt.named('second')(model='b')\n" + " return second\n", + ) + + sidecar, fragments = _sidecar_and_tasks(pipeline_path, project / "compiled.yaml") + + assert list(sidecar) == ["run-dbt"] + assert fragments == {"first": "run-dbt", "second": "run-dbt"} + assert sidecar["run-dbt"]["local_from_python"]["image"] == "registry.example/dbt:1" + + +def test_compile_task_same_function_different_images_emit_distinct_fragments(tmp_path): + """The bug from the shared-``run_dbt`` report: two task-level images for + one function must emit TWO sidecar entries, each task pointing at its own.""" + project = tmp_path / "project" + pipeline_path = _write_shared_task_pipeline( + project, + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from shared import run_dbt\n\n" + "dbt_slim = task(image='registry.example/dbt-slim:1')(run_dbt)\n" + "dbt_fat = task(image='registry.example/dbt-fat:2')(run_dbt)\n\n" + "@pipeline('Two Image Pipeline')\n" + "def two_image_pipeline() -> Out[str]:\n" + " slim = dbt_slim.named('slim')(model='a')\n" + " fat = dbt_fat.named('fat')(model='b')\n" + " return fat\n", + ) + + out = project / "compiled.yaml" + sidecar, fragments = _sidecar_and_tasks(pipeline_path, out) + + assert len(sidecar) == 2 + assert set(sidecar) == set(fragments.values()) + assert fragments["slim"] != fragments["fat"] + assert all(name.startswith("run-dbt--") for name in sidecar) + assert sidecar[fragments["slim"]]["local_from_python"]["image"] == "registry.example/dbt-slim:1" + assert sidecar[fragments["fat"]]["local_from_python"]["image"] == "registry.example/dbt-fat:2" + # Both variants still describe the SAME function in the SAME file. + for entry in sidecar.values(): + assert entry["local_from_python"]["function"] == "run_dbt" + assert entry["local_from_python"]["file"] == "./src/shared.py" + # Fragment names must not leak image / registry / credential-bearing text. + for name in sidecar: + assert "dbt-slim" not in name and "dbt-fat" not in name + assert "registry.example" not in name + validate_dehydrated_data(yaml.safe_load(out.read_text())) + + +def test_compile_task_image_id_variants_resolving_to_same_image_dedupe(tmp_path): + """Identity is the RESOLVED component payload: an explicit image and an + ``image_id`` that resolves to the same ref generate one component.""" + project = tmp_path / "project" + pipeline_path = _write_shared_task_pipeline( + project, + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from shared import run_dbt\n\n" + "dbt_literal = task(image='registry.example/dbt@sha256:abc')(run_dbt)\n" + "dbt_by_id = task(image_id='dbt')(run_dbt)\n\n" + "@pipeline('Image Id Dedup Pipeline')\n" + "def image_id_dedup_pipeline() -> Out[str]:\n" + " literal = dbt_literal.named('literal')(model='a')\n" + " by_id = dbt_by_id.named('by_id')(model='b')\n" + " return by_id\n", + ) + + sidecar, fragments = _sidecar_and_tasks( + pipeline_path, + project / "compiled.yaml", + image_overrides={"dbt": "registry.example/dbt@sha256:abc"}, + ) + + assert list(sidecar) == ["run-dbt"] + assert fragments == {"literal": "run-dbt", "by_id": "run-dbt"} + + +def test_compile_task_same_function_different_dependencies_emit_distinct_fragments(tmp_path): + """``dependencies_from`` also changes the generated component, so it + must split the sidecar entry too.""" + project = tmp_path / "project" + pipeline_path = _write_shared_task_pipeline( + project, + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from shared import run_dbt\n\n" + "dbt_a = task(image='registry.example/dbt:1', dependencies_from='reqs-a.txt')(run_dbt)\n" + "dbt_b = task(image='registry.example/dbt:1', dependencies_from='reqs-b.txt')(run_dbt)\n\n" + "@pipeline('Two Deps Pipeline')\n" + "def two_deps_pipeline() -> Out[str]:\n" + " a = dbt_a.named('a')(model='a')\n" + " b = dbt_b.named('b')(model='b')\n" + " return b\n", + extra_files={"reqs-a.txt": "dbt-core==1.7.0\n", "reqs-b.txt": "dbt-core==1.8.0\n"}, + ) + + sidecar, fragments = _sidecar_and_tasks(pipeline_path, project / "compiled.yaml") + + assert len(sidecar) == 2 + assert fragments["a"] != fragments["b"] + assert sidecar[fragments["a"]]["local_from_python"]["dependencies_from"] == "./src/reqs-a.txt" + assert sidecar[fragments["b"]]["local_from_python"]["dependencies_from"] == "./src/reqs-b.txt" + + +def test_compile_task_same_function_different_mode_emits_distinct_fragments(tmp_path): + """``mode``/``resolve_root`` are generation-affecting too.""" + project = tmp_path / "project" + pipeline_path = _write_shared_task_pipeline( + project, + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from shared import run_dbt\n\n" + "dbt_inline = task(image='registry.example/dbt:1')(run_dbt)\n" + "dbt_bundle = task(image='registry.example/dbt:1', mode='bundle', resolve_root='.')(run_dbt)\n\n" + "@pipeline('Mode Pipeline')\n" + "def mode_pipeline() -> Out[str]:\n" + " inline = dbt_inline.named('inline')(model='a')\n" + " bundled = dbt_bundle.named('bundled')(model='b')\n" + " return bundled\n", + ) + + sidecar, fragments = _sidecar_and_tasks(pipeline_path, project / "compiled.yaml") + + assert len(sidecar) == 2 + assert fragments["inline"] != fragments["bundled"] + assert "mode" not in sidecar[fragments["inline"]]["local_from_python"] + assert sidecar[fragments["bundled"]]["local_from_python"]["mode"] == "bundle" + assert sidecar[fragments["bundled"]]["local_from_python"]["resolve_root"] == "./src" + + +def _two_image_variant_source(first: str, second: str) -> str: + return ( + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from shared import run_dbt\n\n" + "dbt_slim = task(image='registry.example/dbt-slim:1')(run_dbt)\n" + "dbt_fat = task(image='registry.example/dbt-fat:2')(run_dbt)\n\n" + "@pipeline('Order Pipeline')\n" + "def order_pipeline() -> Out[str]:\n" + f" {first}\n" + f" {second}\n" + " return fat\n" + ) + + +def test_compile_task_variant_fragments_are_call_order_independent(tmp_path): + """Fragment names are content-addressed: swapping the call order keeps + each task pointing at the SAME fragment name (no positional suffixes, no + Python-``hash``/dict-order dependence).""" + slim_call = "slim = dbt_slim.named('slim')(model='a')" + fat_call = "fat = dbt_fat.named('fat')(model='b')" + + forward = _write_shared_task_pipeline( + tmp_path / "forward", _two_image_variant_source(slim_call, fat_call) + ) + reverse = _write_shared_task_pipeline( + tmp_path / "reverse", _two_image_variant_source(fat_call, slim_call) + ) + + forward_sidecar, forward_fragments = _sidecar_and_tasks( + forward, tmp_path / "forward" / "compiled.yaml" + ) + reverse_sidecar, reverse_fragments = _sidecar_and_tasks( + reverse, tmp_path / "reverse" / "compiled.yaml" + ) + + assert forward_fragments == reverse_fragments + assert set(forward_sidecar) == set(reverse_sidecar) + + +def test_compile_task_single_variant_keeps_legacy_fragment_name(tmp_path): + """Fragment stability: a single-configuration @task keeps the readable + hyphenated function name, unchanged by this dedup work.""" + project = tmp_path / "project" + pipeline_path = _write_shared_task_pipeline( + project, + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from shared import run_dbt\n\n" + "dbt = task(image='registry.example/dbt:1')(run_dbt)\n\n" + "@pipeline('Single Variant Pipeline')\n" + "def single_variant_pipeline() -> Out[str]:\n" + " return dbt.named('only')(model='a')\n", + ) + + sidecar, fragments = _sidecar_and_tasks(pipeline_path, project / "compiled.yaml") + + assert list(sidecar) == ["run-dbt"] + assert fragments == {"only": "run-dbt"} + + +def test_compile_task_unwrapped_variants_split_by_image_too(tmp_path): + """Unwrap schema hashing and image variance compose: same function, same + key set, two images -> two fragments with the same unwrapped schema.""" + project = tmp_path / "project" + src = project / "src" + src.mkdir(parents=True) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, pipeline, task\n\n" + "@task(image='python:3.12')\n" + "def produce() -> str:\n" + " return 'ok'\n\n" + "def combine(run_data: dict[str, str]) -> str:\n" + " return ','.join(sorted(run_data))\n\n" + "combine_slim = task(image='python:3.12', unwrap='run_data')(combine)\n" + "combine_fat = task(image='python:3.13', unwrap='run_data')(combine)\n\n" + "@pipeline('Unwrap Image Pipeline')\n" + "def unwrap_image_pipeline() -> Out[str]:\n" + " value = produce.named('value')()\n" + " slim = combine_slim.named('slim')(run_data={'value': value})\n" + " fat = combine_fat.named('fat')(run_data={'value': value})\n" + " return fat\n", + encoding="utf-8", + ) + + sidecar, fragments = _sidecar_and_tasks(pipeline_path, project / "compiled.yaml") + + assert fragments["slim"] != fragments["fat"] + slim = sidecar[fragments["slim"]]["local_from_python"] + fat = sidecar[fragments["fat"]]["local_from_python"] + assert slim["image"] == "python:3.12" + assert fat["image"] == "python:3.13" + assert slim["unwrapped_inputs"] == fat["unwrapped_inputs"] + + +def test_compile_task_same_function_name_in_two_packages_emits_both(tmp_path): + """Bare function names are NOT identities: ``pkg_a.tasks.run`` and + ``pkg_b.tasks.run`` are different components, so both are emitted and each + task points at its own — this is no longer a compile error.""" + project = tmp_path / "project" + src = project / "src" + for package in ("package_a", "package_b"): + pkg = src / package + pkg.mkdir(parents=True) + (pkg / "__init__.py").write_text("", encoding="utf-8") + (pkg / "tasks.py").write_text( + "def run(model: str) -> str:\n" + ' """Run a model.\n\n' + " Metadata:\n" + f" Name: Run {package}\n" + ' """\n' + f" return model + '{package}'\n", + encoding="utf-8", + ) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from package_a import tasks as tasks_a\n" + "from package_b import tasks as tasks_b\n\n" + "run_a = task(image='registry.example/dbt:1')(tasks_a.run)\n" + "run_b = task(image='registry.example/dbt:1')(tasks_b.run)\n\n" + "@pipeline('Two Package Pipeline')\n" + "def two_package_pipeline() -> Out[str]:\n" + " a = run_a.named('a')(model='a')\n" + " b = run_b.named('b')(model='b')\n" + " return b\n", + encoding="utf-8", + ) + + out = project / "compiled.yaml" + sidecar, fragments = _sidecar_and_tasks(pipeline_path, out) + + assert len(sidecar) == 2 + assert set(sidecar) == set(fragments.values()) + assert fragments["a"] != fragments["b"] + # No arbitrary "first" variant keeps the unsuffixed readable name. + assert all(name.startswith("run--") for name in sidecar) + assert sidecar[fragments["a"]]["local_from_python"]["file"] == "./src/package_a/tasks.py" + assert sidecar[fragments["b"]]["local_from_python"]["file"] == "./src/package_b/tasks.py" + # Fragment names must not leak the source path / package / file names. + for name in sidecar: + assert "package_a" not in name and "package_b" not in name + assert "tasks" not in name and ".py" not in name and "/" not in name + validate_dehydrated_data(yaml.safe_load(out.read_text())) + + +def _write_two_package_project(root: Path, *, images: tuple[str, str]) -> Path: + """Write the ``package_a.tasks.run`` / ``package_b.tasks.run`` project.""" + src = root / "src" + for package in ("package_a", "package_b"): + pkg = src / package + pkg.mkdir(parents=True) + (pkg / "__init__.py").write_text("", encoding="utf-8") + (pkg / "tasks.py").write_text( + "def run(model: str) -> str:\n" + ' """Run a model.\n\n' + " Metadata:\n" + f" Name: Run {package}\n" + ' """\n' + " return model\n", + encoding="utf-8", + ) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from package_a import tasks as tasks_a\n" + "from package_b import tasks as tasks_b\n\n" + f"run_a = task(image={images[0]!r})(tasks_a.run)\n" + f"run_b = task(image={images[1]!r})(tasks_b.run)\n\n" + "@pipeline('Relocatable Pipeline')\n" + "def relocatable_pipeline() -> Out[str]:\n" + " a = run_a.named('a')(model='a')\n" + " b = run_b.named('b')(model='b')\n" + " return b\n", + encoding="utf-8", + ) + return pipeline_path + + +def test_compile_task_variant_fragments_survive_relocation_and_output_dir(tmp_path): + """Identity is anchored at the PROJECT, not the machine or the output + directory: the same project compiled from another absolute location, and + into a nested output directory, yields identical fragment names.""" + images = ("registry.example/dbt-slim:1", "registry.example/dbt-fat:2") + here = _write_two_package_project(tmp_path / "here", images=images) + moved = _write_two_package_project(tmp_path / "somewhere" / "else" / "deeper", images=images) + + _, here_fragments = _sidecar_and_tasks(here, tmp_path / "here" / "compiled.yaml") + _, moved_fragments = _sidecar_and_tasks( + moved, tmp_path / "somewhere" / "else" / "deeper" / "compiled.yaml" + ) + # Same project, compiled into a DIFFERENT output directory. + nested_out = tmp_path / "here" / "build" / "nested" / "compiled.yaml" + nested_sidecar, nested_fragments = _sidecar_and_tasks(here, nested_out) + + assert here_fragments == moved_fragments == nested_fragments + # The emitted paths still track the output dir even though names do not. + assert nested_sidecar[nested_fragments["a"]]["local_from_python"]["file"] == ( + "../../src/package_a/tasks.py" + ) + + +def test_compile_task_same_module_function_and_config_still_dedupes(tmp_path): + """Same module-qualified function + same config across call sites still + collapses to ONE entry under the readable legacy fragment.""" + project = tmp_path / "project" + src = project / "src" + pkg = src / "package_a" + pkg.mkdir(parents=True) + (pkg / "__init__.py").write_text("", encoding="utf-8") + (pkg / "tasks.py").write_text( + "def run(model: str) -> str:\n" + ' """Run a model.\n\n' + " Metadata:\n" + " Name: Run A\n" + ' """\n' + " return model\n", + encoding="utf-8", + ) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from package_a import tasks\n\n" + "run_once = task(image='registry.example/dbt:1')(tasks.run)\n" + "run_again = task(image='registry.example/dbt:1')(tasks.run)\n\n" + "@pipeline('Same Module Pipeline')\n" + "def same_module_pipeline() -> Out[str]:\n" + " a = run_once.named('a')(model='a')\n" + " b = run_again.named('b')(model='b')\n" + " return b\n", + encoding="utf-8", + ) + + sidecar, fragments = _sidecar_and_tasks(pipeline_path, project / "compiled.yaml") + + assert list(sidecar) == ["run"] + assert fragments == {"a": "run", "b": "run"} + + +def test_compile_task_cross_file_variants_are_call_order_independent(tmp_path): + """Two same-named functions in different packages keep the same fragment + names (and the same sorted sidecar order) when the calls are swapped.""" + images = ("registry.example/dbt:1", "registry.example/dbt:1") + forward = _write_two_package_project(tmp_path / "forward", images=images) + reverse_root = tmp_path / "reverse" + _write_two_package_project(reverse_root, images=images) + reverse = reverse_root / "src" / "pipeline.py" + reverse.write_text( + reverse.read_text(encoding="utf-8") + .replace(" a = run_a.named('a')(model='a')\n b = run_b.named('b')(model='b')\n", "") + .replace( + " return b\n", + " b = run_b.named('b')(model='b')\n a = run_a.named('a')(model='a')\n return b\n", + ), + encoding="utf-8", + ) + + forward_sidecar, forward_fragments = _sidecar_and_tasks( + forward, tmp_path / "forward" / "compiled.yaml" + ) + reverse_sidecar, reverse_fragments = _sidecar_and_tasks( + reverse, reverse_root / "compiled.yaml" + ) + + assert forward_fragments == reverse_fragments + assert list(forward_sidecar) == list(reverse_sidecar) + + # --------------------------------------------------------------------------- # Runnable argument-value emission (raw string constant / graphInput / # taskOutput). Dispatch is on the VALUE's type, never the argument KEY. diff --git a/uv.lock b/uv.lock index cf7bae0..8c0b313 100644 --- a/uv.lock +++ b/uv.lock @@ -2083,7 +2083,7 @@ requires-dist = [{ name = "pydantic", specifier = ">=2.0" }] [[package]] name = "tangle-cli" -version = "0.1.13" +version = "0.1.14" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" }, From 38821a20f7a85487780768b513831375f751638e Mon Sep 17 00:00:00 2001 From: Volv G Date: Wed, 16 Sep 2026 10:41:13 -0700 Subject: [PATCH 2/2] Canonicalize implicit inline mode before sidecar dedup MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review found one identity-equivalence gap in the dedup key: `@task()` omits `mode` from the emitted block while `@task(mode="inline")` writes it, so the two hashed apart and produced two sidecar entries for what is provably ONE component. The hydrator reads `gen_config.get("mode", "inline")` and `CallableRef` generates with `self._task_mode or "inline"`, and an end-to-end check confirmed both spellings regenerate the same component digest. The practical cost was a spurious second entry plus loss of the readable unsuffixed fragment for a pipeline that only has one component; runtime behaviour was already correct. Normalize `mode` to `ref._task_mode or "inline"` inside the identity payload only. The EMITTED block is untouched, so a pipeline that omits the default still writes no `mode:` key. When two spellings of one component do meet, the emitted representative is now chosen canonically instead of by trace order: rank by key count, then by canonical JSON. Key count first means the leanest spelling wins, so a pipeline that already omits the redundant default keeps its existing sidecar bytes when a sibling call site spells it out. Audited the other optional fields for the same implicit-default pattern and deliberately left them alone, with the reasoning recorded beside the normalization: * `dependencies_from` — omission means "auto-discover next to the source AT HYDRATE TIME". Compile-time discovery could disagree with the hydrate-time layout, so an explicit path is not provably the same component. * `resolve_root` — not inert in inline mode: `component_from_func` emits a `tangle_cli_generation_resolve_root` annotation whenever it is set, so it changes the generated component regardless of mode. * `image` — omission means "whatever the generator defaults to", not a value this layer can canonicalize. Three regressions: omitted vs explicit inline dedup to one legacy `run` fragment with both refs on it; both call orders emit byte-identical sidecars with the lean representative; and `mode="bundle"` still splits, so normalization does not blur a real difference. The example's quoted digests move with the identity change and are refreshed; its "observed output" listing now matches the sorted emission order. Co-authored-by: Claude Opus 5 Assisted-By: devx/43e17814-42d0-4ca6-91b1-567962ce0219 --- .../dedup_image_variants/pipeline.py | 14 +-- .../src/tangle_cli/pipeline_compiler.py | 88 +++++++++++++++---- tests/test_pipeline_compiler.py | 84 ++++++++++++++++++ 3 files changed, 165 insertions(+), 21 deletions(-) diff --git a/examples/python_pipeline/dedup_image_variants/pipeline.py b/examples/python_pipeline/dedup_image_variants/pipeline.py index 2e5306f..00074cd 100644 --- a/examples/python_pipeline/dedup_image_variants/pipeline.py +++ b/examples/python_pipeline/dedup_image_variants/pipeline.py @@ -43,14 +43,16 @@ the output directory and not an absolute machine path). Observed output on this revision (the digests are derived only from -project-relative values, so they reproduce on any checkout):: +project-relative values, so they reproduce on any checkout). The sidecar lists +colliding variants in sorted fragment order, which is why the fat entry appears +first:: - run-dbt--b196ad73a4 image: python:3.12-slim - run-dbt--6024a73044 image: python:3.12 + run-dbt--573d8d33bd image: python:3.12 + run-dbt--e61a70231d image: python:3.12-slim - daily_orders -> resolve://./pipeline.components.yaml#run-dbt--b196ad73a4 - hourly_sessions -> resolve://./pipeline.components.yaml#run-dbt--b196ad73a4 - backfill_orders -> resolve://./pipeline.components.yaml#run-dbt--6024a73044 + daily_orders -> resolve://./pipeline.components.yaml#run-dbt--e61a70231d + hourly_sessions -> resolve://./pipeline.components.yaml#run-dbt--e61a70231d + backfill_orders -> resolve://./pipeline.components.yaml#run-dbt--573d8d33bd Note on ``file:``: sidecar paths are relative to the OUTPUT directory, so compiling into ``/tmp`` (outside the source tree) writes a long ``../../..`` diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py b/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py index 27f45e0..c850d4d 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py @@ -1163,19 +1163,30 @@ def _unwrapped_schema_for_task( raise CompileError(str(exc)) from exc +def _canonical_json(payload: Any) -> str: + """Canonical JSON encoding used for hashing and deterministic ordering. + + Args: + payload: Any JSON-serialisable structure. Mapping keys are sorted so + the encoding never depends on Python dict insertion order. + + Returns: + A compact, key-sorted JSON string. + """ + return json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) + + def _stable_payload_hash(payload: Any) -> str: """Short deterministic SHA-256 prefix for a JSON-serialisable payload. Args: - payload: Any JSON-serialisable structure. Mapping keys are sorted so - the digest never depends on Python dict insertion order, and the - digest is stable across processes (unlike :func:`hash`). + payload: Any JSON-serialisable structure. The digest is stable across + processes (unlike :func:`hash`) and independent of dict ordering. Returns: A 10-character lowercase hex digest prefix. """ - encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str) - return hashlib.sha256(encoded.encode("utf-8")).hexdigest()[:10] + return hashlib.sha256(_canonical_json(payload).encode("utf-8")).hexdigest()[:10] def _unwrapped_schema_hash(schema: Mapping[str, Any]) -> str: @@ -1710,6 +1721,24 @@ def _logical_module_id(source: Path, *, identity_root: Path) -> str: return f"{_relpath_posix(directory, identity_root)}:{dotted}" +def _emitted_form_rank(payload: Mapping[str, Any]) -> tuple[int, str]: + """Sort key selecting the canonical spelling of one generated component. + + Several call sites can describe the SAME component with different emitted + text — writing ``mode="inline"`` explicitly versus leaving the implicit + default out. Ranking by key count first prefers the leanest spelling, so a + pipeline that already omits a redundant default keeps its existing sidecar + bytes when a sibling call site spells that default out. + + Args: + payload: An emitted ``local_from_python`` block. + + Returns: + ``(key count, canonical JSON)`` — total and independent of call order. + """ + return (len(payload), _canonical_json(payload)) + + def _task_component_identity( ref: CallableRef, *, @@ -1749,6 +1778,31 @@ def _task_component_identity( # same-named functions nested in different scopes within ONE module; # it equals the function name for the module-level authoring surface. qualname = getattr(ref, "__qualname__", None) or ref._task_function_name + generation = _local_from_python_payload( + ref, + source=source, + components_yaml_dir=identity_root, + image_overrides=image_overrides, + unwrapped_schema=unwrapped_schema, + ) + # Normalise fields whose OMISSION is defined to mean a specific value, so + # two spellings of one component do not hash apart. Only ``mode`` has such + # an implicit default in the emitted contract: the hydrator reads + # ``gen_config.get("mode", "inline")`` and ``CallableRef`` generates with + # ``self._task_mode or "inline"``, so ``@task()`` and ``@task(mode="inline")`` + # regenerate byte-identical components. + # + # Audited and deliberately NOT normalised: + # * ``dependencies_from`` — omission means "auto-discover next to the + # source AT HYDRATE TIME". Compile-time discovery could disagree with the + # hydrate-time layout, so an explicit path is not provably the same + # component as an omission. + # * ``resolve_root`` — not inert in inline mode: ``component_from_func`` + # emits a ``tangle_cli_generation_resolve_root`` annotation whenever it is + # set, so it changes the generated component regardless of mode. + # * ``image`` — omission means "whatever the generator defaults to", which + # is not a value this layer can canonicalise. + generation["mode"] = generation.get("mode") or "inline" return { "module": _logical_module_id(source, identity_root=identity_root), "qualname": qualname, @@ -1756,13 +1810,7 @@ def _task_component_identity( # Generation config, anchored at the stable project root. This mirrors # the emitted block field-for-field so a new generation-affecting # option cannot be added to the sidecar without also splitting dedup. - "generation": _local_from_python_payload( - ref, - source=source, - components_yaml_dir=identity_root, - image_overrides=image_overrides, - unwrapped_schema=unwrapped_schema, - ), + "generation": generation, } @@ -1793,8 +1841,9 @@ def _plan_task_sidecar( """Plan the ``.components.yaml`` entries and per-task fragments. Dedup is by GENERATED COMPONENT IDENTITY — module-qualified function - identity plus every generation-affecting option — never by bare function - name. Two call sites collapse into one entry only when the whole + identity plus every generation-affecting option, with implicit defaults + canonicalised (``@task()`` and ``@task(mode="inline")`` are ONE component) + — never by bare function name. Two call sites collapse into one entry only when the whole :func:`_task_component_identity` payload matches: logical module namespace, ``__qualname__``, function name, image (explicit or resolved ``image_id``), mode, resolve_root, dependencies_from, the persisted unwrap schema, and the @@ -1895,7 +1944,16 @@ def _plan_task_sidecar( unwrapped_schema=unwrapped_schema, ) identity = _stable_payload_hash(identity_payload) - variants.setdefault(base, {}).setdefault(identity, (emitted, identity_payload)) + by_identity = variants.setdefault(base, {}) + previous = by_identity.get(identity) + if previous is None: + by_identity[identity] = (emitted, identity_payload) + elif _emitted_form_rank(emitted) < _emitted_form_rank(previous[0]): + # Same component, different SPELLING (e.g. one call site omits + # ``mode`` while another writes ``mode="inline"``). Pick the + # canonical representative rather than whichever was traced first, + # so the sidecar text does not depend on call order. + by_identity[identity] = (emitted, previous[1]) legacy_fragments.setdefault((base, identity), _fragment_for_task(ref, unwrapped_schema)) task_identities.append((task_id, base, identity)) diff --git a/tests/test_pipeline_compiler.py b/tests/test_pipeline_compiler.py index 1c78179..618dc89 100644 --- a/tests/test_pipeline_compiler.py +++ b/tests/test_pipeline_compiler.py @@ -1106,6 +1106,90 @@ def test_compile_task_cross_file_variants_are_call_order_independent(tmp_path): assert list(forward_sidecar) == list(reverse_sidecar) +def _write_mode_spelling_pipeline(project: Path, *, first: str, second: str) -> Path: + """One shared function decorated twice, differing only in how ``mode`` is + spelled (omitted vs explicit ``"inline"``).""" + src = project / "src" + pkg = src / "package_a" + pkg.mkdir(parents=True) + (pkg / "__init__.py").write_text("", encoding="utf-8") + (pkg / "tasks.py").write_text( + "def run(model: str) -> str:\n" + ' """Run a model.\n\n' + " Metadata:\n" + " Name: Run A\n" + ' """\n' + " return model\n", + encoding="utf-8", + ) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, pipeline, task\n" + "from package_a import tasks\n\n" + f"run_first = task(image='registry.example/dbt:1'{first})(tasks.run)\n" + f"run_second = task(image='registry.example/dbt:1'{second})(tasks.run)\n\n" + "@pipeline('Mode Spelling Pipeline')\n" + "def mode_spelling_pipeline() -> Out[str]:\n" + " a = run_first.named('a')(model='a')\n" + " b = run_second.named('b')(model='b')\n" + " return b\n", + encoding="utf-8", + ) + return pipeline_path + + +def test_compile_task_omitted_mode_and_explicit_inline_are_one_component(tmp_path): + """``@task()`` and ``@task(mode="inline")`` generate the SAME component: + the hydrator reads ``gen_config.get("mode", "inline")``. They must dedup to + one entry keeping the readable legacy fragment, not split into two hashed + variants.""" + project = tmp_path / "project" + pipeline_path = _write_mode_spelling_pipeline( + project, first="", second=", mode='inline'" + ) + + sidecar, fragments = _sidecar_and_tasks(pipeline_path, project / "compiled.yaml") + + assert list(sidecar) == ["run"] + assert fragments == {"a": "run", "b": "run"} + + +def test_compile_task_mode_spelling_representative_is_call_order_independent(tmp_path): + """When two spellings of one component meet, the EMITTED form is chosen + canonically, so swapping the call order yields byte-identical sidecars.""" + forward = _write_mode_spelling_pipeline( + tmp_path / "forward", first="", second=", mode='inline'" + ) + reverse = _write_mode_spelling_pipeline( + tmp_path / "reverse", first=", mode='inline'", second="" + ) + + forward_sidecar, forward_fragments = _sidecar_and_tasks( + forward, tmp_path / "forward" / "compiled.yaml" + ) + reverse_sidecar, reverse_fragments = _sidecar_and_tasks( + reverse, tmp_path / "reverse" / "compiled.yaml" + ) + + assert forward_fragments == reverse_fragments == {"a": "run", "b": "run"} + assert forward_sidecar == reverse_sidecar + # The canonical representative omits the redundant implicit default. + assert "mode" not in forward_sidecar["run"]["local_from_python"] + + +def test_compile_task_explicit_bundle_mode_still_splits_from_inline(tmp_path): + """Normalising the implicit default must not blur a REAL mode difference.""" + project = tmp_path / "project" + pipeline_path = _write_mode_spelling_pipeline( + project, first="", second=", mode='bundle', resolve_root='.'" + ) + + sidecar, fragments = _sidecar_and_tasks(pipeline_path, project / "compiled.yaml") + + assert len(sidecar) == 2 + assert fragments["a"] != fragments["b"] + + # --------------------------------------------------------------------------- # Runnable argument-value emission (raw string constant / graphInput / # taskOutput). Dispatch is on the VALUE's type, never the argument KEY.