From a39c4d1f427932ef03b44e581530e4bccd443098 Mon Sep 17 00:00:00 2001 From: Trevor Elkins Date: Wed, 11 Nov 2026 18:07:43 -0500 Subject: [PATCH] ref(seer): Remove the monitor cleanup automation Monitor cleanup was a test automation. Remove it with its result delivery hook, the POST on the workflow runs endpoint that only started it, and the older run helpers in workflows/runs.py that only it called. --- src/sentry/seer/agent/feature_delivery.py | 2 - .../organization_seer_workflow_runs.py | 37 ---- src/sentry/seer/monitor_cleanup/__init__.py | 0 src/sentry/seer/monitor_cleanup/results.py | 112 ----------- src/sentry/seer/monitor_cleanup/runs.py | 86 --------- src/sentry/seer/monitor_cleanup/schemas.py | 122 ------------ src/sentry/seer/workflows/runs.py | 147 +------------- src/sentry/seer/workflows/schemas.py | 21 +- .../test_organization_seer_workflow_runs.py | 179 ------------------ tests/sentry/seer/workflows/test_runs.py | 161 +--------------- 10 files changed, 4 insertions(+), 863 deletions(-) delete mode 100644 src/sentry/seer/monitor_cleanup/__init__.py delete mode 100644 src/sentry/seer/monitor_cleanup/results.py delete mode 100644 src/sentry/seer/monitor_cleanup/runs.py delete mode 100644 src/sentry/seer/monitor_cleanup/schemas.py diff --git a/src/sentry/seer/agent/feature_delivery.py b/src/sentry/seer/agent/feature_delivery.py index 013094250723..107ea3354bb4 100644 --- a/src/sentry/seer/agent/feature_delivery.py +++ b/src/sentry/seer/agent/feature_delivery.py @@ -13,7 +13,6 @@ from sentry.seer.autofix.feature.models import ( SINGLE_RUN_FEATURE_ID as AUTOFIX_SINGLE_RUN_FEATURE_ID, ) -from sentry.seer.monitor_cleanup.runs import deliver_monitor_cleanup_result from sentry.seer.smart_assignment.delivery import deliver_smart_assignment_result __all__ = ["DELIVERY_HANDLERS", "FeatureDeliveryFn", "FeatureRunStatus"] @@ -32,7 +31,6 @@ def __call__( DELIVERY_HANDLERS: dict[str, FeatureDeliveryFn] = { - "monitor_cleanup": deliver_monitor_cleanup_result, "night_shift": deliver_agentic_triage_result, "smart_assignment": deliver_smart_assignment_result, AUTOFIX_FEATURE_ID: deliver_autofix_feature_result, diff --git a/src/sentry/seer/endpoints/organization_seer_workflow_runs.py b/src/sentry/seer/endpoints/organization_seer_workflow_runs.py index b617b8be765d..8136f889f943 100644 --- a/src/sentry/seer/endpoints/organization_seer_workflow_runs.py +++ b/src/sentry/seer/endpoints/organization_seer_workflow_runs.py @@ -25,7 +25,6 @@ from sentry.db.models.fields.bounded import I64_MAX from sentry.models.organization import Organization from sentry.notebooks.access import has_notebooks_access -from sentry.ratelimits.config import RateLimitConfig from sentry.seer.models.workflow import ( SEER_CONFIGURABLE_WORKFLOW_STRATEGIES, SeerWorkflowExecutionStatus, @@ -35,24 +34,14 @@ SeerWorkflowRunStatus, SeerWorkflowStrategy, ) -from sentry.seer.monitor_cleanup.runs import create_monitor_cleanup_run from sentry.seer.workflows.definitions import is_scheduled from sentry.seer.workflows.metric_alert import MetricBreachInputs from sentry.seer_workflow_outputs.models import SeerWorkflowOutputLink -from sentry.types.ratelimit import RateLimit, RateLimitCategory from sentry.utils.safe import get_path logger = logging.getLogger(__name__) -class WorkflowRunCreateSerializer(serializers.Serializer): - strategy = serializers.ChoiceField(choices=[SeerWorkflowStrategy.DUPLICATE_MONITORS]) - - -class WorkflowRunCreateResponse(TypedDict): - runId: str - - class WorkflowRunListSerializer(serializers.Serializer): config = serializers.IntegerField( required=False, @@ -166,7 +155,6 @@ class SeerWorkflowRunsResponse(TypedDict): class OrganizationSeerWorkflowRunsPermission(OrganizationPermission): scope_map = { "GET": ["org:read"], - "POST": ["org:read"], } @@ -175,18 +163,9 @@ class OrganizationSeerWorkflowRunsPermission(OrganizationPermission): class OrganizationSeerWorkflowRunsEndpoint(OrganizationEndpoint): publish_status = { "GET": ApiPublishStatus.PRIVATE, - "POST": ApiPublishStatus.PRIVATE, } owner = ApiOwner.ML_AI permission_classes = (OrganizationSeerWorkflowRunsPermission,) - enforce_rate_limit = True - rate_limits = RateLimitConfig( - limit_overrides={ - "POST": { - RateLimitCategory.USER: RateLimit(limit=1, window=60), - }, - } - ) @extend_schema( operation_id="List Seer workflow runs", @@ -262,22 +241,6 @@ def get(self, request: Request, organization: Organization) -> Response: ), ) - @extend_schema( - operation_id="Start a Seer workflow run", - request=WorkflowRunCreateSerializer, - responses={ - 202: inline_sentry_response_serializer( - "SeerWorkflowRunCreate", WorkflowRunCreateResponse - ), - }, - ) - def post(self, request: Request, organization: Organization) -> Response: - serializer = WorkflowRunCreateSerializer(data=request.data) - if not serializer.is_valid(): - return Response({"detail": serializer.errors}, status=400) - run = create_monitor_cleanup_run(request, organization, source="manual") - return Response({"runId": str(run.id)}, status=202) - def serialize_workflow_runs( runs: Sequence[SeerWorkflowRun], *, include_investigations: bool diff --git a/src/sentry/seer/monitor_cleanup/__init__.py b/src/sentry/seer/monitor_cleanup/__init__.py deleted file mode 100644 index e69de29bb2d1..000000000000 diff --git a/src/sentry/seer/monitor_cleanup/results.py b/src/sentry/seer/monitor_cleanup/results.py deleted file mode 100644 index 40842efe0377..000000000000 --- a/src/sentry/seer/monitor_cleanup/results.py +++ /dev/null @@ -1,112 +0,0 @@ -from collections.abc import Mapping - -from sentry.auth import access -from sentry.incidents.grouptype import MetricIssue -from sentry.models.organization import Organization -from sentry.models.project import Project -from sentry.seer.monitor_cleanup.schemas import ( - MonitorCleanupComparison, - MonitorCleanupComparisonValue, - MonitorCleanupFinding, - MonitorCleanupOutput, - MonitorCleanupResource, - SeerMonitorCleanupArtifact, - SeerMonitorFinding, - SeerOrganizationMonitorCleanupArtifact, -) -from sentry.seer.workflows.schemas import WorkflowResultError -from sentry.users.services.user.service import user_service -from sentry.workflow_engine.models import Detector, DetectorWorkflow, Workflow - - -def parse_monitor_cleanup_results( - artifact: SeerOrganizationMonitorCleanupArtifact, organization: Organization, user_id: int -) -> list[MonitorCleanupOutput]: - """Parse Seer findings into stored results, resolving resource IDs with read-only lookups.""" - project_ids = [project.project_id for project in artifact.projects] - projects = { - project.id: project - for project in Project.objects.filter(organization=organization, id__in=project_ids) - } - user = user_service.get_user(user_id=user_id) - if user is None: - raise WorkflowResultError("The triggering user no longer exists.") - user_access = access.from_user(user, organization) - if projects.keys() != set(project_ids) or not user_access.has_projects_access( - projects.values() - ): - raise WorkflowResultError("Some scanned projects are no longer accessible.") - outputs: list[MonitorCleanupOutput] = [] - for project_artifact in artifact.projects: - project = projects[project_artifact.project_id] - output = parse_project_monitor_cleanup_result(project_artifact, organization.id, project.id) - output["projectSlug"] = project.slug - outputs.append(output) - return outputs - - -def parse_project_monitor_cleanup_result( - artifact: SeerMonitorCleanupArtifact, organization_id: int, project_id: int -) -> MonitorCleanupOutput: - findings = artifact.findings - ids = {monitor_id for finding in findings for monitor_id in finding.monitor_ids} - alert_ids = {alert_id for finding in findings for alert_id in finding.alert_ids} - monitors: dict[int, MonitorCleanupResource] = { - detector.id: _serialize_resource(detector) - for detector in Detector.objects.filter( - id__in=ids, - project_id=project_id, - project__organization_id=organization_id, - type=MetricIssue.slug, - ) - } - alerts: dict[int, MonitorCleanupResource] = { - link.workflow_id: _serialize_resource(link.workflow) - for link in DetectorWorkflow.objects.filter( - detector_id__in=monitors, - workflow_id__in=alert_ids, - workflow__organization_id=organization_id, - workflow__type=Workflow.Type.ALERT, - ).select_related("workflow") - } - return { - "outputKind": "monitor_cleanup", - "schemaVersion": 1, - "projectId": str(project_id), - "scan": {"status": artifact.scan_status, "monitorsScanned": artifact.monitors_scanned}, - "summary": artifact.summary, - "findings": [ - _serialize_finding(finding, monitors, alerts) - for finding in findings - if all(monitor_id in monitors for monitor_id in finding.monitor_ids) - and all(alert_id in alerts for alert_id in finding.alert_ids) - ], - } - - -def _serialize_resource(resource: Detector | Workflow) -> MonitorCleanupResource: - return {"id": str(resource.id), "name": resource.name, "enabled": resource.enabled} - - -def _serialize_finding( - finding: SeerMonitorFinding, - monitors: Mapping[int, MonitorCleanupResource], - alerts: Mapping[int, MonitorCleanupResource], -) -> MonitorCleanupFinding: - comparisons: list[MonitorCleanupComparison] = [] - for row in finding.comparison: - values: list[MonitorCleanupComparisonValue] = [ - {"monitorId": str(value.monitor_id), "value": value.value} for value in row.values - ] - comparisons.append({"property": row.property, "values": values}) - - return { - "kind": finding.kind, - "monitors": [monitors[monitor_id] for monitor_id in finding.monitor_ids], - "suggestedKeepId": str(finding.suggested_keep_id) - if finding.suggested_keep_id is not None - else None, - "alerts": [alerts[alert_id] for alert_id in dict.fromkeys(finding.alert_ids)], - "reason": finding.reason, - "comparison": comparisons, - } diff --git a/src/sentry/seer/monitor_cleanup/runs.py b/src/sentry/seer/monitor_cleanup/runs.py deleted file mode 100644 index f6dc7fbefecd..000000000000 --- a/src/sentry/seer/monitor_cleanup/runs.py +++ /dev/null @@ -1,86 +0,0 @@ -from __future__ import annotations - -from typing import Any, cast -from uuid import UUID - -from rest_framework.exceptions import NotFound, PermissionDenied, Throttled -from rest_framework.request import Request - -from sentry import features -from sentry.models.organization import Organization -from sentry.ratelimits import backend as ratelimits -from sentry.seer.agent.client import SeerAgentClient -from sentry.seer.agent.types import FeatureRunStatus -from sentry.seer.models import SeerPermissionError -from sentry.seer.models.run import SeerAgentRun -from sentry.seer.models.workflow import SeerWorkflowRun, SeerWorkflowStrategy -from sentry.seer.monitor_cleanup.results import parse_monitor_cleanup_results -from sentry.seer.monitor_cleanup.schemas import SeerMonitorCleanupResponse -from sentry.seer.workflows.runs import create_workflow_run, deliver_workflow_result -from sentry.seer.workflows.schemas import WorkflowResult, WorkflowResultError, WorkflowRunSource -from sentry.users.models.user import User -from sentry.users.services.user import RpcUser - - -def create_monitor_cleanup_run( - request: Request, organization: Organization, *, source: WorkflowRunSource -) -> SeerWorkflowRun: - if not features.has( - "organizations:seer-workflows-monitor-cleanup", organization, actor=request.user - ): - raise NotFound - if not request.user.is_authenticated: - raise PermissionDenied("Sign in to run a monitor scan.") - try: - client = SeerAgentClient(organization=organization, user=cast(User | RpcUser, request.user)) - except SeerPermissionError as error: - raise PermissionDenied("Seer is not available for this organization.") from error - if ratelimits.is_limited( - f"seer-workflow:{organization.id}:{SeerWorkflowStrategy.DUPLICATE_MONITORS}", - limit=5, - window=3600, - ): - raise Throttled(detail="This organization has reached its scan limit. Try again later.") - return create_workflow_run( - client, - strategy=SeerWorkflowStrategy.DUPLICATE_MONITORS, - feature_id="monitor_cleanup", - payload={"response_version": 1}, - title="Monitor cleanup", - extras={"source": source, "project_ids": [], "results": []}, - ) - - -def deliver_monitor_cleanup_result( - organization_id: int, - run_uuid: UUID, - status: FeatureRunStatus, - result: dict[str, Any] | None, - error: str | None, - prompt_version: str | None = None, -) -> None: - deliver_workflow_result( - feature_id="monitor_cleanup", - organization_id=organization_id, - run_uuid=run_uuid, - status=status, - result=result, - error=error, - parse_result=_parse_result, - ) - - -def _parse_result(result: dict[str, Any], agent_run: SeerAgentRun) -> WorkflowResult: - if agent_run.run.user_id is None: - raise WorkflowResultError("The triggering user no longer exists.") - response = SeerMonitorCleanupResponse.parse_obj(result) - outputs = parse_monitor_cleanup_results( - response.data, agent_run.run.organization, agent_run.run.user_id - ) - scan_status = response.data.scan_status - if any(project.scan_status == "partial" for project in response.data.projects): - scan_status = "partial" - return WorkflowResult( - status=scan_status, - extras={"project_ids": [output["projectId"] for output in outputs], "results": outputs}, - ) diff --git a/src/sentry/seer/monitor_cleanup/schemas.py b/src/sentry/seer/monitor_cleanup/schemas.py deleted file mode 100644 index 726bb98b2968..000000000000 --- a/src/sentry/seer/monitor_cleanup/schemas.py +++ /dev/null @@ -1,122 +0,0 @@ -from __future__ import annotations - -from datetime import datetime -from typing import Literal, NotRequired, TypedDict - -from pydantic import BaseModel, Field - -from sentry.seer.workflows.schemas import WorkflowRunExtras, WorkflowRunSource, WorkflowRunStatus - - -class SeerMonitorPropertyValue(BaseModel): - monitor_id: int - value: str - - -class SeerMonitorPropertyComparison(BaseModel): - property: str - values: list[SeerMonitorPropertyValue] - - -class SeerMonitorFinding(BaseModel): - kind: Literal["exact_duplicate", "overlapping_coverage", "duplicate_notifications"] - monitor_ids: list[int] - suggested_keep_id: int | None = None - alert_ids: list[int] = Field(default_factory=list) - reason: str - comparison: list[SeerMonitorPropertyComparison] = Field(default_factory=list) - - -class SeerMonitorCleanupArtifact(BaseModel): - scan_status: Literal["complete", "partial"] - monitors_scanned: int - summary: str - findings: list[SeerMonitorFinding] - - -class SeerProjectMonitorCleanupArtifact(SeerMonitorCleanupArtifact): - project_id: int - - -class SeerOrganizationMonitorCleanupArtifact(BaseModel): - scan_status: Literal["complete", "partial"] - projects: list[SeerProjectMonitorCleanupArtifact] - - -class SeerMonitorCleanupResponse(BaseModel): - schema_version: Literal[1] - data: SeerOrganizationMonitorCleanupArtifact - - -# Sentry-built results and run extras: TypedDict describes the JSON stored on SeerAgentRun. - - -class MonitorCleanupOutput(TypedDict): - outputKind: Literal["monitor_cleanup"] - schemaVersion: Literal[1] - projectId: str - projectSlug: NotRequired[str] - scan: MonitorCleanupScan - summary: str - findings: list[MonitorCleanupFinding] - - -class MonitorCleanupScan(TypedDict): - status: Literal["complete", "partial"] - monitorsScanned: int - - -class MonitorCleanupFinding(TypedDict): - kind: Literal["exact_duplicate", "overlapping_coverage", "duplicate_notifications"] - monitors: list[MonitorCleanupResource] - suggestedKeepId: str | None - alerts: list[MonitorCleanupResource] - reason: str - comparison: list[MonitorCleanupComparison] - - -class MonitorCleanupResource(TypedDict): - id: str - name: str - enabled: bool - - -class MonitorCleanupComparison(TypedDict): - property: str - values: list[MonitorCleanupComparisonValue] - - -class MonitorCleanupComparisonValue(TypedDict): - monitorId: str - value: str - - -class MonitorCleanupRunExtras(WorkflowRunExtras): - project_ids: list[str] - results: list[MonitorCleanupOutput] - - -# Workflow API responses: datetimes are rendered as ISO strings by DRF. - - -class MonitorCleanupRunResponse(TypedDict): - id: str - source: WorkflowRunSource | None - seerRunId: str - dateAdded: datetime - dateCompleted: datetime | None - strategy: Literal["duplicate_monitors"] - extras: MonitorCleanupRunStatus - errorMessage: str | None - results: list[MonitorCleanupResultResponse] - - -class MonitorCleanupRunStatus(TypedDict): - status: WorkflowRunStatus - - -class MonitorCleanupResultResponse(TypedDict): - id: str - kind: Literal["duplicate_monitors"] - seerRunId: str - extras: MonitorCleanupOutput diff --git a/src/sentry/seer/workflows/runs.py b/src/sentry/seer/workflows/runs.py index d540a6a414d1..eefcc1041d68 100644 --- a/src/sentry/seer/workflows/runs.py +++ b/src/sentry/seer/workflows/runs.py @@ -1,21 +1,16 @@ from __future__ import annotations import logging -from collections.abc import Callable, Iterator, Mapping, Sequence +from collections.abc import Iterator, Sequence from contextlib import contextmanager from dataclasses import dataclass -from functools import partial -from typing import Any, Literal -from uuid import UUID +from typing import Literal from django.db import router, transaction from django.db.models import QuerySet from django.utils import timezone from pydantic import ValidationError -from sentry.seer.agent.client import SeerAgentClient -from sentry.seer.agent.types import FeatureRunStatus -from sentry.seer.models.run import SeerAgentRun, SeerRun, SeerRunMirrorStatus from sentry.seer.models.workflow import ( SeerWorkflowConfig, SeerWorkflowExecutionExtras, @@ -24,10 +19,8 @@ SeerWorkflowRunExecution, SeerWorkflowRunExtras, SeerWorkflowRunStatus, - SeerWorkflowStrategy, WorkflowRunError, ) -from sentry.seer.workflows.schemas import WorkflowResult, WorkflowResultError, WorkflowRunExtras from sentry.utils import json logger = logging.getLogger(__name__) @@ -239,139 +232,3 @@ def _parse_extras[Extras: SeerWorkflowRunExtras]( extra={"run_id": run.id, "organization_id": run.organization_id}, ) raise - - -def create_workflow_run( - client: SeerAgentClient, - *, - strategy: SeerWorkflowStrategy, - feature_id: str, - payload: dict[str, Any], - title: str, - extras: Mapping[str, Any] | None = None, -) -> SeerWorkflowRun: - """Atomically create a workflow with one execution and enqueue its dispatch.""" - run = client.start_feature_run( - feature_id=feature_id, - payload=payload, - title=title, - flush=False, - extras={**(extras or {}), "status": "running", "error": None}, - referrer=feature_id, - on_run_created=partial(_create_workflow_run, strategy=strategy), - ) - return run.workflow_execution.run - - -def _create_workflow_run(seer_run: SeerRun, *, strategy: SeerWorkflowStrategy) -> None: - config = SeerWorkflowConfig.get_or_create_for_strategy(seer_run.organization_id, strategy) - run = SeerWorkflowRun.objects.create( - organization_id=seer_run.organization_id, workflow_config=config - ) - SeerWorkflowRunExecution.objects.create(run=run, seer_run=seer_run) - - -def deliver_workflow_result( - *, - feature_id: str, - organization_id: int, - run_uuid: UUID, - status: FeatureRunStatus, - result: dict[str, Any] | None, - error: str | None, - parse_result: Callable[[dict[str, Any], SeerAgentRun], WorkflowResult], -) -> None: - log_extra = { - "feature_id": feature_id, - "organization_id": organization_id, - "run_uuid": str(run_uuid), - "status": status, - } - logger.info("seer.workflow.delivery.received", extra=log_extra) - agent_run = ( - SeerAgentRun.objects.select_related("run__organization") - .filter(run__organization_id=organization_id, source=feature_id, run__uuid=run_uuid) - .first() - ) - if agent_run is None: - logger.warning("seer.workflow.delivery.missing_run", extra=log_extra) - return - if agent_run.extras.get("status") in TERMINAL_STATUSES: - logger.info("seer.workflow.delivery.already_finished", extra=log_extra) - return - finish = partial( - finish_workflow_run, - agent_run.run_id, - organization_id=organization_id, - feature_id=feature_id, - ) - if status != "completed" or result is None: - logger.warning("seer.workflow.delivery.failed", extra={**log_extra, "error": error}) - finish(error="Seer could not complete this workflow.") - return - try: - parsed = parse_result(result, agent_run) - except WorkflowResultError as exc: - logger.exception("seer.workflow.result_processing_failed", extra=log_extra) - finish(error=str(exc)) - return - except Exception: - logger.exception("seer.workflow.invalid_output", extra=log_extra) - finish(error="Seer returned results that could not be loaded.") - return - finish(result=parsed) - - -def finish_workflow_run( - run_id: int, - *, - organization_id: int, - feature_id: str, - result: WorkflowResult | None = None, - error: str | None = None, -) -> None: - """Finalize the single-execution workflow created by create_workflow_run.""" - with transaction.atomic(router.db_for_write(SeerAgentRun)): - agent_run = ( - SeerAgentRun.objects.select_for_update(of=("self",)) - .filter(run_id=run_id, run__organization_id=organization_id, source=feature_id) - .first() - ) - if agent_run is None or agent_run.extras.get("status") in TERMINAL_STATUSES: - return - extras: WorkflowRunExtras = { - "status": "failed" if error else result.status if result else "complete", - "error": error, - } - agent_run.update(extras={**agent_run.extras, **(result.extras if result else {}), **extras}) - SeerWorkflowRun.objects.filter( - organization_id=organization_id, - executions__seer_run_id=run_id, - date_completed__isnull=True, - ).update(date_completed=timezone.now()) - logger.info( - "seer.workflow.run.finished", - extra={ - "feature_id": feature_id, - "run_id": run_id, - "organization_id": organization_id, - "status": extras["status"], - "error": error, - }, - ) - - -def get_workflow_run_status(agent_run: SeerAgentRun) -> WorkflowRunExtras: - extras: WorkflowRunExtras = agent_run.extras - if ( - extras["status"] not in TERMINAL_STATUSES - and agent_run.run.mirror_status == SeerRunMirrorStatus.FAILED - ): - return { - "status": "failed", - "error": "Seer could not start this workflow.", - } - return { - "status": extras["status"], - "error": extras["error"], - } diff --git a/src/sentry/seer/workflows/schemas.py b/src/sentry/seer/workflows/schemas.py index 55281fda900d..c5843254bf94 100644 --- a/src/sentry/seer/workflows/schemas.py +++ b/src/sentry/seer/workflows/schemas.py @@ -1,22 +1,3 @@ -from collections.abc import Mapping -from dataclasses import dataclass, field -from typing import Any, Literal, NotRequired, TypedDict +from typing import Literal WorkflowRunSource = Literal["cron", "manual"] -WorkflowRunStatus = Literal["running", "complete", "partial", "failed"] - - -class WorkflowRunExtras(TypedDict): - source: NotRequired[WorkflowRunSource] - status: WorkflowRunStatus - error: str | None - - -@dataclass(frozen=True) -class WorkflowResult: - extras: Mapping[str, Any] = field(default_factory=dict) - status: Literal["complete", "partial"] = "complete" - - -class WorkflowResultError(Exception): - """A result processing failure with a message safe to show to the user.""" diff --git a/tests/sentry/seer/endpoints/test_organization_seer_workflow_runs.py b/tests/sentry/seer/endpoints/test_organization_seer_workflow_runs.py index 735e277d98c5..d6e392d92648 100644 --- a/tests/sentry/seer/endpoints/test_organization_seer_workflow_runs.py +++ b/tests/sentry/seer/endpoints/test_organization_seer_workflow_runs.py @@ -1,7 +1,5 @@ from collections.abc import Sequence from datetime import UTC, datetime -from typing import Any -from unittest.mock import patch from django.db import connections, router from django.db.models import F @@ -13,15 +11,12 @@ from sentry.apidocs.examples.seer_workflow_examples import WORKFLOW_RUNS_RESPONSE from sentry.apidocs.spectacular_ports import resolve_type_hint from sentry.db.models.fields.bounded import I64_MAX -from sentry.hybridcloud.models.outbox import CellOutbox -from sentry.hybridcloud.outbox.category import OutboxCategory from sentry.notebooks.models import NotebookStatus from sentry.seer.endpoints.organization_seer_workflow_runs import ( SeerWorkflowRunsResponse, WorkflowRunListErrorResponse, serialize_workflow_runs, ) -from sentry.seer.models.run import SeerAgentRun from sentry.seer.models.workflow import ( SeerWorkflowExecutionStatus, SeerWorkflowRun, @@ -29,13 +24,11 @@ SeerWorkflowRunStatus, SeerWorkflowStrategy, ) -from sentry.seer.monitor_cleanup.runs import deliver_monitor_cleanup_result from sentry.testutils.cases import APITestCase from sentry.testutils.factories import Factories from sentry.testutils.helpers import parse_link_header from sentry.testutils.helpers.features import with_feature from sentry.utils import json -from sentry.workflow_engine.models import Workflow @override_settings(SENTRY_SELF_HOSTED=False) @@ -565,175 +558,3 @@ def create_run( return Factories.create_seer_workflow_run( self.organization, workflow_config=self.config, **kwargs ) - - -@override_settings(SENTRY_SELF_HOSTED=False) -class OrganizationSeerMonitorCleanupTest(APITestCase): - endpoint = "sentry-api-0-organization-seer-workflow-runs" - method = "post" - - def setUp(self) -> None: - super().setUp() - rate_limit_patcher = patch( - "sentry.middleware.ratelimit.get_rate_limit_value", return_value=None - ) - rate_limit_patcher.start() - self.addCleanup(rate_limit_patcher.stop) - self.keep = self.create_detector(project=self.project, type="metric_issue", name="Keep") - self.duplicate = self.create_detector( - project=self.project, type="metric_issue", name="Copy" - ) - self.login_as(self.user) - - def test_scan_stores_findings(self) -> None: - run = self.trigger() - assert run.source == "monitor_cleanup" - assert run.extras["source"] == "manual" - outbox = CellOutbox.objects.get( - category=OutboxCategory.SEER_RUN_CREATE, object_identifier=run.run_id - ) - assert outbox.payload is not None - assert outbox.payload["body"]["payload"] == {"response_version": 1} - result = self.result() - result["data"]["projects"][0]["scan_status"] = "partial" - findings = result["data"]["projects"][0]["findings"] - findings.extend( - [ - {**findings[0], "monitor_ids": [str(self.keep.id), "999999"]}, - {**findings[0], "alert_ids": ["999999"]}, - ] - ) - deliver_monitor_cleanup_result( - self.organization.id, run.run.uuid, "completed", result, None - ) - run.refresh_from_db() - assert run.run.workflow_execution.run.date_completed is not None - assert run.extras["status"] == "partial" - assert len(run.extras["results"]) == 1 - (finding,) = run.extras["results"][0]["findings"] - assert finding["suggestedKeepId"] == str(self.keep.id) - assert finding["monitors"] == [ - {"id": str(self.keep.id), "name": "Keep", "enabled": self.keep.enabled}, - {"id": str(self.duplicate.id), "name": "Copy", "enabled": self.duplicate.enabled}, - ] - - def test_hidden_workflow_findings_are_dropped(self) -> None: - alert_workflow = self.create_workflow(organization=self.organization, name="Alert") - - hidden_workflow = self.create_workflow( - organization=self.organization, name="Hidden", type=Workflow.Type.INBOX - ) - - self.create_detector_workflow(detector=self.keep, workflow=alert_workflow) - self.create_detector_workflow(detector=self.keep, workflow=hidden_workflow) - - run = self.trigger() - - result = self.result() - - findings = result["data"]["projects"][0]["findings"] - findings[0]["alert_ids"] = [str(alert_workflow.id)] - findings.append({**findings[0], "alert_ids": [str(hidden_workflow.id)]}) - - deliver_monitor_cleanup_result( - self.organization.id, run.run.uuid, "completed", result, None - ) - - run.refresh_from_db() - (finding,) = run.extras["results"][0]["findings"] - - assert finding["alerts"] == [ - {"id": str(alert_workflow.id), "name": "Alert", "enabled": alert_workflow.enabled} - ] - - def test_invalid_results_report_safe_errors(self) -> None: - run = self.trigger() - deliver_monitor_cleanup_result( - self.organization.id, run.run.uuid, "completed", {"schema_version": 1, "data": {}}, None - ) - run.refresh_from_db() - assert run.extras["status"] == "failed" - assert not run.extras["results"] - assert run.extras["error"] == "Seer returned results that could not be loaded." - - run = self.trigger() - result = self.result() - result["data"]["projects"][0]["project_id"] = "999999999" - deliver_monitor_cleanup_result( - self.organization.id, run.run.uuid, "completed", result, None - ) - run.refresh_from_db() - assert run.extras["status"] == "failed" - assert run.extras["error"] == "Some scanned projects are no longer accessible." - - def test_requires_feature_and_seer_access(self) -> None: - with patch("sentry.seer.monitor_cleanup.runs.ratelimits.is_limited") as limit: - self.get_error_response( - self.organization.slug, strategy="duplicate_monitors", status_code=404 - ) - with ( - override_settings(SENTRY_SELF_HOSTED=True), - self.feature("organizations:seer-workflows-monitor-cleanup"), - ): - response = self.get_error_response( - self.organization.slug, strategy="duplicate_monitors", status_code=403 - ) - assert response.data == {"detail": "Seer is not available for this organization."} - limit.assert_not_called() - - limit.return_value = True - with self.feature("organizations:seer-workflows-monitor-cleanup"): - self.get_error_response( - self.organization.slug, strategy="duplicate_monitors", status_code=429 - ) - limit.assert_called_once_with( - f"seer-workflow:{self.organization.id}:duplicate_monitors", limit=5, window=3600 - ) - - def test_requires_organization_access(self) -> None: - other = self.create_organization() - with self.feature("organizations:seer-workflows-monitor-cleanup"): - self.get_error_response(other.slug, strategy="duplicate_monitors", status_code=403) - - def test_callback_for_deleted_user_marks_run_failed(self) -> None: - agent_run = self.trigger() - agent_run.run.update(user_id=999999999) - deliver_monitor_cleanup_result( - self.organization.id, agent_run.run.uuid, "completed", self.result(), None - ) - agent_run.refresh_from_db() - assert agent_run.extras["status"] == "failed" - assert agent_run.extras["error"] == "The triggering user no longer exists." - - def trigger(self) -> SeerAgentRun: - with self.feature("organizations:seer-workflows-monitor-cleanup"): - response = self.get_success_response( - self.organization.slug, strategy="duplicate_monitors", status_code=202 - ) - return SeerAgentRun.objects.select_related("run").get( - run__workflow_execution__run_id=response.data["runId"] - ) - - def result(self) -> dict[str, Any]: - return { - "schema_version": 1, - "data": { - "scan_status": "complete", - "projects": [ - { - "project_id": str(self.project.id), - "scan_status": "complete", - "monitors_scanned": 2, - "summary": "Duplicate monitors found.", - "findings": [ - { - "kind": "exact_duplicate", - "monitor_ids": [str(self.keep.id), str(self.duplicate.id)], - "suggested_keep_id": str(self.keep.id), - "reason": "Identical settings.", - } - ], - } - ], - }, - } diff --git a/tests/sentry/seer/workflows/test_runs.py b/tests/sentry/seer/workflows/test_runs.py index 30c1065a2d42..cd1a58222e0e 100644 --- a/tests/sentry/seer/workflows/test_runs.py +++ b/tests/sentry/seer/workflows/test_runs.py @@ -1,14 +1,9 @@ -from functools import partial -from unittest.mock import Mock, patch +from unittest.mock import patch import pytest from django.test import override_settings from pydantic import BaseModel -from sentry.hybridcloud.models.outbox import CellOutbox -from sentry.hybridcloud.outbox.category import OutboxCategory -from sentry.seer.agent.client import SeerAgentClient -from sentry.seer.models.run import SeerAgentRun, SeerRun, SeerRunMirrorStatus from sentry.seer.models.workflow import ( SeerWorkflowConfig, SeerWorkflowExecutionExtras, @@ -24,12 +19,7 @@ from sentry.seer.workflows.runs import ( InvalidWorkflowTransition, SeerWorkflowRunLifecycle, - create_workflow_run, - deliver_workflow_result, - finish_workflow_run, - get_workflow_run_status, ) -from sentry.seer.workflows.schemas import WorkflowResult from sentry.seer.workflows.workflow import SeerWorkflow from sentry.testutils.cases import TestCase from sentry.testutils.factories import Factories @@ -40,155 +30,6 @@ class RunInputs(BaseModel): @override_settings(SENTRY_SELF_HOSTED=False) -class WorkflowRunTest(TestCase): - def test_creation_links_execution_and_defers_dispatch(self) -> None: - with patch("sentry.receivers.outbox.cell.make_feature_run_request") as dispatch: - run = self.create_run() - dispatch.assert_not_called() - workflow = run.workflow_execution.run - assert workflow.organization_id == self.organization.id - assert workflow.date_completed is None - assert workflow.workflow_config is not None - assert workflow.workflow_config.strategy == SeerWorkflowStrategy.AGENTIC_TRIAGE - assert run.agent.extras == { - "status": "running", - "error": None, - "summary": None, - } - outbox = CellOutbox.objects.get( - category=OutboxCategory.SEER_RUN_CREATE, object_identifier=run.id - ) - assert outbox.payload is not None - assert outbox.payload["body"]["feature_id"] == "test_workflow" - assert outbox.payload["viewer_context"]["user_id"] == self.user.id - - def test_creation_rolls_back_if_execution_creation_fails(self) -> None: - with ( - patch( - "sentry.seer.workflows.runs.SeerWorkflowRunExecution.objects.create", - side_effect=RuntimeError("Cannot create execution"), - ), - pytest.raises(RuntimeError, match="Cannot create execution"), - ): - self.create_run() - assert not SeerWorkflowRun.objects.filter(organization=self.organization).exists() - assert not SeerRun.objects.filter(organization=self.organization).exists() - assert not SeerAgentRun.objects.filter(run__organization=self.organization).exists() - assert not CellOutbox.objects.filter(category=OutboxCategory.SEER_RUN_CREATE).exists() - - def test_delivery_stores_feature_results_and_ignores_duplicates(self) -> None: - run = self.create_run() - parser = Mock(return_value=WorkflowResult(extras={"summary": "Finished"}, status="partial")) - deliver = partial( - deliver_workflow_result, - feature_id="test_workflow", - organization_id=self.organization.id, - run_uuid=run.uuid, - parse_result=parser, - ) - deliver(status="completed", result={"raw": "output"}, error=None) - run.agent.refresh_from_db() - assert run.agent.extras["status"] == "partial" - assert run.agent.extras["summary"] == "Finished" - workflow = run.workflow_execution.run - workflow.refresh_from_db() - assert workflow.date_completed is not None - assert "date_completed" not in run.agent.extras - parser.assert_called_once_with({"raw": "output"}, run.agent) - completed_extras = run.agent.extras.copy() - completed_at = workflow.date_completed - - deliver(status="error", result=None, error="Late failure") - finish_workflow_run( - run.id, - organization_id=self.organization.id, - feature_id="test_workflow", - error="Late failure", - ) - run.agent.refresh_from_db() - assert run.agent.extras == completed_extras - workflow.refresh_from_db() - assert workflow.date_completed == completed_at - - def test_delivery_and_finish_are_scoped_to_organization_and_feature(self) -> None: - run = self.create_run() - parser = Mock() - deliver = partial( - deliver_workflow_result, - run_uuid=run.uuid, - status="completed", - result={}, - error=None, - parse_result=parser, - ) - other_org = self.create_organization() - deliver(organization_id=other_org.id, feature_id="test_workflow") - deliver(organization_id=self.organization.id, feature_id="another_workflow") - finish_workflow_run( - run.id, organization_id=other_org.id, feature_id="test_workflow", error="Wrong org" - ) - finish_workflow_run( - run.id, - organization_id=self.organization.id, - feature_id="another_workflow", - error="Wrong workflow", - ) - parser.assert_not_called() - run.agent.refresh_from_db() - assert run.agent.extras["status"] == "running" - assert SeerWorkflowRun.objects.get(executions__seer_run=run).date_completed is None - - def test_failed_delivery_records_safe_error_without_parsing(self) -> None: - run = self.create_run() - parser = Mock() - deliver_workflow_result( - feature_id="test_workflow", - organization_id=self.organization.id, - run_uuid=run.uuid, - status="error", - result=None, - error="Internal error details", - parse_result=parser, - ) - parser.assert_not_called() - run.agent.refresh_from_db() - assert run.agent.extras["status"] == "failed" - assert SeerWorkflowRun.objects.get(executions__seer_run=run).date_completed is not None - assert run.agent.extras["error"] == "Seer could not complete this workflow." - - def test_dispatch_failure_is_reported_without_overwriting_a_completed_result(self) -> None: - run = self.create_run() - run.update(mirror_status=SeerRunMirrorStatus.FAILED) - assert get_workflow_run_status(run.agent) == { - "status": "failed", - "error": "Seer could not start this workflow.", - } - finish_workflow_run( - run.id, - organization_id=self.organization.id, - feature_id="test_workflow", - result=WorkflowResult(extras={"summary": "Finished"}), - ) - run.agent.refresh_from_db() - status = get_workflow_run_status(run.agent) - assert status["status"] == "complete" - assert SeerWorkflowRun.objects.get(executions__seer_run=run).date_completed is not None - assert status["error"] is None - - def create_run(self) -> SeerRun: - workflow = create_workflow_run( - SeerAgentClient(self.organization, self.user), - strategy=SeerWorkflowStrategy.AGENTIC_TRIAGE, - feature_id="test_workflow", - title="Test workflow", - payload={}, - extras={"summary": None}, - ) - run = workflow.executions.get().seer_run - assert run is not None - return run - - def admit(workflow: SeerWorkflow, *, trigger_key: str, inputs: BaseModel) -> AdmittedRun: with lock_config(workflow.organization_id, workflow.config_id) as config: return get_definition(config.strategy).admit_run(