artifacts: parent lineage, duck-typed declaration protocol, content-hash versions

- Metadata gains parents (ordered ArtifactParent | bare-version entries,
  serialized to the new repeated parent_artifacts on both ProducedArtifact
  and ArtifactSpec) and version_from_content (opt-in: an empty version
  resolves to the literal's content hash at conversion/create time, making
  republish of identical content idempotent).
- Output conversion duck-types get_flyte_metadata() on top-level outputs so
  offloaded-asset types outside flyte.io (e.g. plugin volumes) can declare
  themselves; a new current_output_declares_artifact contextvar tells the
  value's transformer when a declaration is actually being emitted.
- ensure_artifactable and Artifact.create accept protocol-carrying values;
  create() gains parents= and seeds defaults from duck metadata like it does
  from the wrapper.
- flyteidl2 pinned to 2.0.48 (repeated parent_artifacts), incl. rs_controller.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01N415hnSCNmZ2e7WrwEPJo7
Signed-off-by: Haytham Abuelfutuh <haytham@union.ai>

Author: EngHabu

pyproject.toml

@@ -26,7 +26,7 @@ dependencies = [
     "async-lru>=2.0.5",
     "mashumaro>=3.15",
     "aiolimiter>=1.2.1",
-    "flyteidl2==2.0.45",
+    "flyteidl2==2.0.48",
     "packaging",
     "sentry-sdk>=2.0",
     "pyOpenSSL>=24.0.0",

rs_controller/Cargo.lock

@@ -56,7 +56,7 @@ async-trait = "0.1"
 thiserror = "1.0"
 # Uncomment this if you need to use local flyteidl2
 #flyteidl2 = { path = "/Users/ytong/go/src/github.com/flyteorg/flyte/gen/rust" }
-flyteidl2 = "=2.0.45"
+flyteidl2 = "=2.0.48"
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"

rs_controller/Cargo.toml

@@ -9,7 +9,7 @@ description = "Rust controller for Union"
 requires-python = ">=3.10"
 classifiers = ["Programming Language :: Python", "Programming Language :: Rust"]
 dependencies = [
-    "flyteidl2==2.0.45",
+    "flyteidl2==2.0.48",
 ]
 [tool.maturin]
 module-name = "flyte_controller_base"

rs_controller/pyproject.toml

@@ -42,6 +42,26 @@ def current_output_name() -> Optional[str]:
     return _output_name_var.get()
 
 
+# True only while converting a top-level output for which
+# ``convert_from_native_to_outputs`` captured artifact metadata and will emit a
+# ``ProducedArtifact`` declaration. Lets a TypeTransformer know the value it is
+# serializing is being registered as an artifact — e.g. to stamp the version
+# it will be registered under into the serialized value itself. Deliberately
+# NOT set for values nested inside containers/models on the same output: those
+# are never declared.
+_output_declares_artifact_var: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "flyte_current_output_declares_artifact", default=False
+)
+
+
+def current_output_declares_artifact() -> bool:
+    """Whether the output being converted right now will be declared as a
+    produced artifact on the Outputs envelope. See
+    `_output_declares_artifact_var`.
+    """
+    return _output_declares_artifact_var.get()
+
+
 @dataclass(frozen=True)
 class Inputs:
     proto_inputs: common_pb2.Inputs
@@ -429,7 +449,7 @@ async def convert_from_native_to_outputs(o: Any, interface: NativeInterface, tas
             f"Received {len(o)} outputs but return annotation has {len(interface.outputs)} outputs specified. "
         )
     from flyte.artifacts._metadata import to_produced_artifact
-    from flyte.artifacts._wrapper import ArtifactWrapper, raise_if_nested_wrapper
+    from flyte.artifacts._wrapper import raise_if_nested_wrapper
 
     named = []
     produced: list[common_pb2.ProducedArtifact] = []
