Add event timestamps to automation conditions for precise trigger/reset resolution (#21494)

When both trigger and reset conditions fire on the same evaluation tick, timestamp comparison now determines which event happened more recently (per-partition). This eliminates the special-case hack for CronTickPassed + NewlyUpdated and enables correct behavior for all condition combinations.

## Key Changes

- Add `TimingMetadata` class to map event timestamps to entity subsets, stored in-memory on `AutomationResult`
- Create `TimedSubsetAutomationCondition` base class for conditions that support per-partition timestamps
- Enrich `CronTickPassedCondition` and `NewlyUpdatedCondition` to return timing metadata (cron tick time and materialization timestamps respectively)
- Add `compute_subsets_by_latest_materialization_timestamp()` to asset graph view for efficient per-partition timestamp lookup with configurable threshold fallback
- Replace SinceCondition's hardcoded `CronTickPassed + NewlyUpdated` special case with generic timestamp-based resolution via `subset_with_later_timestamps_than()`

<details>
<summary>Files Changed</summary>

### Added (2 files)
- `dagster/_core/asset_graph_view/timing_metadata.py` - TimingMetadata class for timestamp → EntitySubset mapping
- `dagster_tests/declarative_automation_tests/automation_condition_tests/builtins/test_event_timestamps.py` - Comprehensive timestamp resolution tests

### Modified (5 files)
- `dagster/_core/definitions/declarative_automation/automation_condition.py` - Add timing_metadata field to AutomationResult
- `dagster/_core/definitions/declarative_automation/operands/subset_automation_condition.py` - New TimedSubsetAutomationCondition base class
- `dagster/_core/definitions/declarative_automation/operands/operands.py` - CronTickPassed and NewlyUpdatedCondition now return TimingMetadata
- `dagster/_core/asset_graph_view/asset_graph_view.py` - Add compute_subsets_by_latest_materialization_timestamp() with threshold-based fallback
- `dagster_tests/declarative_automation_tests/automation_condition_tests/fundamentals/test_result_value_hash.py` - Adjust timing in tests

</details>

## Critical Notes

- Timestamps are in-memory only (not serialized); consumed same-tick by parent SinceCondition
- Per-partition timestamp fetching respects configurable threshold (env var `DAGSTER_MAX_PARTITIONS_FOR_DA_TIMESTAMP_FETCH`, default 100); exceeding it falls back to tick timestamp
- Fully backward compatible: conditions without timing metadata fall back to current behavior (reset wins when both fire simultaneously)

<details>
<summary>original-plan</summary>

# Plan: Add Event Timestamps to Automation Condition Results

## Context

The `SinceCondition` resolves trigger/reset at tick granularity (30s–few minutes). When both trigger and reset fire on the same tick, ordering is ambiguous — there's even a special-case hack (lines 136-170 in `since_operator.py`) for the `NewlyUpdated` + `CronTickPassed` combination. With real event timestamps, SinceCondition could make **per-partition, timestamp-precise** decisions about whether the trigger or reset happened more recently, eliminating the hack and enabling correct behavior for all condition combinations.

### Current problem illustrated

1. Cron tick fires at 12:00:00
2. Asset partition materializes at 12:00:05
3. DA tick runs at 12:00:30 — sees both CronTickPassed and NewlyUpdated as true
4. Without real timestamps: ambiguous ordering, resolved by a hardcoded special case
5. With real timestamps: CronTickPassed has event_timestamp=12:00:00, NewlyUpdated has event_timestamp=12:00:05 → trigger happened after reset → partition stays in true_subset

## Approach: `subsets_with_metadata` + `SubsetResult` return type

Use the existing `subsets_with_metadata` field (already on `AutomationResult`, `AutomationConditionNodeCursor`, `AutomationConditionEvaluation`, and wired through GraphQL) to carry event timestamps. Conditions opt in by returning a `SubsetResult` from `compute_subset()`.

Always-on for conditions that support it. Exact per-partition timestamps (not bucketed).

---

## Implementation Steps

### Step 1: Define `SubsetResult` in `SubsetAutomationCondition`

**File**: `dagster-oss/python_modules/dagster/dagster/_core/definitions/declarative_automation/operands/subset_automation_condition.py`

Add a result type that wraps a subset with optional metadata:

```python
@record
class SubsetResult:
    subset: EntitySubset
    subsets_with_metadata: Sequence[AssetSubsetWithMetadata] = ()
```

Modify `SubsetAutomationCondition.evaluate()` to detect `SubsetResult` vs plain `EntitySubset` from `compute_subset()`, and pass `subsets_with_metadata` through to `AutomationResult`:

```python
async def evaluate(self, context):
    if context.candidate_subset.is_empty:
        true_subset = context.get_empty_subset()
        subsets_with_metadata = []
    elif inspect.iscoroutinefunction(self.compute_subset):
        result = await self.compute_subset(context)
    else:
        result = self.compute_subset(context)

    if isinstance(result, SubsetResult):
        true_subset = result.subset
        subsets_with_metadata = result.subsets_with_metadata
    else:
        true_subset = result
        subsets_with_metadata = []

    return AutomationResult(context, true_subset, subsets_with_metadata=subsets_with_metadata)
```

Note: `SubsetAutomationCondition.requires_cursor` is currently `False`. Since `subsets_with_metadata` is stored via `node_cursor`, and `node_cursor` returns `None` when `requires_cursor` is `False`, we need to either:
- Override `requires_cursor` to return `True` when metadata is present, or
- Store the metadata on `AutomationConditionEvaluation` only (not on cursor) since it's for same-tick consumption by the parent SinceCondition

**Decision needed**: Since the timestamps are consumed by the parent SinceCondition on the *same* tick (not across ticks), we may not need cursor persistence. The `subsets_with_metadata` on the child's `AutomationResult` is directly accessible to SinceCondition via `trigger_result._subsets_with_metadata`. This avoids needing to enable cursoring on leaf conditions.

### Step 2: Enrich `CronTickPassedCondition`

**File**: `dagster-oss/python_modules/dagster/dagster/_core/definitions/declarative_automation/operands/operands.py`

When the cron tick has passed, return a `SubsetResult` with the exact cron tick timestamp:

```python
def compute_subset(self, context):
    previous_cron_tick = self._get_previous_cron_tick(context.evaluation_time)
    if context.previous_evaluation_time is None or previous_cron_tick < context.previous_evaluation_time:
        return context.get_empty_subset()
    else:
        return SubsetResult(
            subset=context.candidate_subset,
            subsets_with_metadata=[AssetSubsetWithMetadata(
                subset=context.candidate_subset.convert_to_serializable_subset(),
                metadata={"event_timestamp": FloatMetadataValue(previous_cron_tick.timestamp())},
            )],
        )
```

Single entry — the cron tick time is uniform across all partitions.

### Step 3: Enrich `NewlyUpdatedCondition` with per-partition timestamps

This requires extending the DB query layer.

**File**: `dagster-oss/python_modules/dagster/dagster/_utils/caching_instance_queryer.py`

The existing `get_asset_partitions_updated_after_cursor()` calls `_get_latest_materialization_or_observation_storage_ids_by_asset_partition()` which returns `Mapping[AssetKeyPartitionKey, int]` (partition → storage_id). We need to also get timestamps.

Option A: Add a companion method `get_asset_partition_timestamps_updated_after_cursor()` that returns `Mapping[AssetKeyPartitionKey, float]`.

Option B: Modify the return type to include both storage_id and timestamp. The underlying `instance.get_latest_storage_id_by_partition()` would need to also return timestamps.

**Recommended**: Option A — less invasive. After getting the updated partition keys (which we already do), make a follow-up query to get event timestamps for those specific storage IDs. Since the set of newly updated partitions per tick is small, this is cheap.

**File**: `dagster-oss/python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py`

Add a new method `compute_updated_since_temporal_context_with_timestamps()` that returns both the subset and a mapping of partition keys to timestamps. The `NewlyUpdatedCondition` calls this instead.

**File**: `dagster-oss/python_modules/dagster/dagster/_core/definitions/declarative_automation/operands/operands.py`

Modify `NewlyUpdatedCondition.compute_subset()`:

```python
async def compute_subset(self, context):
    if context.previous_temporal_context is None:
        return context.get_empty_subset()
    subset, partition_timestamps = await context.asset_graph_view.compute_updated_since_temporal_context_with_timestamps(
        key=context.key, temporal_context=context.previous_temporal_context
    )
    # Group partitions by timestamp
    subsets_with_metadata = _group_by_timestamp(subset, partition_timestamps)
    return SubsetResult(subset=subset, subsets_with_metadata=subsets_with_metadata)
```

Helper `_group_by_timestamp` groups partition keys sharing the same timestamp into compressed subsets with metadata.

### Step 4: Use timestamps in `SinceCondition` for precise resolution

**File**: `dagster-oss/python_modules/dagster/dagster/_core/definitions/declarative_automation/operators/since_operator.py`

This is the core change. Replace the special-case ordering hack with timestamp-based per-partition resolution:

```python
async def evaluate(self, context):
    # ... evaluate trigger and reset as before ...

    true_subset = context.previous_true_subset or context.get_empty_subset()

    # Extract per-partition timestamps from children
    trigger_timestamps = _extract_partition_timestamps(trigger_result)
    reset_timestamps = _extract_partition_timestamps(reset_result)

    # For partitions in trigger but not reset (or vice versa), simple union/difference
    trigger_only = trigger_result.true_subset.compute_difference(reset_result.true_subset)
    reset_only = reset_result.true_subset.compute_difference(trigger_result.true_subset)
    both = trigger_result.true_subset.compute_intersection(reset_result.true_subset)

    true_subset = true_subset.compute_union(trigger_only)
    true_subset = true_subset.compute_difference(reset_only)

    # For partitions in BOTH trigger and reset on same tick:
    # use timestamps to determine which happened more recently
    if not both.is_empty:
        if trigger_timestamps and reset_timestamps:
            # Per-partition comparison: keep if trigger timestamp > reset timestamp
            trigger_wins = _partitions_where_trigger_after_reset(
                both, trigger_timestamps, reset_timestamps
            )
            true_subset = true_subset.compute_union(trigger_wins)
            # (partitions where reset wins are already excluded by not being unioned)
        else:
            # Fallback: no timestamps available, use current ordering behavior
            true_subset = true_subset.compute_union(both)  # or difference, depending on default

    # ... update SinceConditionData, return result ...
```

The `_extract_partition_timestamps` helper reads `result._subsets_with_metadata` to build a `Mapping[str, float]` from partition key to `event_timestamp`.

**Key benefit**: The special-case `reset_newly_true` hack for `CronTickPassed + NewlyUpdated` can be **removed** — timestamp comparison handles this generically for all condition combinations.

### Step 5: Propagate timestamps through SinceCondition result

SinceCondition should also emit `subsets_with_metadata` on its own result, carrying the trigger timestamps for partitions in its `true_subset`. This allows parent conditions or the UI to see when each partition was triggered.

---

## Critical Files

| File | Change |
|------|--------|
| `.../operands/subset_automation_condition.py` | Add `SubsetResult`, modify `evaluate()` |
| `.../operands/operands.py` | Modify `CronTickPassedCondition` and `NewlyUpdatedCondition` |
| `.../operators/since_operator.py` | Timestamp-based resolution, remove ordering hack |
| `.../caching_instance_queryer.py` | Add timestamp fetching for updated partitions |
| `.../asset_graph_view/asset_graph_view.py` | Add `compute_updated_since_..._with_timestamps()` |
| `.../serialized_objects.py` | No changes needed — `subsets_with_metadata` already exists |

## Backward Compatibility

- `subsets_with_metadata` was previously empty for modern conditions — populating it is additive
- SinceCondition falls back to current behavior when timestamps aren't available
- No cursor schema changes needed (timestamps are consumed same-tick, not persisted across ticks for this use case)
- Existing serialized cursors deserialize fine

## Verification

1. Existing tests pass: `test_since_condition.py`, `test_updated_since_cron_condition.py`
2. New tests:
   - CronTickPassed returns `SubsetResult` with correct cron tick timestamp
   - NewlyUpdated returns `SubsetResult` with per-partition materialization timestamps
   - SinceCondition with both trigger+reset on same tick resolves correctly using timestamps
   - SinceCondition without timestamps (legacy conditions) falls back to current behavior
   - The CronTickPassed+NewlyUpdated special case works correctly without the hardcoded hack
3. Integration: verify via the automation condition evaluation UI that timestamps appear in metadata

</details>

<!-- WARNING: Machine-generated. Manual edits may break erk tooling. -->
<!-- erk:metadata-block:plan-header -->
<details>
<summary>plan-header</summary>

```yaml

schema_version: '2'
created_at: '2026-03-09T11:09:55.909154+00:00'
created_by: OwenKephart
plan_comment_id: null
last_dispatched_run_id: null
last_dispatched_node_id: null
last_dispatched_at: null
last_local_impl_at: '2026-03-09T18:35:34.455616+00:00'
last_local_impl_event: ended
last_local_impl_session: 58b2ad56-024a-4c2e-968c-96499cb08551
last_local_impl_user: owen
last_remote_impl_at: null
last_remote_impl_run_id: null
last_remote_impl_session_id: null
branch_name: plnd/add-event-timestamps-since-03-09-1109
created_from_session: 97ef5a88-1256-4b7b-bdf7-1499b38cd062
lifecycle_stage: impl
last_session_branch: async-learn/21494
last_session_id: 58b2ad56-024a-4c2e-968c-96499cb08551
last_session_at: '2026-03-09T11:35:41.024671+00:00'
last_session_source: local
worktree_name: internal-d

```

</details>
<!-- /erk:metadata-block:plan-header -->
---

To replicate this PR locally, run:

```
erk pr teleport 21494
```

Internal-RevId: 94ca2d5557265db32db21c23a4721acc3858f8d4

Author: OwenKephart

python_modules/dagster/dagster/_core/asset_graph_view/asset_graph_view.py

@@ -923,6 +923,56 @@ async def _compute_updated_since_cursor_subset(
         ).value
         return EntitySubset(self, key=key, value=_ValidatedEntitySubsetValue(value))
 
+    def compute_subsets_by_latest_materialization_timestamp(
+        self, subset: EntitySubset[AssetKey]
+    ) -> dict[float, EntitySubset[AssetKey]]:
+        """Returns a mapping from materialization event timestamps to entity subsets.
+
+        For unpartitioned assets, uses the cached asset record. For partitioned assets,
+        fetches storage IDs per partition and groups by timestamp.
+        """
+        from dagster._core.event_api import AssetRecordsFilter
+
+        key = check.inst(subset.key, AssetKey)
+
+        # Fast path for unpartitioned assets — use cached asset record
+        if not subset.is_partitioned:
+            asset_record = self._queryer.get_asset_record(key)
+            record = asset_record.asset_entry.last_materialization_record if asset_record else None
+            if record is not None:
+                return {record.timestamp: subset}
+            return {}
+
+        # Partitioned: use caching queryer to get storage IDs per partition
+        asset_partitions = subset.expensively_compute_asset_partitions()
+        valid_storage_ids = {}
+        for ap in asset_partitions:
+            sid = self._queryer.get_latest_materialization_or_observation_storage_id(ap)
+            if sid is not None:
+                valid_storage_ids[ap] = sid
+        if not valid_storage_ids:
+            return {}
+
+        result = self._queryer.instance.fetch_materializations(
+            AssetRecordsFilter(
+                asset_key=key,
+                storage_ids=list(valid_storage_ids.values()),
+            ),
+            limit=len(valid_storage_ids),
+        )
+        storage_id_to_ts = {r.storage_id: r.timestamp for r in result.records}
+
+        # Group partitions by timestamp
+        by_timestamp: dict[float, set[AssetKeyPartitionKey]] = {}
+        for ap, sid in valid_storage_ids.items():
+            if sid in storage_id_to_ts:
+                by_timestamp.setdefault(storage_id_to_ts[sid], set()).add(ap)
+
+        return {
+            ts: self.get_asset_subset_from_asset_partitions(key, akpks)
+            for ts, akpks in by_timestamp.items()
+        }
+
     async def _compute_updated_since_time_subset(
         self, key: AssetCheckKey, time: datetime
     ) -> EntitySubset[AssetCheckKey]:

python_modules/dagster/dagster/_core/asset_graph_view/timing_metadata.py

@@ -0,0 +1,33 @@
+from collections.abc import Mapping
+from typing import Generic
+
+from dagster._core.asset_graph_view.entity_subset import EntitySubset
+from dagster._core.definitions.asset_key import T_EntityKey
+from dagster._record import record
+
+
+@record
+class TimingMetadata(Generic[T_EntityKey]):
+    """Encapsulates per-timestamp entity subsets from an AutomationResult.
+
+    Stores a mapping from event timestamp to the EntitySubset that occurred at that
+    timestamp. Used by SinceCondition to make timestamp-precise decisions when both
+    trigger and reset conditions fire on the same evaluation tick.
+    """
+
+    timestamps: Mapping[float, EntitySubset[T_EntityKey]]
+
+    def subset_with_later_timestamps_than(
+        self,
+        other: "TimingMetadata[T_EntityKey]",
+        empty: EntitySubset[T_EntityKey],
+    ) -> EntitySubset[T_EntityKey]:
+        """Returns the subset where self has a later timestamp than other."""
+        result = empty
+        for self_ts, self_subset in self.timestamps.items():
+            to_add = self_subset
+            for other_ts, other_subset in other.timestamps.items():
+                if other_ts >= self_ts:
+                    to_add = to_add.compute_difference(other_subset)
+            result = result.compute_union(to_add)
+        return result

python_modules/dagster/dagster/_core/definitions/declarative_automation/automation_condition.py

@@ -11,6 +11,7 @@
 from dagster._annotations import beta, hidden_param, only_allow_hidden_params_in_kwargs, public
 from dagster._core.asset_graph_view.entity_subset import EntitySubset
 from dagster._core.asset_graph_view.serializable_entity_subset import SerializableEntitySubset
+from dagster._core.asset_graph_view.timing_metadata import TimingMetadata
 from dagster._core.definitions.asset_key import (
     AssetCheckKey,
     AssetKey,
@@ -884,6 +885,7 @@ def _get_stable_unique_id(self, target_key: EntityKey | None) -> str:
 @hidden_param(param="subsets_with_metadata", breaking_version="", emit_runtime_warning=False)
 @hidden_param(param="structured_cursor", breaking_version="", emit_runtime_warning=False)
 @hidden_param(param="metadata", breaking_version="", emit_runtime_warning=False)
+@hidden_param(param="timing_metadata", breaking_version="", emit_runtime_warning=False)
 class AutomationResult(Generic[T_EntityKey]):
     """The result of evaluating an AutomationCondition."""
 
@@ -916,6 +918,11 @@ def __init__(
             kwargs.get("metadata"), "metadata", key_type=str, value_type=object
         )
 
+        # in-memory-only timing metadata for timestamp-precise trigger/reset resolution
+        self._timing_metadata: TimingMetadata | None = check.opt_inst_param(
+            kwargs.get("timing_metadata"), "timing_metadata", TimingMetadata
+        )
+
         # hidden_param which should only be set by builtin conditions which require high performance
         # in their serdes layer
         structured_cursor = kwargs.get("structured_cursor")
@@ -939,6 +946,10 @@ def key(self) -> T_EntityKey:
     def true_subset(self) -> EntitySubset[T_EntityKey]:
         return self._true_subset
 
+    @property
+    def timing_metadata(self) -> TimingMetadata | None:
+        return self._timing_metadata
+
     @property
     def start_timestamp(self) -> float:
         return self._start_timestamp

python_modules/dagster/dagster/_core/definitions/declarative_automation/operands/operands.py

@@ -1,9 +1,12 @@
 import datetime
+import logging
+import os
 
 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.asset_graph_view.timing_metadata import TimingMetadata
 from dagster._core.definitions.asset_key import AssetCheckKey, AssetKey
 from dagster._core.definitions.declarative_automation.automation_condition import (
     AutomationResult,
@@ -12,6 +15,7 @@
 from dagster._core.definitions.declarative_automation.automation_context import AutomationContext
 from dagster._core.definitions.declarative_automation.operands.subset_automation_condition import (
     SubsetAutomationCondition,
+    TimedSubsetAutomationCondition,
 )
 from dagster._core.definitions.freshness import FreshnessState
 from dagster._core.definitions.partitions.snap.snap import PartitionsSnap
@@ -185,18 +189,53 @@ def compute_subset(self, context: AutomationContext) -> EntitySubset:
 
 @whitelist_for_serdes
 @record
-class NewlyUpdatedCondition(SubsetAutomationCondition):
+class NewlyUpdatedCondition(TimedSubsetAutomationCondition):
     @property
     def name(self) -> str:
         return "newly_updated"
 
-    async def compute_subset(self, context: AutomationContext) -> EntitySubset:  # pyright: ignore[reportIncompatibleMethodOverride]
+    async def compute_subset_with_timing_metadata(  # pyright: ignore[reportIncompatibleMethodOverride]
+        self, context: AutomationContext
+    ) -> tuple[EntitySubset, TimingMetadata | None]:
         # if it's the first time evaluating, just return the empty subset
         if context.previous_temporal_context is None:
-            return context.get_empty_subset()
-        return await context.asset_graph_view.compute_updated_since_temporal_context_subset(
+            return context.get_empty_subset(), None
+
+        if not isinstance(context.key, AssetKey):
+            # For non-asset keys (e.g. checks), no timestamp enrichment yet
+            subset = await context.asset_graph_view.compute_updated_since_temporal_context_subset(
+                key=context.key, temporal_context=context.previous_temporal_context
+            )
+            return subset, None
+
+        subset = await context.asset_graph_view.compute_updated_since_temporal_context_subset(
             key=context.key, temporal_context=context.previous_temporal_context
         )
+        if subset.is_empty:
+            return subset, None
+
+        max_partitions = int(os.environ.get("DAGSTER_MAX_PARTITIONS_FOR_DA_TIMESTAMP_FETCH", "100"))
+        if subset.size > max_partitions:
+            # Too many partitions to fetch individual timestamps — use the current
+            # tick timestamp for the entire subset so that downstream SinceCondition
+            # logic still has timing metadata to work with.
+            logging.getLogger("dagster").warning(
+                "Asset %s has %d updated partitions, exceeding the maximum of %d for"
+                " per-partition timestamp fetching. Using tick timestamp as fallback.",
+                context.key.to_user_string(),
+                subset.size,
+                max_partitions,
+            )
+            return subset, TimingMetadata(
+                timestamps={context.asset_graph_view.effective_dt.timestamp(): subset}
+            )
+
+        timing_subsets = (
+            context.asset_graph_view.compute_subsets_by_latest_materialization_timestamp(subset)
+        )
+        if not timing_subsets:
+            return subset, None
+        return subset, TimingMetadata(timestamps=timing_subsets)
 
 
 @whitelist_for_serdes
@@ -232,7 +271,7 @@ async def compute_subset(self, context: AutomationContext) -> EntitySubset:  # p
 
 @whitelist_for_serdes
 @record
-class CronTickPassedCondition(SubsetAutomationCondition):
+class CronTickPassedCondition(TimedSubsetAutomationCondition):
     cron_schedule: str
     cron_timezone: str
 
@@ -248,17 +287,21 @@ def _get_previous_cron_tick(self, effective_dt: datetime.datetime) -> datetime.d
         )
         return next(previous_ticks)
 
-    def compute_subset(self, context: AutomationContext) -> EntitySubset:
+    def compute_subset_with_timing_metadata(
+        self, context: AutomationContext
+    ) -> tuple[EntitySubset, TimingMetadata | None]:
         previous_cron_tick = self._get_previous_cron_tick(context.evaluation_time)
         if (
             # no previous evaluation
             context.previous_evaluation_time is None
             # cron tick was not newly passed
             or previous_cron_tick < context.previous_evaluation_time
         ):
-            return context.get_empty_subset()
+            return context.get_empty_subset(), None
         else:
-            return context.candidate_subset
+            candidate_subset = context.candidate_subset
+            cron_tick_ts = previous_cron_tick.timestamp()
+            return candidate_subset, TimingMetadata(timestamps={cron_tick_ts: candidate_subset})
 
 
 @whitelist_for_serdes

python_modules/dagster/dagster/_core/definitions/declarative_automation/operands/subset_automation_condition.py

@@ -2,6 +2,7 @@
 from abc import abstractmethod
 
 from dagster._core.asset_graph_view.entity_subset import EntitySubset
+from dagster._core.asset_graph_view.timing_metadata import TimingMetadata
 from dagster._core.definitions.asset_key import T_EntityKey
 from dagster._core.definitions.declarative_automation.automation_condition import (
     AutomationResult,
@@ -36,3 +37,41 @@ async def evaluate(  # pyright: ignore[reportIncompatibleMethodOverride]
             true_subset = self.compute_subset(context)
 
         return AutomationResult(context, true_subset)
+
+
+@record
+class TimedSubsetAutomationCondition(BuiltinAutomationCondition[T_EntityKey]):
+    """Base class for conditions that compute a subset along with timing metadata.
+
+    Subclasses implement compute_subset_with_timing_metadata() which returns both the subset
+    and an optional TimingMetadata. The evaluate() method stores timing metadata directly
+    on the AutomationResult for use by SinceCondition.
+    """
+
+    @property
+    def requires_cursor(self) -> bool:
+        return False
+
+    @abstractmethod
+    def compute_subset_with_timing_metadata(
+        self, context: AutomationContext[T_EntityKey]
+    ) -> tuple[EntitySubset[T_EntityKey], TimingMetadata[T_EntityKey] | None]: ...
+
+    async def evaluate(  # pyright: ignore[reportIncompatibleMethodOverride]
+        self, context: AutomationContext[T_EntityKey]
+    ) -> AutomationResult[T_EntityKey]:
+        # don't compute anything if there are no candidates
+        if context.candidate_subset.is_empty:
+            true_subset = context.get_empty_subset()
+            return AutomationResult(context, true_subset)
+
+        if inspect.iscoroutinefunction(self.compute_subset_with_timing_metadata):
+            true_subset, timing_metadata = await self.compute_subset_with_timing_metadata(context)
+        else:
+            true_subset, timing_metadata = self.compute_subset_with_timing_metadata(context)
+
+        return AutomationResult(
+            context,
+            true_subset,
+            timing_metadata=timing_metadata,
+        )

python_modules/dagster/dagster/_core/definitions/declarative_automation/operators/since_operator.py

@@ -111,19 +111,6 @@ async def evaluate(  # pyright: ignore[reportIncompatibleMethodOverride]
         # must evaluate child condition over the entire subset to avoid missing state transitions
         child_candidate_subset = context.asset_graph_view.get_full_subset(key=context.key)
 
-        from dagster._core.definitions.declarative_automation.operands import (
-            CronTickPassedCondition,
-            NewlyUpdatedCondition,
-        )
-
-        # Bit of a hack to ensure that all_deps_updated_since_cron doesn't drop new updates that
-        # happen in the same tick as the evaluation where the cron tick passes.
-
-        reset_newly_true = not (
-            isinstance(self.reset_condition, CronTickPassedCondition)
-            and isinstance(self.trigger_condition, NewlyUpdatedCondition)
-        )
-
         # compute result for trigger and reset conditions
         trigger_result, reset_result = await asyncio.gather(
             *[
@@ -143,17 +130,22 @@ async def evaluate(  # pyright: ignore[reportIncompatibleMethodOverride]
         # take the previous subset that this was true for
         true_subset = context.previous_true_subset or context.get_empty_subset()
 
-        if reset_newly_true:
-            # add in any newly true trigger asset partitions
-            true_subset = true_subset.compute_union(trigger_result.true_subset)
-            # remove any newly true reset asset partitions
-            true_subset = true_subset.compute_difference(reset_result.true_subset)
-        else:
-            # remove any newly true reset asset partitions
-            true_subset = true_subset.compute_difference(reset_result.true_subset)
+        trigger_timing = trigger_result.timing_metadata
+        reset_timing = reset_result.timing_metadata
+
+        # Step 1: Add all newly-true trigger partitions
+        true_subset = true_subset.compute_union(trigger_result.true_subset)
+
+        # Step 2: Remove all newly-true reset partitions
+        true_subset = true_subset.compute_difference(reset_result.true_subset)
 
-            # add in any newly true trigger asset partitions
-            true_subset = true_subset.compute_union(trigger_result.true_subset)
+        # Step 3: Use TimingMetadata to re-add partitions where trigger fired after reset
+        both = trigger_result.true_subset.compute_intersection(reset_result.true_subset)
+        if not both.is_empty and trigger_timing and reset_timing:
+            trigger_wins = trigger_timing.subset_with_later_timestamps_than(
+                reset_timing, empty=context.get_empty_subset()
+            ).compute_intersection(both)
+            true_subset = true_subset.compute_union(trigger_wins)
 
         # if anything changed since the previous evaluation, update the metadata
         condition_data = SinceConditionData.from_metadata(context.previous_metadata).update(
@@ -167,6 +159,7 @@ async def evaluate(  # pyright: ignore[reportIncompatibleMethodOverride]
             context=context,
             true_subset=true_subset,
             child_results=[trigger_result, reset_result],
+            timing_metadata=trigger_timing,
             metadata=condition_data.to_metadata(),
         )