add report_internal_runless_asset_event

Author: anuthebananu

python_modules/dagster/dagster/_core/instance/__init__.py

@@ -3483,7 +3483,6 @@ def report_runless_asset_event(
             "AssetObservation",
             "AssetCheckEvaluation",
             "FreshnessStateEvaluation",
-            "FreshnessStateChange",
         ],
     ):
         """Record an event log entry related to assets that does not belong to a Dagster run."""
@@ -3507,9 +3506,6 @@ def report_runless_asset_event(
         elif isinstance(asset_event, FreshnessStateEvaluation):
             event_type_value = DagsterEventType.FRESHNESS_STATE_EVALUATION.value
             data_payload = asset_event
-        elif isinstance(asset_event, FreshnessStateChange):
-            event_type_value = DagsterEventType.FRESHNESS_STATE_CHANGE.value
-            data_payload = asset_event
         else:
             raise DagsterInvariantViolationError(
                 f"Received unexpected asset event type {asset_event}, expected"
@@ -3525,6 +3521,29 @@ def report_runless_asset_event(
             ),
         )
 
+    def report_internal_runless_asset_event(self, asset_event: Union["FreshnessStateChange",]):
+        """Does exactly the same thing as report_runless_asset_event, only used for internal events.
+
+        Use this over report_runless_asset_event for internal events.
+        """
+        from dagster._core.events import DagsterEvent, DagsterEventType
+
+        if isinstance(asset_event, FreshnessStateChange):
+            event_type_value = DagsterEventType.FRESHNESS_STATE_CHANGE.value
+            data_payload = asset_event
+        else:
+            raise DagsterInvariantViolationError(
+                f"Received unexpected asset event type {asset_event}, expected FreshnessStateChange"
+            )
+        return self.report_dagster_event(
+            run_id=RUNLESS_RUN_ID,
+            dagster_event=DagsterEvent(
+                event_type_value=event_type_value,
+                event_specific_data=data_payload,
+                job_name=RUNLESS_JOB_NAME,
+            ),
+        )
+
     def get_entity_freshness_state(self, entity_key: AssetKey) -> Optional[FreshnessStateRecord]:
         warnings.warn("`get_entity_freshness_state` is not yet implemented for OSS.")
         return None