@@ -438,22 +458,40 @@ async def convert_from_native_to_outputs(o: Any, interface: NativeInterface, tas
         # the wrapper and discards it, then emit a ProducedArtifact declaration on the Outputs
         # envelope so the backend can register the artifact. The declaration carries the
         # declared output type (this SDK is authoritative for it).
+        #
+        # The check is a protocol, not the wrapper class: any top-level output
+        # exposing ``get_flyte_metadata() -> Metadata | None`` participates, so
+        # offloaded-asset types outside flyte.io (e.g. plugin-provided volumes)
+        # can declare themselves without being wrappable.
         raise_if_nested_wrapper(v)
-        produced_md = v.get_flyte_metadata() if isinstance(v, ArtifactWrapper) else None
+        produced_md = None
+        md_getter = getattr(v, "get_flyte_metadata", None)
+        if callable(md_getter):
+            produced_md = md_getter()
 
         # Expose the output slot name to transformers for the duration of this
         # single conversion (see ``current_output_name``), then always clear it.
         tok = _output_name_var.set(output_name)
+        decl_tok = _output_declares_artifact_var.set(produced_md is not None)
         try:
             literal_type = TypeEngine.to_literal_type(python_type)
             lit = await TypeEngine.to_literal(v, python_type, literal_type)
             if produced_md is not None:
-                produced.append(to_produced_artifact(produced_md, output=output_name, literal_type=literal_type))
+                pa = to_produced_artifact(produced_md, output=output_name, literal_type=literal_type)
+                # Content-addressed default version: when the metadata opts in and
+                # the transformer stamped a content hash on the literal, use it
+                # instead of leaving the version to the backend's
+                # run-action-attempt default. Deterministic versions make
+                # re-declaring the same content idempotent (AlreadyExists).
+                if not pa.version and produced_md.version_from_content and lit.hash:
+                    pa.version = lit.hash
+                produced.append(pa)
             named.append(common_pb2.NamedLiteral(name=output_name, value=lit))
         except TypeTransformerFailedError as e:
             raise flyte.errors.RuntimeDataValidationError(output_name, e, task_name)
         finally:
             _output_name_var.reset(tok)
+            _output_declares_artifact_var.reset(decl_tok)
 
     return Outputs(proto_outputs=common_pb2.Outputs(literals=named, produced_artifacts=produced))
 

src/flyte/_internal/runtime/convert.py

@@ -52,7 +52,7 @@ async def my_task() -> File:
 """
 
 from ._card import Card, CardFormat, CardType
-from ._metadata import KIND_KEY, Kind, Metadata
+from ._metadata import KIND_KEY, ArtifactParent, Kind, Metadata
 from ._wrapper import Artifact, new
 
-__all__ = ["KIND_KEY", "Artifact", "Card", "CardFormat", "CardType", "Kind", "Metadata", "new"]
+__all__ = ["KIND_KEY", "Artifact", "ArtifactParent", "Card", "CardFormat", "CardType", "Kind", "Metadata", "new"]

src/flyte/artifacts/__init__.py

@@ -27,6 +27,29 @@
 Kind = Literal["model", "data", "generic"]
 
 
+@dataclass(frozen=True, kw_only=True)
+class ArtifactParent:
+    """One parent edge for an artifact version's lineage (`Metadata.parents`).
+
+    Every field except `version` defaults to "inherit from the child being
+    published": an empty `name` means a same-name parent (an earlier version of
+    this artifact), and empty scope fields inherit the child's
+    org/project/domain. A bare version string is accepted anywhere an
+    `ArtifactParent` is — it is shorthand for `ArtifactParent(version=...)`.
+
+    Parents are stored as given and never resolved: a parent that hasn't been
+    (or never will be) published is legal, like a git remote missing commits
+    that were never pushed. Only a direct self-reference is rejected by the
+    service.
+    """
+
+    version: str
+    name: Optional[str] = None
+    project: Optional[str] = None
+    domain: Optional[str] = None
+    org: Optional[str] = None
+
+
 @dataclass(frozen=True, kw_only=True)
 class Metadata:
     """Structured metadata for Flyte artifacts."""
@@ -41,6 +64,18 @@ class Metadata:
     #: serialization time; an explicit `attrs["kind"]` wins, so a caller who
     #: sets the key by hand is never silently overridden.
     kind: Optional[Kind] = None
+    #: Lineage: the artifact versions this version derives from, ordered with
+    #: the primary parent first (git-style merge lineage; up to 32). Each entry
+    #: is an `ArtifactParent` or a bare version string (shorthand for a
+    #: same-name parent).
+    parents: Optional[Tuple[typing.Union[str, ArtifactParent], ...]] = None
+    #: When no explicit `version` is given, publish under the content hash the
+    #: type transformer stamped on the literal (`Literal.hash`) instead of the
+    #: backend's run-action-attempt default. Opt-in: content-addressed versions
+    #: make re-publishing identical content idempotent, but they also mean
+    #: unchanged content produces NO new version — only types whose literals
+    #: carry a meaningful content hash should set this.
+    version_from_content: bool = False
 
     @classmethod
     def create_model_metadata(
@@ -101,6 +136,31 @@ def resolve_attrs(md: Metadata) -> dict[str, str]:
     return attrs
 
 
+def parents_to_pb2(
+    parents: Optional[typing.Sequence[typing.Union[str, ArtifactParent]]],
+) -> list[artifact_id_pb2.ArtifactVersionId]:
+    """
+    Serialize parent edges for the wire (`ArtifactSpec.parent_artifacts` /
+    `ProducedArtifact.parent_artifacts`). A bare string becomes a keyless
+    entry — the service inherits the child's name and scope for empty key
+    fields — and an `ArtifactParent` carries a key only when it overrides
+    something, keeping the common same-name case terse on the wire.
+    """
+    out: list[artifact_id_pb2.ArtifactVersionId] = []
+    for entry in parents or ():
+        parent = ArtifactParent(version=entry) if isinstance(entry, str) else entry
+        key = None
+        if parent.name or parent.project or parent.domain or parent.org:
+            key = artifact_id_pb2.ArtifactKey(
+                org=parent.org or "",
+                project=parent.project or "",
+                domain=parent.domain or "",
+                name=parent.name or "",
+            )
+        out.append(artifact_id_pb2.ArtifactVersionId(key=key, version=parent.version))
+    return out
+
+
 def to_produced_artifact(
     md: Metadata,
     *,
@@ -127,4 +187,5 @@ def to_produced_artifact(
         version=md.version or "",
         info=info,
         type=literal_type,
+        parent_artifacts=parents_to_pb2(md.parents) or None,
     )

src/flyte/artifacts/_metadata.py

@@ -108,17 +108,23 @@ def __contains__(self, item):
 def ensure_artifactable(obj: Any) -> None:
     """
     Validate that a value is allowed to be an artifact. Artifacts are offloaded
