Make AMP use most recent materialization OR observation for observable assets (#22350)

## Summary & Motivation

Fixes
https://linear.app/dagster-labs/issue/FOU-219/[schedulingcondition]-fix-handling-of-externalassets-with-the.

For observable external assets, AMP is unresponsive to reported asset
materializations because it only looks for the most recent asset
observation. This PR changes `CachingInstanceQueryer` to look for the
most recent materialization or observation for observables, which allows
AMP to respond to reported materializations.

## How I Tested These Changes

Modified the unit tests for the `NewlyUpdated` conditions to include an
observable asset.

Author: smackesey

Diff for: python_modules/dagster/dagster/_utils/caching_instance_queryer.py

@@ -265,15 +265,25 @@ def _get_latest_materialization_or_observation_record(
             ),
             before_storage_id=before_cursor,
         )
+
+        # For observable assets, we fetch the most recent observation and materialization and return
+        # whichever is more recent. For non-observable assets, we just fetch the most recent
+        # materialization.
+        materialization_records = self.instance.fetch_materializations(
+            records_filter, ascending=False, limit=1
+        ).records
         if self.asset_graph.get(asset_partition.asset_key).is_observable:
-            records = self.instance.fetch_observations(
+            observation_records = self.instance.fetch_observations(
                 records_filter, ascending=False, limit=1
             ).records
+            all_records = sorted(
+                [*materialization_records, *observation_records],
+                key=lambda x: x.timestamp,
+                reverse=True,
+            )
         else:
-            records = self.instance.fetch_materializations(
-                records_filter, ascending=False, limit=1
-            ).records
-        return next(iter(records), None)
+            all_records = materialization_records
+        return next(iter(all_records), None)
 
     @cached_method
     def _get_latest_materialization_or_observation_storage_ids_by_asset_partition(

Diff for: python_modules/dagster/dagster_tests/definitions_tests/auto_materialize_tests/asset_condition_tests/test_newly_updated_condition.py

@@ -1,21 +1,22 @@
+import pytest
 from dagster import AutomationCondition
 
-from ..base_scenario import run_request
-from ..scenario_specs import one_asset
+from ..scenario_specs import one_asset, one_observable_asset
 from .asset_condition_scenario import AutomationConditionScenarioState
 
 
-def test_newly_updated_condition() -> None:
+@pytest.mark.parametrize("scenario", [one_asset, one_observable_asset])
+def test_newly_updated_condition(scenario) -> None:
     state = AutomationConditionScenarioState(
-        one_asset, automation_condition=AutomationCondition.newly_updated()
+        scenario, automation_condition=AutomationCondition.newly_updated()
     )
 
     # not updated
     state, result = state.evaluate("A")
     assert result.true_subset.size == 0
 
     # newly updated
-    state = state.with_runs(run_request("A"))
+    state = state.with_reported_materialization("A")
     state, result = state.evaluate("A")
     assert result.true_subset.size == 1
 
@@ -28,11 +29,11 @@ def test_newly_updated_condition() -> None:
     assert result.true_subset.size == 0
 
     # newly updated twice in a row
-    state = state.with_runs(run_request("A"))
+    state = state.with_reported_materialization("A")
     state, result = state.evaluate("A")
     assert result.true_subset.size == 1
 
-    state = state.with_runs(run_request("A"))
+    state = state.with_reported_materialization("A")
     state, result = state.evaluate("A")
     assert result.true_subset.size == 1
 

Diff for: python_modules/dagster/dagster_tests/definitions_tests/auto_materialize_tests/scenario_specs.py

@@ -3,6 +3,10 @@
 import pendulum
 from dagster import AssetSpec, MultiPartitionKey, StaticPartitionsDefinition
 from dagster._core.definitions.asset_dep import AssetDep
+from dagster._core.definitions.asset_spec import (
+    SYSTEM_METADATA_KEY_ASSET_EXECUTION_TYPE,
+    AssetExecutionType,
+)
 from dagster._core.definitions.multi_dimensional_partitions import MultiPartitionsDefinition
 from dagster._core.definitions.partition import DynamicPartitionsDefinition
 from dagster._core.definitions.partition_mapping import StaticPartitionMapping
@@ -40,6 +44,17 @@
 ##############
 one_asset = ScenarioSpec(asset_specs=[AssetSpec("A")])
 
+one_observable_asset = ScenarioSpec(
+    [
+        AssetSpec(
+            "A",
+            metadata={
+                SYSTEM_METADATA_KEY_ASSET_EXECUTION_TYPE: AssetExecutionType.OBSERVATION.value
+            },
+        )
+    ]
+)
+
 two_assets_in_sequence = ScenarioSpec(
     asset_specs=[AssetSpec("A"), AssetSpec("B", deps=["A"])],
 )

Diff for: python_modules/dagster/dagster_tests/definitions_tests/auto_materialize_tests/scenario_state.py

@@ -29,7 +29,7 @@
 from dagster._core.definitions import materialize
 from dagster._core.definitions.asset_graph import AssetGraph
 from dagster._core.definitions.definitions_class import create_repository_using_definitions_args
-from dagster._core.definitions.events import CoercibleToAssetKey
+from dagster._core.definitions.events import AssetMaterialization, CoercibleToAssetKey
 from dagster._core.definitions.executor_definition import in_process_executor
 from dagster._core.definitions.repository_definition.repository_definition import (
     RepositoryDefinition,
@@ -152,6 +152,7 @@ def _multi_asset(context: AssetExecutionContext):
                     "auto_materialize_policy",
                     "freshness_policy",
                     "partitions_def",
+                    "metadata",
                 }
                 assets.append(
                     asset(
@@ -325,6 +326,16 @@ def test_time_fn() -> float:
             ),
         )
 
+    def with_reported_materialization(
+        self, asset_key: CoercibleToAssetKey, partition_key: Optional[str] = None
+    ) -> Self:
+        mat = AssetMaterialization(
+            asset_key=asset_key,
+            partition=partition_key,
+        )
+        self.instance.report_runless_asset_event(mat)
+        return self
+
     def _with_runs_with_status(self, status: DagsterRunStatus) -> Self:
         """Create new runs that will run to completion for all existing runs with the given status,
         and delete the runs with that status from the instance.