A new asset freshness API (#28828)

## Summary & Motivation
Introduces a new API for defining freshness as a top-level attribute on an asset. This API will replace freshness checks and freshness policies as the way to model asset freshness in Dagster.

For more details on the motivation behind needing a new API, please refer to the [internal product review](https://www.notion.so/dagster/New-Asset-Freshness-System-Product-Review-1c318b92e46280fe859ce86f5d78934a?d=1c718b92e462807ca2e1001c0a331dc0#1c718b92e46280c8973ef9358ec38367).

The implementation here is slightly different than the desired end state, which is to reclaim the deprecated `FreshnessPolicy` construct. Instead we're writing the new freshness policy to the asset spec metadata. This will unblock internal testing and give us time to implement a longer-term migration plan for `FreshnessPolicy`.

### Auxiliary Changes
- Relocate `SerializableTimeDelta` from `dagster._core.definitions.declarative_automation.util` to `dagster_shared.serdes.utils`

## How I Tested These Changes
- new + existing unit tests for backward compatibility

## Changelog

> Insert changelog entry or delete this section.

Author: anuthebananu

Diff for: js_modules/dagster-ui/packages/ui-core/src/graphql/schema.graphql

@@ -11,7 +11,7 @@
     overload,
 )
 
-from dagster_shared.serdes import whitelist_for_serdes
+from dagster_shared.serdes import serialize_value, whitelist_for_serdes
 
 import dagster._check as check
 from dagster._annotations import (
@@ -26,6 +26,10 @@
     AutomationCondition,
 )
 from dagster._core.definitions.events import AssetKey, CoercibleToAssetKey
+from dagster._core.definitions.freshness import (
+    INTERNAL_FRESHNESS_POLICY_METADATA_KEY,
+    InternalFreshnessPolicy,
+)
 from dagster._core.definitions.freshness_policy import FreshnessPolicy
 from dagster._core.definitions.partition import PartitionsDefinition
 from dagster._core.definitions.partition_mapping import PartitionMapping
@@ -108,6 +112,11 @@ def validate_kind_tags(kinds: Optional[AbstractSet[str]]) -> None:
     breaking_version="1.10.0",
     additional_warn_text="use `automation_condition` instead",
 )