-    assets only — flyte.io File, Dir, or DataFrame. Everything else (primitives,
-    bytes, dataclasses, pydantic models, arbitrary objects) raises TypeError.
+    assets only — flyte.io File, Dir, or DataFrame, plus any type that opts in
+    by exposing the artifact-metadata protocol (`get_flyte_metadata() ->
+    Metadata | None`, e.g. plugin-provided volumes). Everything else
+    (primitives, bytes, dataclasses, pydantic models, arbitrary objects)
+    raises TypeError.
     """
     from flyte.io import DataFrame, Dir, File
 
-    if not isinstance(obj, (File, Dir, DataFrame)):
-        raise TypeError(
-            f"values of type {type(obj).__name__!r} cannot be artifacts; artifacts are offloaded "
-            "assets: flyte.io.File, flyte.io.Dir, or flyte.io.DataFrame "
-            "(wrap a raw dataframe with DataFrame.from_df())"
-        )
+    if isinstance(obj, (File, Dir, DataFrame)):
+        return
+    if not isinstance(obj, type) and callable(getattr(obj, "get_flyte_metadata", None)):
+        return
+    raise TypeError(
+        f"values of type {type(obj).__name__!r} cannot be artifacts; artifacts are offloaded "
+        "assets: flyte.io.File, flyte.io.Dir, or flyte.io.DataFrame "
+        "(wrap a raw dataframe with DataFrame.from_df())"
+    )
 
 
 def raise_if_nested_wrapper(obj: Any, _depth: int = 0) -> None:

src/flyte/artifacts/_wrapper.py

@@ -12,7 +12,7 @@
 
 from flyte._initialize import ensure_client, get_client, get_init_config
 from flyte.artifacts._card import Card as CoreCard
-from flyte.artifacts._metadata import KIND_KEY, Kind, Metadata, resolve_attrs
+from flyte.artifacts._metadata import KIND_KEY, ArtifactParent, Kind, Metadata, parents_to_pb2, resolve_attrs
 from flyte.artifacts._wrapper import ArtifactWrapper, ensure_artifactable
 from flyte.remote._common import ToJSONMixin
 from flyte.syncify import syncify
@@ -221,6 +221,7 @@ async def create(
         project: str | None = None,
         domain: str | None = None,
         external_ref: str | None = None,
+        parents: Sequence[str | ArtifactParent] | None = None,
     ) -> Artifact:
         """
         Publish an artifact from the local machine.
@@ -248,6 +249,11 @@ async def create(
                 model id, dataset id, ...) recorded as the artifact's source. When omitted
                 and called from inside a running task, the producing task action is
                 recorded automatically instead.
+            parents: Lineage — the artifact versions this one derives from, ordered
+                with the primary parent first. Each entry is an
+                `flyte.artifacts.ArtifactParent` or a bare version string (a same-name
+                parent). Stored as given, never resolved; a not-yet-published parent
+                is legal.
 
         Returns:
             The published Artifact.
