Don't write a cursor during the global DA => DA sensor migration for sensors that don't target any asset keys (#32794)

Summary:
Prevents an issue where:
- there is a large cursor
- we are migrating from global DA to sensors
- there is one or more sensors that don't actually target any assets
- the large cursor is written to that sensor
- the large cursor is never removed because the sensor never
executes/updates

Simulate a DA transition via script
Existing test coverage of the migration transition

## Summary & Motivation

## How I Tested These Changes

## Changelog

> Insert changelog entry or delete this section.

Author: gibsondan

python_modules/dagster/dagster/_daemon/asset_daemon.py

@@ -644,9 +644,35 @@ def _create_initial_sensor_cursors_from_raw_cursor(
 
         result = {}
 
-        initial_cursor = asset_daemon_cursor_to_instigator_serialized_cursor(pre_sensor_cursor)
-
         for sensor, repo in sensors_and_repos:
+            selection = sensor.asset_selection
+            if not selection:
+                continue
+
+            repo_asset_graph = repo.asset_graph
+            resolved_keys = selection.resolve(repo_asset_graph) | selection.resolve_checks(
+                repo_asset_graph
+            )
+
+            serialized_cursor = None
+
+            if len(resolved_keys) > 0:
+                # filter down the cursor to just the keys targeted by the sensor
+                condition_cursors = [
+                    condition_cursor
+                    for condition_cursor in (pre_sensor_cursor.previous_condition_cursors or [])
+                    if condition_cursor.key in resolved_keys
+                ]
+
+                cursor_to_use = dataclasses.replace(
+                    pre_sensor_cursor,
+                    previous_condition_cursors=condition_cursors,
+                )
+
+                serialized_cursor = asset_daemon_cursor_to_instigator_serialized_cursor(
+                    cursor_to_use
+                )
+
             new_auto_materialize_state = InstigatorState(
                 sensor.get_remote_origin(),
                 InstigatorType.SENSOR,
@@ -657,7 +683,7 @@ def _create_initial_sensor_cursors_from_raw_cursor(
                 ),
                 SensorInstigatorData(
                     min_interval=sensor.min_interval_seconds,
-                    cursor=initial_cursor,
+                    cursor=serialized_cursor,
                     last_sensor_start_timestamp=get_current_timestamp(),
                     sensor_type=sensor.sensor_type,
                 ),

python_modules/dagster/dagster_tests/declarative_automation_tests/daemon_tests/test_asset_daemon.py

@@ -6,6 +6,7 @@
 from unittest import mock
 
 import dagster as dg
+import dagster._check as check
 import pytest
 from dagster import AutoMaterializeRule, AutomationCondition, DagsterInstance
 from dagster._core.definitions.asset_daemon_cursor import AssetDaemonCursor
@@ -449,6 +450,12 @@ def test_daemon_skip_env_var() -> None:
                 default_status=DefaultSensorStatus.STOPPED,
                 minimum_interval_seconds=15,
             ),
+            dg.AutomationConditionSensorDefinition(
+                name="auto_materialize_sensor_no_assets",
+                target=AssetSelection.groups("nonexistant"),
+                default_status=DefaultSensorStatus.STOPPED,
+                minimum_interval_seconds=15,
+            ),
             # default sensor picks up "C"
         ]
     ).with_all_eager(3),
@@ -573,7 +580,7 @@ def test_auto_materialize_sensor_transition():
             instigator_type=InstigatorType.SENSOR
         )
 
-        assert len(sensor_states) == 3  # sensor states for each sensor were created
+        assert len(sensor_states) == 4  # sensor states for each sensor were created
 
         # Only sensor that was set with default status RUNNING turned on and ran
         _assert_sensor_state(
@@ -594,16 +601,33 @@ def test_auto_materialize_sensor_transition():
             expected_num_ticks=1,
             expected_status=InstigatorStatus.RUNNING,
         )
+        _assert_sensor_state(
+            instance,
+            "auto_materialize_sensor_no_assets",
+            expected_num_ticks=0,
+            expected_status=InstigatorStatus.RUNNING,
+        )
+
+        asset_graph = daemon_sensor_scenario.initial_spec.asset_graph
+        defs = daemon_sensor_scenario.initial_spec.defs
 
         for sensor_state in sensor_states:
-            # cursor was propagated to each sensor, so all subsequent evaluation IDs are higher
-            assert (
-                asset_daemon_cursor_from_instigator_serialized_cursor(
+            if sensor_state.instigator_name == "auto_materialize_sensor_no_assets":
+                assert cast("SensorInstigatorData", sensor_state.instigator_data).cursor is None
+            else:
+                asset_daemon_cursor = asset_daemon_cursor_from_instigator_serialized_cursor(
                     cast("SensorInstigatorData", sensor_state.instigator_data).cursor,
                     None,
-                ).evaluation_id
-                > pre_sensor_evaluation_id
-            )
+                )
+
+                assert asset_daemon_cursor.evaluation_id > pre_sensor_evaluation_id
+
+                assert (
+                    asset_daemon_cursor.previous_condition_cursors_by_key.keys()
+                    == check.not_none(
+                        defs.get_sensor_def(sensor_state.instigator_name).asset_selection
+                    ).resolve(asset_graph)
+                )
 
 
 # this scenario simulates the true default case in which we have an implicit default sensor
@@ -737,7 +761,7 @@ def test_auto_materialize_sensor_ticks(num_threads):
                 instigator_type=InstigatorType.SENSOR
             )
 
-            assert len(sensor_states) == 3  # sensor states for each sensor were created
+            assert len(sensor_states) == 4  # sensor states for each sensor were created
 
             # Only sensor that was set with default status RUNNING turned on and ran
             _assert_sensor_state(
@@ -778,7 +802,7 @@ def test_auto_materialize_sensor_ticks(num_threads):
             sensor_states = instance.schedule_storage.all_instigator_state(  # pyright: ignore[reportOptionalMemberAccess]
                 instigator_type=InstigatorType.SENSOR
             )
-            assert len(sensor_states) == 3
+            assert len(sensor_states) == 4
 
             # No new tick yet for A since only 15 seconds have passed
             _assert_sensor_state(
@@ -809,7 +833,7 @@ def test_auto_materialize_sensor_ticks(num_threads):
                 instigator_type=InstigatorType.SENSOR
             )
 
-            assert len(sensor_states) == 3
+            assert len(sensor_states) == 4
             _assert_sensor_state(
                 instance,
                 "auto_materialize_sensor_a",