fix(gdd): Require that aggregator scope is a superset of their feature dependencies

src/sentry/issues/derived/aggregators.py

@@ -43,6 +43,7 @@
 from sentry.issues.derived.framework import (
     Aggregator,
     AggregatorResult,
+    Scope,
     StateView,
     aggregator,
     emit,
@@ -157,7 +158,6 @@ def track_root_cause(state: StateView, entry: GroupActionLogEntry) -> Aggregator
 
 @aggregator(
     (HAS_OPEN_FIX_PR,),
-    deps=(STATUS,),
     scope=(
         ResolvedInPullRequestAction,
         PullRequestClosedAction,
@@ -261,6 +261,7 @@ def track_last_completed_autofix_step(
 @aggregator(
     (BLOCKER,),
     deps=(STATUS, HAS_OPEN_FIX_PR, LAST_COMPLETED_AUTOFIX_STEP),
+    scope=Scope.DEPS,
 )
 def track_blocker(state: StateView, entry: GroupActionLogEntry) -> AggregatorResult:
     """Track the human action blocking the issue's progress toward resolution.

src/sentry/issues/derived/framework.py

@@ -10,7 +10,7 @@
 from dataclasses import dataclass
 from datetime import datetime
 from enum import IntEnum, StrEnum
-from typing import Any, ClassVar, Final, Protocol, runtime_checkable
+from typing import Any, ClassVar, Final, Literal, Protocol, runtime_checkable
 
 _MISSING = object()
 
@@ -281,6 +281,19 @@ def emit(*entries: FeatureEntry) -> AggregatorResult:
     return StateUpdate(dict(entries))
 
 
+class Scope(StrEnum):
+    """Special aggregator scopes.
+
+    ALL runs for every entry. DEPS runs whenever any dependency producer runs.
+    """
+
+    ALL = "all"
+    DEPS = "deps"
+
+
+type EffectiveScope = Literal[Scope.ALL] | frozenset[int]
+
+
 @dataclass(frozen=True)
 class Aggregator[E: HasType]:
     """A named function that reads from dep features and writes to output features."""
@@ -289,7 +302,7 @@ class Aggregator[E: HasType]:
     deps: tuple[Feature[Any], ...]
     outputs: tuple[Feature[Any], ...]
     fn: AggregatorFn[E]
-    scope: tuple[int, ...] | None = None
+    scope: Scope | tuple[int, ...] = Scope.ALL
 
 
 class _HasGetType(Protocol):
@@ -310,19 +323,28 @@ def aggregator[E: HasType](
     outputs: tuple[Feature[Any], ...],
     *,
     deps: tuple[Feature[Any], ...] = (),
-    scope: tuple[ScopeItem, ...] | None = None,
+    scope: Scope | tuple[ScopeItem, ...] = Scope.ALL,
 ) -> Callable[[AggregatorFn[E]], Aggregator[E]]:
-    """Decorator to create an Aggregator. `scope` accepts enum members or classes with get_type()."""
+    """Create an aggregator with a fixed, universal, or dependency-derived scope."""
     if not outputs:
         raise ValueError("aggregator must declare at least one output")
-    raw_scope = tuple(_scope_int(s) for s in scope) if scope is not None else None
+    if scope is Scope.DEPS and not deps:
+        raise ValueError("Scope.DEPS requires at least one dependency")
+    raw_scope = scope if isinstance(scope, Scope) else tuple(_scope_int(s) for s in scope)
 
     def decorator(fn: AggregatorFn[E]) -> Aggregator[E]:
         return Aggregator(name=fn.__name__, deps=deps, outputs=outputs, fn=fn, scope=raw_scope)
 
     return decorator
 
 
+@dataclass(frozen=True)
+class _ValidatedPipeline[E: HasType]:
+    aggregators: tuple[Aggregator[E], ...]
+    features: tuple[Feature[Any], ...]
+    effective_scopes: dict[str, EffectiveScope]
+
+
 # ---------------------------------------------------------------------------
 # Pipeline
 # ---------------------------------------------------------------------------
@@ -344,9 +366,16 @@ def __init__(
     ) -> None:
         self._check_mutations = check_mutations
         aggregators = tuple(aggregators)
-        self._aggregators, self._features = _validate_and_sort(aggregators)
+        validated = _validate_and_sort(aggregators)
+        self._aggregators = validated.aggregators
+        self._features = validated.features
         self._steps = tuple(
-            (agg, frozenset({*agg.deps, *agg.outputs}), frozenset(agg.outputs))
+            (
+                agg,
+                frozenset({*agg.deps, *agg.outputs}),
+                frozenset(agg.outputs),
+                validated.effective_scopes[agg.name],
+            )
             for agg in self._aggregators
         )
         payload = f"{self._version}:" + ",".join(sorted(f.content_id for f in self._features))
@@ -371,9 +400,10 @@ def initial_state(self) -> State:
 
     def step(self, state: State, entry: E) -> State:
         entry_type = entry.type
-        for agg, view_fields, output_fields in self._steps:
-            if agg.scope is not None and entry_type not in agg.scope:
-                continue
+        for agg, view_fields, output_fields, effective_scope in self._steps:
+            if effective_scope is not Scope.ALL:
+                if entry_type not in effective_scope:
+                    continue
             subset = state.view(view_fields)
             snapshot = copy.deepcopy(subset._data) if self._check_mutations else None
             result = agg.fn(subset, entry)
@@ -442,7 +472,7 @@ def _ensure_no_aliasing(features: Iterable[Feature[Any]]) -> tuple[Feature[Any],
 
 def _validate_and_sort[E: HasType](
     aggregators: tuple[Aggregator[E], ...],
-) -> tuple[tuple[Aggregator[E], ...], tuple[Feature[Any], ...]]:
+) -> _ValidatedPipeline[E]:
     output_owners: dict[str, Aggregator[E]] = {}
     for agg in aggregators:
         for feature in agg.outputs:
@@ -488,6 +518,43 @@ def _validate_and_sort[E: HasType](
         remaining = {a.name for a in aggregators} - {a.name for a in order}
         raise ValueError(f"Cycle detected among aggregators: {remaining}")
 
+    effective_scopes: dict[str, EffectiveScope] = {}
+    for agg in order:
+        if agg.scope is Scope.DEPS:
+            if not agg.deps:
+                raise ValueError(f"Aggregator {agg.name!r} uses Scope.DEPS without dependencies")
+            scope_values: set[int] = set()
+            for dep in agg.deps:
+                producer = output_owners[dep.name]
+                if producer.name == agg.name:
+                    raise ValueError(
+                        f"Aggregator {agg.name!r} cannot derive its scope from its own output"
+                    )
+                producer_scope = effective_scopes[producer.name]
+                if producer_scope is Scope.ALL:
+                    effective_scopes[agg.name] = Scope.ALL
+                    break
+                scope_values.update(producer_scope)
+            else:
+                effective_scopes[agg.name] = frozenset(scope_values)
+        elif agg.scope is Scope.ALL:
+            effective_scopes[agg.name] = Scope.ALL
+        else:
+            effective_scopes[agg.name] = frozenset(agg.scope)
+
+    for agg in aggregators:
+        scope = effective_scopes[agg.name]
+        for dep in agg.deps:
+            producer = output_owners[dep.name]
+            producer_scope = effective_scopes[producer.name]
+            if scope is Scope.ALL:
+                continue
+            if producer_scope is Scope.ALL or not scope.issuperset(producer_scope):
+                raise ValueError(
+                    f"Aggregator {agg.name!r} has a scope that does not cover "
+                    f"dependency {dep.name!r} produced by {producer.name!r}"
+                )
+
     all_features = _ensure_no_aliasing(f for agg in aggregators for f in (*agg.deps, *agg.outputs))
 
-    return tuple(order), all_features
+    return _ValidatedPipeline(tuple(order), all_features, effective_scopes)

tests/sentry/issues/derived/test_framework.py

@@ -1,4 +1,5 @@
 from datetime import datetime, timezone
+from enum import IntEnum
 
 import pytest
 
@@ -10,6 +11,7 @@
     Feature,
     OptionalCodec,
     Pipeline,
+    Scope,
     State,
     StateUpdate,
     StateView,
@@ -18,6 +20,12 @@
 from sentry.issues.progress_state import IssueProgressState
 
 
+class EntryType(IntEnum):
+    FIRST = 1
+    SECOND = 2
+    THIRD = 3
+
+
 def test_mutation_checking_catches_in_place_mutation() -> None:
     ITEMS = Feature[list[str]]("items", default_factory=list)
 
@@ -49,6 +57,157 @@ def test_state_updated_tracks_merged_features() -> None:
     assert state[B] == 0
 
 
+def test_dependency_scope_must_cover_producer_scope() -> None:
+    A = Feature[int]("a", default=0)
+    B = Feature[int]("b", default=0)
+
+    @aggregator((A,), scope=(EntryType.FIRST, EntryType.SECOND))
+    def produce_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    @aggregator((B,), deps=(A,), scope=(EntryType.FIRST,))
+    def use_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    with pytest.raises(ValueError, match="scope that does not cover dependency 'a'"):
+        Pipeline([produce_a, use_a])
+
+
+def test_scoped_aggregator_cannot_depend_on_all_scope_aggregator() -> None:
+    A = Feature[int]("a", default=0)
+    B = Feature[int]("b", default=0)
+
+    @aggregator((A,))
+    def produce_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    @aggregator((B,), deps=(A,), scope=(EntryType.FIRST,))
+    def use_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    with pytest.raises(ValueError, match="scope that does not cover dependency 'a'"):
+        Pipeline([produce_a, use_a])
+
+
+def test_dependency_scope_can_be_a_superset_of_producer_scope() -> None:
+    A = Feature[int]("a", default=0)
+    B = Feature[int]("b", default=0)
+
+    @aggregator((A,), scope=(EntryType.FIRST,))
+    def produce_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    @aggregator((B,), deps=(A,), scope=(EntryType.FIRST, EntryType.SECOND))
+    def use_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    assert Pipeline([produce_a, use_a]).aggregators == (produce_a, use_a)
+
+
+def test_all_scope_aggregator_covers_scoped_dependency() -> None:
+    A = Feature[int]("a", default=0)
+    B = Feature[int]("b", default=0)
+
+    @aggregator((A,), scope=(EntryType.FIRST,))
+    def produce_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    @aggregator((B,), deps=(A,))
+    def use_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    assert Pipeline([produce_a, use_a]).aggregators == (produce_a, use_a)
+
+
+def test_default_scope_is_all() -> None:
+    A = Feature[int]("a", default=0)
+
+    @aggregator((A,))
+    def produce_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    assert produce_a.scope is Scope.ALL
+
+
+def test_deps_scope_runs_for_union_of_dependency_scopes() -> None:
+    A = Feature[int]("a", default=0)
+    B = Feature[int]("b", default=0)
+    C = Feature[int]("c", default=0)
+
+    @aggregator((A,), scope=(EntryType.FIRST,))
+    def produce_a(state: StateView, entry: object) -> AggregatorResult:
+        return StateUpdate({A: state[A] + 1})
+
+    @aggregator((B,), scope=(EntryType.SECOND,))
+    def produce_b(state: StateView, entry: object) -> AggregatorResult:
+        return StateUpdate({B: state[B] + 1})
+
+    @aggregator((C,), deps=(A, B), scope=Scope.DEPS)
+    def use_deps(state: StateView, entry: object) -> AggregatorResult:
+        return StateUpdate({C: state[C] + 1})
+
+    class Entry:
+        def __init__(self, type: EntryType) -> None:
+            self.type = type
+
+    state = Pipeline([produce_a, produce_b, use_deps]).run(
+        [Entry(EntryType.FIRST), Entry(EntryType.SECOND), Entry(EntryType.THIRD)]
+    )
+
+    assert state[A] == 1
+    assert state[B] == 1
+    assert state[C] == 2
+
+
+def test_deps_scope_is_resolved_transitively() -> None:
+    A = Feature[int]("a", default=0)
+    B = Feature[int]("b", default=0)
+    C = Feature[int]("c", default=0)
+
+    @aggregator((A,), scope=(EntryType.FIRST,))
+    def produce_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    @aggregator((B,), deps=(A,), scope=Scope.DEPS)
+    def use_a(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    @aggregator((C,), deps=(B,), scope=(EntryType.SECOND,))
+    def use_b(state: StateView, entry: object) -> AggregatorResult:
+        return None
+
+    with pytest.raises(ValueError, match="scope that does not cover dependency 'b'"):
+        Pipeline([produce_a, use_a, use_b])
+
+
+def test_deps_scope_resolves_to_all_when_dependency_scope_is_all() -> None:
+    A = Feature[int]("a", default=0)
+    B = Feature[int]("b", default=0)
+
+    @aggregator((A,))
+    def produce_a(state: StateView, entry: object) -> AggregatorResult:
+        return StateUpdate({A: state[A] + 1})
+
+    @aggregator((B,), deps=(A,), scope=Scope.DEPS)
+    def use_a(state: StateView, entry: object) -> AggregatorResult:
+        return StateUpdate({B: state[B] + 1})
+
+    class Entry:
+        type = EntryType.THIRD
+
+    state = Pipeline([produce_a, use_a]).run([Entry()])
+
+    assert state[A] == 1
+    assert state[B] == 1
+
+
+def test_deps_scope_requires_dependency() -> None:
+    A = Feature[int]("a", default=0)
+
+    with pytest.raises(ValueError, match="requires at least one dependency"):
+        aggregator((A,), scope=Scope.DEPS)
+
+
 class TestDateTimeCodec:
     def test_json_round_trip(self) -> None:
         codec = DateTimeCodec()