fix(seer): attribute remaining grouping cleanup paths (#128247)

## Summary

- carry organization and project attribution through per-project
retention cleanup workers
- restore the user actor already recorded by project and organization
deletion entry points
- preserve the explicit Taskworker scheduler system context for
actorless scheduled deletions
- establish explicit system context for manual `sentry deletions run`
executions
- reuse the scheduled-deletion context boundary so manual runs restore
recorded users and target scope
- derive organization and project from the scheduled-deletion target
- enrich Seer hash-deletion dispatch with the resolved organization and
project
- never create a system fallback at the deletion or Seer call site

The Taskworker scheduler and manual deletion CLI establish system
context at their entrypoints. Scheduled deletions with a recorded actor
override that context with the user; actorless deletions retain the
entrypoint context and add their organization/project attribution.

## Testing

- focused CLI, scheduled-deletion, and cleanup context suite: 70 passed
- repository formatting and lint hooks
- pre-push mypy checks

Author: sehr-m

src/sentry/deletions/tasks/scheduled.py

@@ -1,4 +1,6 @@
 import logging
+from contextlib import AbstractContextManager, nullcontext
+from dataclasses import replace
 from datetime import timedelta
 from typing import Any
 
@@ -16,18 +18,60 @@
     ScheduledDeletion,
 )
 from sentry.exceptions import DeleteAborted
+from sentry.models.organization import Organization
+from sentry.models.project import Project
 from sentry.signals import pending_delete
 from sentry.silo.base import SiloMode
 from sentry.tasks.base import instrumented_task
 from sentry.taskworker.namespaces import deletion_control_tasks, deletion_tasks
 from sentry.utils.env import in_test_environment
+from sentry.viewer_context import (
+    ActorType,
+    ViewerContext,
+    get_viewer_context,
+    viewer_context_scope,
+)
 
 logger = logging.getLogger("sentry.deletions.api")
 
 
 MAX_RETRIES = 5
 
 