+@hidden_param(
+    param="internal_freshness_policy",
+    breaking_version="1.10.0",
+    additional_warn_text="experimental feature, use freshness checks instead",
+)
 @record_custom
 class AssetSpec(IHasInternalInit, IHaveNew, LegacyNamedTupleMixin):
     """Specifies the core attributes of an asset, except for the function that materializes or
@@ -195,6 +204,15 @@ def __new__(
         }
         validate_kind_tags(kind_tags)
 
+        internal_freshness_policy: Optional[InternalFreshnessPolicy] = kwargs.get(
+            "internal_freshness_policy"
+        )
+        if internal_freshness_policy:
+            metadata = {
+                **(metadata or {}),
+                INTERNAL_FRESHNESS_POLICY_METADATA_KEY: serialize_value(internal_freshness_policy),
+            }
+
         return super().__new__(
             cls,
             key=key,
@@ -435,3 +453,13 @@ def map_asset_specs(
         obj.map_asset_specs(func) if isinstance(obj, AssetsDefinition) else func(obj)
         for obj in iterable
     ]
+
+
+def attach_internal_freshness_policy(spec: AssetSpec, policy: InternalFreshnessPolicy) -> AssetSpec:
+    """Apply a freshness policy to an asset spec, attaching it to the spec's metadata.
+
+    You can use this in Definitions.map_asset_specs to attach a freshness policy to an asset spec.
+    """
+    return spec.merge_attributes(
+        metadata={INTERNAL_FRESHNESS_POLICY_METADATA_KEY: serialize_value(policy)}  # pyright: ignore[reportArgumentType]
+    )

Diff for: js_modules/dagster-ui/packages/ui-core/src/graphql/types.ts

@@ -3,6 +3,7 @@
 from typing import TYPE_CHECKING, Optional
 
 from dagster_shared.serdes import whitelist_for_serdes
+from dagster_shared.serdes.utils import SerializableTimeDelta
 
 from dagster._core.asset_graph_view.entity_subset import EntitySubset
 from dagster._core.definitions.asset_key import AssetCheckKey, AssetKey
@@ -14,7 +15,6 @@
 from dagster._core.definitions.declarative_automation.operands.subset_automation_condition import (
     SubsetAutomationCondition,
 )
-from dagster._core.definitions.declarative_automation.utils import SerializableTimeDelta
 from dagster._record import record
 from dagster._utils.schedules import reverse_cron_string_iterator
 

Diff for: python_modules/dagster/dagster/_core/definitions/asset_spec.py

@@ -1,27 +1 @@
-import datetime
-from typing import NamedTuple
 
-from dagster_shared.serdes import whitelist_for_serdes
-
-
-@whitelist_for_serdes
-class SerializableTimeDelta(NamedTuple):
-    """A Dagster-serializable version of a datetime.timedelta. The datetime.timedelta class
-    internally stores values as an integer number of days, seconds, and microseconds. This class
-    handles converting between the in-memory and serializable formats.
-    """
-
-    days: int
-    seconds: int
-    microseconds: int
-
-    @staticmethod
-    def from_timedelta(timedelta: datetime.timedelta) -> "SerializableTimeDelta":
-        return SerializableTimeDelta(
-            days=timedelta.days, seconds=timedelta.seconds, microseconds=timedelta.microseconds
-        )
-
-    def to_timedelta(self) -> datetime.timedelta:
-        return datetime.timedelta(
-            days=self.days, seconds=self.seconds, microseconds=self.microseconds
-        )

Diff for: python_modules/dagster/dagster/_core/definitions/declarative_automation/operands/operands.py

@@ -34,6 +34,7 @@
     CoercibleToAssetKey,
     CoercibleToAssetKeyPrefix,
 )
+from dagster._core.definitions.freshness import INTERNAL_FRESHNESS_POLICY_METADATA_KEY
 from dagster._core.definitions.freshness_policy import FreshnessPolicy
 from dagster._core.definitions.input import GraphIn
 from dagster._core.definitions.metadata import ArbitraryMetadataMapping, RawMetadataMapping
@@ -50,6 +51,7 @@
 from dagster._core.errors import DagsterInvalidDefinitionError
 from dagster._core.storage.tags import KIND_PREFIX
 from dagster._core.types.dagster_type import DagsterType
+from dagster._serdes import serialize_value
 from dagster._utils.tags import normalize_tags
 from dagster._utils.warnings import disable_dagster_warnings
 
@@ -138,6 +140,12 @@ def _validate_hidden_non_argument_dep_param(
     emit_runtime_warning=False,
     breaking_version="1.10.0",
 )
+@hidden_param(
+    param="internal_freshness_policy",
+    emit_runtime_warning=False,
+    breaking_version="1.10.0",
+    additional_warn_text="experimental, use freshness checks instead.",
+)
 def asset(
     compute_fn: Optional[Callable[..., Any]] = None,
     *,
@@ -293,6 +301,13 @@ def downstream_asset(conditional_asset):
         **{f"{KIND_PREFIX}{kind}": "" for kind in kinds or []},
     }
 
+    internal_freshness_policy = kwargs.get("internal_freshness_policy")
+    if internal_freshness_policy:
+        metadata = {
+            **(metadata or {}),
+            INTERNAL_FRESHNESS_POLICY_METADATA_KEY: serialize_value(internal_freshness_policy),
+        }
+
     only_allow_hidden_params_in_kwargs(asset, kwargs)
 
     args = AssetDecoratorArgs(

Diff for: python_modules/dagster/dagster/_core/definitions/declarative_automation/utils.py

@@ -0,0 +1,66 @@
+import datetime
+from abc import ABC
+from collections.abc import Mapping
+from enum import Enum
+from typing import Any, Optional
+
+from dagster_shared.serdes.utils import SerializableTimeDelta
+
+from dagster._core.definitions.asset_key import AssetKey
+from dagster._record import IHaveNew, record
+from dagster._serdes import deserialize_value, whitelist_for_serdes
+from dagster._utils import check
+
+
+@whitelist_for_serdes
+class FreshnessState(str, Enum):
+    PASS = "PASS"
+    WARN = "WARN"
+    FAIL = "FAIL"
+    UNKNOWN = "UNKNOWN"
+
+
+@whitelist_for_serdes
+@record
+class FreshnessStateEvaluation:
+    key: AssetKey
+    freshness_state: FreshnessState
+
+
+INTERNAL_FRESHNESS_POLICY_METADATA_KEY = "dagster/internal_freshness_policy"
+
+
+class InternalFreshnessPolicy(ABC):
+    @classmethod
+    def from_asset_spec_metadata(
+        cls, metadata: Mapping[str, Any]
+    ) -> Optional["InternalFreshnessPolicy"]:
+        serialized_policy = metadata.get(INTERNAL_FRESHNESS_POLICY_METADATA_KEY)
+        if serialized_policy is None:
+            return None
+        return deserialize_value(serialized_policy.value, cls)  # pyright: ignore
+
+    @staticmethod
+    def time_window(
+        fail_window: datetime.timedelta, warn_window: Optional[datetime.timedelta] = None
+    ) -> "TimeWindowFreshnessPolicy":
+        return TimeWindowFreshnessPolicy.from_timedeltas(fail_window, warn_window)
+
+
+@whitelist_for_serdes
+@record
+class TimeWindowFreshnessPolicy(InternalFreshnessPolicy, IHaveNew):
+    fail_window: SerializableTimeDelta
+    warn_window: Optional[SerializableTimeDelta] = None
+
+    @classmethod
+    def from_timedeltas(
+        cls, fail_window: datetime.timedelta, warn_window: Optional[datetime.timedelta] = None
+    ):
+        if warn_window:
+            check.invariant(warn_window < fail_window, "warn_window must be less than fail_window")
+
+        return cls(
+            fail_window=SerializableTimeDelta.from_timedelta(fail_window),
+            warn_window=SerializableTimeDelta.from_timedelta(warn_window) if warn_window else None,
+        )

Diff for: python_modules/dagster/dagster/_core/definitions/decorators/asset_decorator.py

@@ -67,6 +67,7 @@
 
 if TYPE_CHECKING:
     from dagster._core.definitions.events import ObjectStoreOperation
+    from dagster._core.definitions.freshness import FreshnessStateEvaluation
     from dagster._core.execution.plan.plan import ExecutionPlan
     from dagster._core.execution.plan.step import StepKind
 
@@ -93,6 +94,7 @@
     "AssetCheckEvaluationPlanned",
     "AssetFailedToMaterializeData",
     "RunEnqueuedData",
+    "FreshnessStateEvaluation",
 ]
 
 
@@ -169,6 +171,8 @@ class DagsterEventType(str, Enum):
 
     LOGS_CAPTURED = "LOGS_CAPTURED"
 
+    FRESHNESS_STATE_EVALUATION = "FRESHNESS_STATE_EVALUATION"
+
 
 EVENT_TYPE_TO_DISPLAY_STRING = {
     DagsterEventType.PIPELINE_ENQUEUED: "RUN_ENQUEUED",

Diff for: python_modules/dagster/dagster/_core/definitions/freshness.py

@@ -37,6 +37,7 @@
 )
 from dagster._core.definitions.data_version import extract_data_provenance_from_entry
 from dagster._core.definitions.events import AssetKey, AssetObservation
+from dagster._core.definitions.freshness import FreshnessStateEvaluation
 from dagster._core.definitions.partition_key_range import PartitionKeyRange
 from dagster._core.errors import (
     DagsterHomeNotSetError,
@@ -3351,7 +3352,12 @@ def get_latest_materialization_code_versions(
     @public
     def report_runless_asset_event(
         self,
-        asset_event: Union["AssetMaterialization", "AssetObservation", "AssetCheckEvaluation"],
+        asset_event: Union[
+            "AssetMaterialization",
+            "AssetObservation",
+            "AssetCheckEvaluation",
+            "FreshnessStateEvaluation",
+        ],
     ):
         """Record an event log entry related to assets that does not belong to a Dagster run."""
         from dagster._core.events import (
@@ -3371,10 +3377,13 @@ def report_runless_asset_event(
         elif isinstance(asset_event, AssetObservation):
             event_type_value = DagsterEventType.ASSET_OBSERVATION.value
             data_payload = AssetObservationData(asset_event)
+        elif isinstance(asset_event, FreshnessStateEvaluation):
+            event_type_value = DagsterEventType.FRESHNESS_STATE_EVALUATION.value
+            data_payload = asset_event
         else:
             raise DagsterInvariantViolationError(
                 f"Received unexpected asset event type {asset_event}, expected"
-                " AssetMaterialization, AssetObservation or AssetCheckEvaluation"
+                " AssetMaterialization, AssetObservation, AssetCheckEvaluation or FreshnessStateEvaluation"
             )
 
         return self.report_dagster_event(

Diff for: python_modules/dagster/dagster/_core/events/__init__.py

@@ -61,6 +61,7 @@
     OpNode,
 )
 from dagster._core.definitions.events import AssetKey
+from dagster._core.definitions.freshness import InternalFreshnessPolicy
 from dagster._core.definitions.freshness_policy import FreshnessPolicy
 from dagster._core.definitions.metadata import (
     MetadataFieldSerializer,
@@ -1553,6 +1554,10 @@ def automation_condition(self) -> Optional[AutomationCondition]:
         else:
             return None
 
+    @property
+    def internal_freshness_policy(self) -> Optional[InternalFreshnessPolicy]:
+        return InternalFreshnessPolicy.from_asset_spec_metadata(self.metadata)
+
 
 ResourceJobUsageMap: TypeAlias = dict[str, list[ResourceJobUsageEntry]]