@@ -258,16 +264,25 @@ async def create(
         cfg = get_init_config()
 
         obj = value
+        md: Metadata | None = None
         if type(value) is ArtifactWrapper:
-            md: Metadata = value.get_flyte_metadata()
+            md = value.get_flyte_metadata()
             obj = value._obj
+        elif callable(getattr(value, "get_flyte_metadata", None)):
+            # The artifact-metadata protocol (see ensure_artifactable): a value
+            # that carries its own metadata seeds the same defaults a wrapper
+            # does. May legitimately return None (the type participates in the
+            # protocol but this instance declares nothing).
+            md = value.get_flyte_metadata()
+        if md is not None:
             name = name or md.name
             version = version or md.version
             description = description if description is not None else md.description
             # resolve_attrs folds the wrapper's kind= into attrs; reading md.attrs
             # directly would drop it for values wrapped by flyte.artifacts.new().
             attrs = attrs if attrs is not None else resolve_attrs(md)
             card = card if card is not None else md.card
+            parents = parents if parents is not None else md.parents
         if kind is not None:
             # Same precedence as Metadata: an explicit reserved key already in attrs
             # is deliberate and wins.
@@ -282,6 +297,11 @@ async def create(
         pt = python_type or type(obj)
         lt = TypeEngine.to_literal_type(pt)
         lit = await TypeEngine.to_literal(obj, pt, lt)
+        # Content-addressed default version (same rule as the declarative path in
+        # convert.py): metadata opted in and the transformer stamped a content
+        # hash — use it, so republishing identical content is idempotent.
+        if not version and md is not None and md.version_from_content and lit.hash:
+            version = lit.hash
 
         if external_ref is not None:
             source: artifact_pb2.ArtifactSource | None = artifact_pb2.ArtifactSource(external_ref=external_ref)
@@ -307,6 +327,7 @@ async def create(
                     card=_card_to_pb2(card),
                 ),
                 source=source,
+                parent_artifacts=parents_to_pb2(parents) or None,
             ),
         )
         resp = await get_client().artifact_service.create_artifact(request)

src/flyte/remote/_artifact.py

@@ -493,3 +493,70 @@ async def test_limit_stops_early(self):
             got = [g async for g in Artifact.list_names.aio(limit=2)]
 
         assert len(got) == 2
+
+
+class TestCreateParents:
+    """`create(parents=...)` and metadata-protocol seeding land on
+    `ArtifactSpec.parent_artifacts` with the same shapes as declarations."""
+
+    @staticmethod
+    def _client():
+        client = MagicMock()
+
+        async def _create(req):
+            return artifact_service_pb2.CreateArtifactResponse(artifact=await _stored_artifact("v"))
+
+        client.artifact_service.create_artifact = AsyncMock(side_effect=_create)
+        return client
+
+    @pytest.mark.asyncio
+    async def test_explicit_parents(self):
+        client = self._client()
+        p1, p2, p3 = _patched(client)
+        with p1, p2, p3:
+            await Artifact.create.aio(
+                _payload(),
+                name="child",
+                version="v2",
+                parents=["v1", artifacts.ArtifactParent(version="v0", name="base")],
+            )
+        req = client.artifact_service.create_artifact.await_args[0][0]
+        assert [p.version for p in req.spec.parent_artifacts] == ["v1", "v0"]
+        assert not req.spec.parent_artifacts[0].HasField("key")
+        assert req.spec.parent_artifacts[1].key.name == "base"
+
+    @pytest.mark.asyncio
+    async def test_no_parents_leaves_field_empty(self):
+        client = self._client()
+        p1, p2, p3 = _patched(client)
+        with p1, p2, p3:
+            await Artifact.create.aio(_payload(), name="solo", version="v1")
+        req = client.artifact_service.create_artifact.await_args[0][0]
+        assert len(req.spec.parent_artifacts) == 0
+
+    @pytest.mark.asyncio
+    async def test_wrapper_metadata_seeds_parents(self):
+        client = self._client()
+        md = artifacts.Metadata(name="wrapped", version="2.0", parents=("v1",))
+        p1, p2, p3 = _patched(client)
+        with p1, p2, p3:
+            await Artifact.create.aio(artifacts.new(_payload(), md))
+        req = client.artifact_service.create_artifact.await_args[0][0]
+        assert [p.version for p in req.spec.parent_artifacts] == ["v1"]
+
+    @pytest.mark.asyncio
+    async def test_duck_metadata_seeds_and_hash_versions(self):
+        # A protocol-carrying value (no wrapper) seeds name/parents, and
+        # version_from_content resolves to the literal's content hash.
+        class DuckFile(File):
+            def get_flyte_metadata(self):
+                return artifacts.Metadata(name="ducked", parents=("v1",), version_from_content=True)
+
+        client = self._client()
+        p1, p2, p3 = _patched(client)
+        with p1, p2, p3:
+            await Artifact.create.aio(DuckFile(path="s3://bucket/w.pt", hash="cafe"), python_type=File)
+        req = client.artifact_service.create_artifact.await_args[0][0]
+        assert req.artifact_id.name.name == "ducked"
+        assert req.artifact_id.version == "cafe"
+        assert [p.version for p in req.spec.parent_artifacts] == ["v1"]