From 3a0566ece2ff10b09651f03360d3d01aee5eaef8 Mon Sep 17 00:00:00 2001 From: Kyle Consalus Date: Thu, 6 Aug 2026 16:40:32 -0700 Subject: [PATCH 1/2] fix(gdd): Require that aggregator scope is a superset of their feature dependencies --- src/sentry/issues/derived/aggregators.py | 1 - src/sentry/issues/derived/framework.py | 8 +++ tests/sentry/issues/derived/test_framework.py | 68 +++++++++++++++++++ 3 files changed, 76 insertions(+), 1 deletion(-) diff --git a/src/sentry/issues/derived/aggregators.py b/src/sentry/issues/derived/aggregators.py index 8d05ad129694..ead9f23fc612 100644 --- a/src/sentry/issues/derived/aggregators.py +++ b/src/sentry/issues/derived/aggregators.py @@ -157,7 +157,6 @@ def track_root_cause(state: StateView, entry: GroupActionLogEntry) -> Aggregator @aggregator( (HAS_OPEN_FIX_PR,), - deps=(STATUS,), scope=( ResolvedInPullRequestAction, PullRequestClosedAction, diff --git a/src/sentry/issues/derived/framework.py b/src/sentry/issues/derived/framework.py index 4e20d28e0001..6548a4bb4b30 100644 --- a/src/sentry/issues/derived/framework.py +++ b/src/sentry/issues/derived/framework.py @@ -460,6 +460,14 @@ def _validate_and_sort[E: HasType]( f"Aggregator {agg.name!r} depends on {dep.name!r}, " f"which is not output by any aggregator in the pipeline" ) + producer = output_owners[dep.name] + if agg.scope is not None and ( + producer.scope is None or not set(agg.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}" + ) agg_by_name: dict[str, Aggregator[E]] = {a.name: a for a in aggregators} predecessors: dict[str, set[str]] = {a.name: set() for a in aggregators} diff --git a/tests/sentry/issues/derived/test_framework.py b/tests/sentry/issues/derived/test_framework.py index 9c825d978b14..0613573fcc6c 100644 --- a/tests/sentry/issues/derived/test_framework.py +++ b/tests/sentry/issues/derived/test_framework.py @@ -1,4 +1,5 @@ from datetime import datetime, timezone +from enum import IntEnum import pytest @@ -18,6 +19,11 @@ from sentry.issues.progress_state import IssueProgressState +class EntryType(IntEnum): + FIRST = 1 + SECOND = 2 + + def test_mutation_checking_catches_in_place_mutation() -> None: ITEMS = Feature[list[str]]("items", default_factory=list) @@ -49,6 +55,68 @@ 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_unscoped_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_unscoped_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) + + class TestDateTimeCodec: def test_json_round_trip(self) -> None: codec = DateTimeCodec() From 7ca0038fbf5b3df13e663682cd116db787a0625b Mon Sep 17 00:00:00 2001 From: Kyle Consalus Date: Thu, 6 Aug 2026 17:00:02 -0700 Subject: [PATCH 2/2] Scope.DEPS --- src/sentry/issues/derived/aggregators.py | 2 + src/sentry/issues/derived/framework.py | 99 +++++++++++++++---- tests/sentry/issues/derived/test_framework.py | 95 +++++++++++++++++- 3 files changed, 174 insertions(+), 22 deletions(-) diff --git a/src/sentry/issues/derived/aggregators.py b/src/sentry/issues/derived/aggregators.py index ead9f23fc612..0e64239345c4 100644 --- a/src/sentry/issues/derived/aggregators.py +++ b/src/sentry/issues/derived/aggregators.py @@ -43,6 +43,7 @@ from sentry.issues.derived.framework import ( Aggregator, AggregatorResult, + Scope, StateView, aggregator, emit, @@ -260,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. diff --git a/src/sentry/issues/derived/framework.py b/src/sentry/issues/derived/framework.py index 6548a4bb4b30..3263244c7bac 100644 --- a/src/sentry/issues/derived/framework.py +++ b/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,12 +323,14 @@ 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) @@ -323,6 +338,13 @@ def decorator(fn: AggregatorFn[E]) -> Aggregator[E]: 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: @@ -460,14 +490,6 @@ def _validate_and_sort[E: HasType]( f"Aggregator {agg.name!r} depends on {dep.name!r}, " f"which is not output by any aggregator in the pipeline" ) - producer = output_owners[dep.name] - if agg.scope is not None and ( - producer.scope is None or not set(agg.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}" - ) agg_by_name: dict[str, Aggregator[E]] = {a.name: a for a in aggregators} predecessors: dict[str, set[str]] = {a.name: set() for a in aggregators} @@ -496,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) diff --git a/tests/sentry/issues/derived/test_framework.py b/tests/sentry/issues/derived/test_framework.py index 0613573fcc6c..51634bf54e6b 100644 --- a/tests/sentry/issues/derived/test_framework.py +++ b/tests/sentry/issues/derived/test_framework.py @@ -11,6 +11,7 @@ Feature, OptionalCodec, Pipeline, + Scope, State, StateUpdate, StateView, @@ -22,6 +23,7 @@ class EntryType(IntEnum): FIRST = 1 SECOND = 2 + THIRD = 3 def test_mutation_checking_catches_in_place_mutation() -> None: @@ -71,7 +73,7 @@ def use_a(state: StateView, entry: object) -> AggregatorResult: Pipeline([produce_a, use_a]) -def test_scoped_aggregator_cannot_depend_on_unscoped_aggregator() -> None: +def test_scoped_aggregator_cannot_depend_on_all_scope_aggregator() -> None: A = Feature[int]("a", default=0) B = Feature[int]("b", default=0) @@ -102,7 +104,7 @@ def use_a(state: StateView, entry: object) -> AggregatorResult: assert Pipeline([produce_a, use_a]).aggregators == (produce_a, use_a) -def test_unscoped_aggregator_covers_scoped_dependency() -> None: +def test_all_scope_aggregator_covers_scoped_dependency() -> None: A = Feature[int]("a", default=0) B = Feature[int]("b", default=0) @@ -117,6 +119,95 @@ def use_a(state: StateView, entry: object) -> AggregatorResult: 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()