+def scheduled_deletion_viewer_context(
+    deletion: BaseScheduledDeletion,
+    instance: Any,
+    *,
+    origin_context: ViewerContext | None = None,
+) -> AbstractContextManager[None]:
+    """Apply the recorded actor and target scope while executing a scheduled deletion."""
+    if not isinstance(instance, (Organization, Project)):
+        return nullcontext()
+
+    viewer_context = (
+        ViewerContext(user_id=deletion.actor_id, actor_type=ActorType.USER)
+        if deletion.actor_id is not None
+        else origin_context or get_viewer_context()
+    )
+    if viewer_context is None:
+        return nullcontext()
+
+    if isinstance(instance, Project):
+        viewer_context = replace(
+            viewer_context,
+            organization_id=instance.organization_id,
+            project_id=instance.id,
+        )
+    else:
+        viewer_context = replace(
+            viewer_context,
+            organization_id=instance.id,
+            project_id=None,
+        )
+
+    return viewer_context_scope(viewer_context)
+
+
 @instrumented_task(
     name="sentry.deletions.tasks.reattempt_deletions_control",
     namespace=deletion_control_tasks,
@@ -201,24 +245,25 @@ def _run_deletion(
         actor = deletion.get_actor()
         pending_delete.send(sender=type(instance), instance=instance, actor=actor)
 
-    try:
-        has_more = task.chunk()
-        if has_more:
-            process_task.delay(deletion_id=deletion_id, first_pass=False)
-        else:
-            deletion.delete()
-    except Exception as err:
-        sentry_sdk.set_context(
-            "deletion",
-            {
-                "id": deletion.id,
-                "model": deletion.model_name,
-                "object_id": deletion.object_id,
-            },
-        )
-        sentry_sdk.set_attribute("deletion.id", deletion.id)
-        sentry_sdk.set_attribute("deletion.model", deletion.model_name)
-        sentry_sdk.set_attribute("deletion.object_id", deletion.object_id)
-        sentry_sdk.capture_exception(err)
-        if in_test_environment():
-            raise
+    with scheduled_deletion_viewer_context(deletion, instance):
+        try:
+            has_more = task.chunk()
+            if has_more:
+                process_task.delay(deletion_id=deletion_id, first_pass=False)
+            else:
+                deletion.delete()
+        except Exception as err:
+            sentry_sdk.set_context(
+                "deletion",
+                {
+                    "id": deletion.id,
+                    "model": deletion.model_name,
+                    "object_id": deletion.object_id,
+                },
+            )
+            sentry_sdk.set_attribute("deletion.id", deletion.id)
+            sentry_sdk.set_attribute("deletion.model", deletion.model_name)
+            sentry_sdk.set_attribute("deletion.object_id", deletion.object_id)
+            sentry_sdk.capture_exception(err)
+            if in_test_environment():
+                raise

src/sentry/runner/commands/cleanup.py

@@ -5,6 +5,7 @@
 import os
 import time
 from collections.abc import Callable, Mapping, Sequence
+from contextlib import nullcontext
 from datetime import timedelta
 from multiprocessing import JoinableQueue as Queue
 from multiprocessing import Process
@@ -22,6 +23,7 @@
 
 from sentry.runner.decorators import log_options
 from sentry.silo.base import SiloLimit, SiloMode
+from sentry.viewer_context import ActorType, ViewerContext, viewer_context_scope
 
 logger = logging.getLogger(__name__)
 
@@ -76,6 +78,7 @@ def get_organization(value: str) -> int | None:
     "tuple[str, tuple[int, ...]]"
     " | tuple[str, tuple[int, ...], int | None]"
     " | tuple[str, tuple[int, ...], int | None, dict[str, Any]]"
+    " | tuple[str, tuple[int, ...], int | None, dict[str, Any], int]"
 )
 _WorkQueue: TypeAlias = "Queue[Literal['91650ec271ae4b3e8a67cdc909d80f8c'] | _WorkItem]"
 
@@ -118,16 +121,23 @@ def multiprocess_worker(task_queue: _WorkQueue) -> None:
             task_queue.task_done()
             return
 
-        # Handle all formats: (model_name, chunk), (model_name, chunk, project_id) and (model_name, chunk, project_id, deferred_filter)
+        # Handle all formats: (model_name, chunk), (model_name, chunk, project_id),
+        # (model_name, chunk, project_id, deferred_filter), and the project-scoped
+        # form with organization_id appended.
         if len(j) == 2:
             model_name, chunk = j
             project_id = None
             deferred_filter: dict[str, Any] = {}
+            organization_id = None
         elif len(j) == 3:
             model_name, chunk, project_id = j
             deferred_filter = {}
-        else:
+            organization_id = None
+        elif len(j) == 4:
             model_name, chunk, project_id, deferred_filter = j
+            organization_id = None
+        else:
+            model_name, chunk, project_id, deferred_filter, organization_id = j
 
         if options.get("cleanup.abort_execution"):
             logger.warning("Cleanup worker aborting due to cleanup.abort_execution flag")
@@ -154,7 +164,9 @@ def multiprocess_worker(task_queue: _WorkQueue) -> None:
                 propagation_context.custom_sampling_context = prev_sampling_context
 
             with span:
-                task_execution(model_name, chunk, project_id, deferred_filter)
+                task_execution(
+                    model_name, chunk, project_id, deferred_filter, organization_id=organization_id
+                )
         except Exception:
             metrics.incr(
                 "cleanup.error",
@@ -188,6 +200,7 @@ def task_execution(
     chunk: tuple[int, ...],
     project_id: int | None,
     deferred_filter: Mapping[str, Any] | None = None,
+    organization_id: int | None = None,
 ) -> None:
     """
     Execute deletion for a chunk of objects.
@@ -224,25 +237,38 @@ def task_execution(
     if deferred_filter:
         query.update(deferred_filter)
 
-    task = deletions.get(
-        model=model,
-        query=query,
-        skip_models=skip_child_relations_models,
-        transaction_id=uuid4().hex,
+    context = (
+        viewer_context_scope(
+            ViewerContext(
+                organization_id=organization_id,
+                project_id=project_id,
+                actor_type=ActorType.SYSTEM,
+            )
+        )
+        if organization_id is not None and project_id is not None
+        else nullcontext()
     )
 
-    # Track deletion iterations (task.chunk() may need multiple calls to fully delete)
-    iteration = 0
-    while True:
-        iteration += 1
-        has_more = task.chunk(apply_filter=True)
-
-        debug_output(
-            f"[ITERATION {iteration}] {model_name} chunk (project_id={project_id} has_more={has_more})"
+    with context:
+        task = deletions.get(
+            model=model,
+            query=query,
+            skip_models=skip_child_relations_models,
+            transaction_id=uuid4().hex,
         )
 
-        if not has_more:
-            break
+        # Track deletion iterations (task.chunk() may need multiple calls to fully delete)
+        iteration = 0
+        while True:
+            iteration += 1
+            has_more = task.chunk(apply_filter=True)
+
+            debug_output(
+                f"[ITERATION {iteration}] {model_name} chunk (project_id={project_id} has_more={has_more})"
+            )
+
+            if not has_more:
+                break
 
     # Increment metric once per chunk, after all iterations complete
     metrics.incr("cleanup.chunk_processed", tags={"model": model_name}, amount=chunk_size)
@@ -903,6 +929,7 @@ def _schedule_bulk_delete_chunks(
     q: BulkDeleteQuery,  # Imported locally in functions that use it
     model_tp: type[BaseModel],
     project_id: int | None,
+    organization_id: int | None = None,
     context_str: str = "",
 ) -> tuple[int, int]:
     """
@@ -939,7 +966,10 @@ def _schedule_bulk_delete_chunks(
                 },
             )
             _last_progress_log[imp] = time.monotonic()
-        task_queue.put((imp, chunk, project_id, q.deferred_filter))
+        if organization_id is None:
+            task_queue.put((imp, chunk, project_id, q.deferred_filter))
+        else:
+            task_queue.put((imp, chunk, project_id, q.deferred_filter, organization_id))
         chunk_count += 1
         total_objects += len(chunk)
 
@@ -1027,9 +1057,9 @@ def run_bulk_deletes_by_project(
     if project_deletion_query is not None and len(to_delete_by_project):
         debug_output("Running bulk deletes in DELETES_BY_PROJECT")
 
-        for project_id_for_deletion in RangeQuerySetWrapper(
-            project_deletion_query.values_list("id", flat=True),
-            result_value_getter=lambda item: item,
+        for project_id_for_deletion, organization_id_for_deletion in RangeQuerySetWrapper(
+            project_deletion_query.values_list("id", "organization_id"),
+            result_value_getter=lambda item: item[0],
         ):
             for model_tp, dtfield, order_by in to_delete_by_project:
                 models_attempted.add(model_tp.__name__.lower())
@@ -1046,7 +1076,13 @@ def run_bulk_deletes_by_project(
                         order_by=order_by,
                     )
 
-                    _schedule_bulk_delete_chunks(task_queue, q, model_tp, project_id_for_deletion)
+                    _schedule_bulk_delete_chunks(
+                        task_queue,
+                        q,
+                        model_tp,
+                        project_id_for_deletion,
+                        organization_id_for_deletion,
+                    )
                 except Exception:
                     capture_exception(
                         tags={"model": model_tp.__name__, "project_id": project_id_for_deletion}

src/sentry/runner/commands/deletions.py

@@ -5,6 +5,7 @@
 import click
 
 from sentry.runner.decorators import configuration
+from sentry.viewer_context import ActorType, ViewerContext
 
 if TYPE_CHECKING:
     from sentry.deletions.models.scheduleddeletion import BaseScheduledDeletion
@@ -146,6 +147,7 @@ def _run_one(*, deletion: BaseScheduledDeletion, verbose: bool = False) -> None:
     from django.core.exceptions import ObjectDoesNotExist
 
     from sentry import deletions as deletions_module
+    from sentry.deletions.tasks.scheduled import scheduled_deletion_viewer_context
     from sentry.signals import pending_delete
 
     click.echo(f"Running deletion {deletion.id} ({deletion.model_name} #{deletion.object_id})...")
@@ -156,32 +158,37 @@ def _run_one(*, deletion: BaseScheduledDeletion, verbose: bool = False) -> None:
         deletion.delete()
         return
 
-    task = deletions_module.get(
-        model=deletion.get_model(),
-        query={"id": deletion.object_id},
-        transaction_id=deletion.guid,
-        actor_id=deletion.actor_id,
-    )
-
-    if not task.should_proceed(instance):
-        click.echo("  Deletion aborted (should_proceed returned False).")
-        deletion.delete()
-        return
+    with scheduled_deletion_viewer_context(
+        deletion,
+        instance,
+        origin_context=ViewerContext(actor_type=ActorType.SYSTEM),
+    ):
+        task = deletions_module.get(
+            model=deletion.get_model(),
+            query={"id": deletion.object_id},
+            transaction_id=deletion.guid,
+            actor_id=deletion.actor_id,
+        )
 
-    actor = deletion.get_actor()
-    pending_delete.send(sender=type(instance), instance=instance, actor=actor)
+        if not task.should_proceed(instance):
+            click.echo("  Deletion aborted (should_proceed returned False).")
+            deletion.delete()
+            return
 
-    try:
-        has_more = True
-        while has_more:
-            has_more = task.chunk()
-        deletion.delete()
-        click.echo("  Done.")
-    except Exception as e:
-        if verbose:
-            import traceback
+        actor = deletion.get_actor()
+        pending_delete.send(sender=type(instance), instance=instance, actor=actor)
 
-            click.echo(f"  Failed:\n{traceback.format_exc()}", err=True)
-        else:
-            click.echo(f"  Failed: {e}", err=True)
-        click.echo("  Deletion record preserved for retry. Re-run to continue.")
+        try:
+            has_more = True
+            while has_more:
+                has_more = task.chunk()
+            deletion.delete()
+            click.echo("  Done.")
+        except Exception as e:
+            if verbose:
+                import traceback
+
+                click.echo(f"  Failed:\n{traceback.format_exc()}", err=True)
+            else:
+                click.echo(f"  Failed: {e}", err=True)
+            click.echo("  Deletion record preserved for retry. Re-run to continue.")