feat(investigations): Validate action request events

src/sentry/investigations/contracts.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-from typing import Any
+import logging
+import re
+from typing import Any, Literal, TypedDict, cast
 
 import orjson
 from django.utils import timezone
@@ -14,6 +16,8 @@
     InvestigationOrchestrationStatus,
 )
 
+logger = logging.getLogger(__name__)
+
 MAX_TABLE_ROWS = 100
 MAX_CHART_SERIES = 5
 MAX_POINTS_PER_SERIES = 200
@@ -715,6 +719,30 @@ def validate_heartbeatAt(self, value: str) -> str:
 
 
 MAX_REPORT_BLOCKS = 200
+MAX_ACTION_REQUESTS = 50
+ACTION_REQUEST_KEY_PATTERN = re.compile(r"[a-z0-9][a-z0-9_-]{0,63}")
+ANCHOR_TYPES = ("block", "hypothesis")
+
+
+class ActionRequestAnchor(TypedDict):
+    type: Literal["block", "hypothesis"]
+    id: str
+
+
+class ActionRequestItem(TypedDict):
+    """One request from Seer, after the contract validated it."""
+
+    key: str
+    reportRevision: int
+    title: str
+    rationale: str
+    impact: str | None
+    primary: bool
+    anchor: ActionRequestAnchor | None
+    # Owned by the action kind; only the kind reads inside it.
+    action: dict[str, Any] | None
+
+
 MAX_TEXT_DELTA_CHARS = 64 * 1024
 MAX_SUMMARY_DESCRIPTION_CHARS = 10_000
 
@@ -738,6 +766,7 @@ class StateSnapshotPayloadSerializer(RelaxedContractSerializer):
         child=serializers.DictField(), required=False, max_length=MAX_REPORT_BLOCKS
     )
     metadata = serializers.DictField(required=False, allow_null=True)
+    actionRequests = serializers.ListField(child=serializers.JSONField(), required=False)
 
 
 class ReportClearPayloadSerializer(_ReportRevisionMixin):
@@ -780,6 +809,117 @@ class MetadataCompletedPayloadSerializer(_ReportRevisionMixin):
     title = StrictCharField(required=False, max_length=255)
 
 
+class ActionPayloadSerializer(RelaxedContractSerializer):
+    """The base for an action kind's payload. Each kind validates its own fields."""
+
+    kind = StrictCharField(max_length=64)
+
+    @classmethod
+    def trim(cls, action: dict[str, Any]) -> dict[str, Any] | None:
+        """The action without values this kind won't accept. None keeps the request as advice."""
+        return action
+
+
+def _payload_serializer(action: Any) -> type[ActionPayloadSerializer] | None:
+    from sentry.investigations.services.action_kinds import get_action_kind
+
+    kind = action.get("kind") if isinstance(action, dict) else None
+    action_kind = get_action_kind(kind) if isinstance(kind, str) else None
+    return action_kind.payload_serializer if action_kind is not None else None
+
+
+class ActionRequestAnchorSerializer(StrictContractSerializer):
+    type = serializers.ChoiceField(choices=ANCHOR_TYPES)
+    id = StrictCharField(max_length=128)
+
+
+class ActionRequestSerializer(RelaxedContractSerializer):
+    key = StrictCharField(max_length=64)
+    reportRevision = StrictIntegerField(min_value=0, max_value=I32_MAX)
+    title = StrictCharField(max_length=200)
+    rationale = StrictCharField(max_length=1_000)
+    impact = OptionalStrictCharField(1_000)
+    primary = StrictBooleanField(required=False, default=False)
+    anchor = ActionRequestAnchorSerializer(required=False, allow_null=True, default=None)
+    action = serializers.JSONField(required=False, allow_null=True, default=None)
+
+    def validate_key(self, value: str) -> str:
+        if ACTION_REQUEST_KEY_PATTERN.fullmatch(value) is None:
+            raise serializers.ValidationError("Must be a lowercase slug.")
+        return value
+
+    def validate_action(self, value: Any) -> dict[str, Any] | None:
+        if value is None:
+            return None
+        serializer_cls = _payload_serializer(value)
+        if serializer_cls is None:
+            raise serializers.ValidationError("Unknown action kind.")
+        validator = serializer_cls(data=value)
+        if not validator.is_valid():
+            raise serializers.ValidationError(validator.errors)
+        return dict(validator.validated_data)
+
+
+def _log_extra(item: Any) -> dict[str, Any]:
+    key = item.get("key") if isinstance(item, dict) else None
+    action = item.get("action") if isinstance(item, dict) else None
+    kind = action.get("kind") if isinstance(action, dict) else None
+    return {
+        "key": key[:64] if isinstance(key, str) else None,
+        "kind": kind[:64] if isinstance(kind, str) else None,
+    }
+
+
+def _trimmed(item: Any, *, log: bool) -> Any:
+    action = item.get("action") if isinstance(item, dict) else None
+    serializer_cls = _payload_serializer(action)
+    if serializer_cls is None or not isinstance(action, dict):
+        return item
+    trimmed = serializer_cls.trim(action)
+    if log and trimmed != action:
+        logger.warning("investigations.action_requests.action_trimmed", extra=_log_extra(item))
+    return {**item, "action": trimmed}
+
+
+def valid_action_request(item: Any, *, log: bool = False) -> ActionRequestItem | None:
+    """The request Sentry can store, or None.
+
+    Seer can ship kinds, providers, or fields Sentry does not know yet, so a bad request is
+    dropped, never rejected with its event. Only the apply path logs, so each drop logs once.
+    """
+    trimmed = _trimmed(item, log=log)
+    validator = ActionRequestSerializer(data=trimmed)
+    if validator.is_valid():
+        return cast(ActionRequestItem, dict(validator.validated_data))
+    if log:
+        logger.warning(
+            "investigations.action_requests.dropped",
+            extra={**_log_extra(item), "fields": sorted(validator.errors)[:10]},
+        )
+    return None
+
+
+def valid_action_requests(items: list[Any], *, log: bool = False) -> list[ActionRequestItem]:
+    kept: list[ActionRequestItem] = []
+    for item in items[:MAX_ACTION_REQUESTS]:
+        request = valid_action_request(item, log=log)
+        if request is not None and all(other["key"] != request["key"] for other in kept):
+            kept.append(request)
+    return kept
+
+
+class ActionRequestedPayloadSerializer(RelaxedContractSerializer):
+    request = serializers.JSONField()
+
+    def validate_request(self, value: Any) -> ActionRequestItem | None:
+        return valid_action_request(value, log=self.context.get("log_drops", False))
+
+
+class ActionWithdrawnPayloadSerializer(RelaxedContractSerializer):
+    key = StrictCharField(max_length=64)
+    reason = StrictCharField(max_length=1_000)
+
+
 class WorkflowFailedPayloadSerializer(RelaxedContractSerializer):
     error = ProjectionErrorSerializer()
     projection = OrchestrationProjectionSerializer(required=False, allow_null=True)
@@ -843,5 +983,7 @@ def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
     "report_failed": ReportFailedPayloadSerializer,
     "title_delta": TitleDeltaPayloadSerializer,
     "metadata_completed": MetadataCompletedPayloadSerializer,
+    "action_requested": ActionRequestedPayloadSerializer,
+    "action_withdrawn": ActionWithdrawnPayloadSerializer,
     "workflow_failed": WorkflowFailedPayloadSerializer,
 }

src/sentry/investigations/endpoints/validators/orchestration.py

@@ -26,6 +26,8 @@
     "report_failed",
     "title_delta",
     "metadata_completed",
+    "action_requested",
+    "action_withdrawn",
     "workflow_failed",
 )
 

src/sentry/investigations/services/action_kinds/__init__.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+from sentry.investigations.services.action_kinds.base import ActionKind
+from sentry.investigations.services.action_kinds.connect_telemetry import ConnectTelemetryKind
+from sentry.investigations.services.action_kinds.update_detector import UpdateDetectorKind
+
+ACTION_KINDS: dict[str, ActionKind] = {
+    kind.kind: kind for kind in (UpdateDetectorKind(), ConnectTelemetryKind())
+}
+
+
+def get_action_kind(kind: str | None) -> ActionKind | None:
+    return ACTION_KINDS.get(kind) if kind is not None else None

src/sentry/investigations/services/action_kinds/base.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+import abc
+
+from sentry.investigations.contracts import ActionPayloadSerializer
+
+
+class ActionKind(abc.ABC):
+    """How Sentry checks, applies, and records one kind of action."""
+
+    kind: str
+    payload_serializer: type[ActionPayloadSerializer]

src/sentry/investigations/services/action_kinds/connect_telemetry.py

@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+from rest_framework import serializers
+
+from sentry.investigations.contracts import ActionPayloadSerializer
+from sentry.investigations.services.action_kinds.base import ActionKind
+
+CONNECT_TELEMETRY_ACTION = "connect_telemetry"
+TELEMETRY_PROVIDERS = ("datadog", "gcp")
+
+
+class ConnectTelemetryActionSerializer(ActionPayloadSerializer):
+    provider = serializers.ChoiceField(choices=TELEMETRY_PROVIDERS)
+
+
+class ConnectTelemetryKind(ActionKind):
+    kind = CONNECT_TELEMETRY_ACTION
+    payload_serializer = ConnectTelemetryActionSerializer

src/sentry/investigations/services/action_kinds/update_detector.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+from typing import Any
+
+from rest_framework import serializers
+
+from sentry.db.models.fields.bounded import I32_MAX, I64_MAX
+from sentry.investigations.contracts import (
+    ActionPayloadSerializer,
+    StrictCharField,
+    StrictContractSerializer,
+    StrictFloatField,
+    StrictIntegerField,
+)
+from sentry.investigations.services.action_kinds.base import ActionKind
+
+UPDATE_DETECTOR_ACTION = "update_detector"
+
+
+_DETECTOR_TEXT_LIMITS = {"query": 2_000, "aggregate": 500, "environment": 64}
+
+
+class DetectorChangesSerializer(StrictContractSerializer):
+    """The detector settings an `update_detector` action reads or writes."""
+
+    timeWindowSeconds = StrictIntegerField(required=False, min_value=1, max_value=I32_MAX)
+    highThreshold = StrictFloatField(required=False)
+    mediumThreshold = StrictFloatField(required=False)
+    resolutionThreshold = StrictFloatField(required=False)
+    query = StrictCharField(
+        required=False, allow_blank=True, max_length=_DETECTOR_TEXT_LIMITS["query"]
+    )
+    aggregate = StrictCharField(required=False, max_length=_DETECTOR_TEXT_LIMITS["aggregate"])
+    environment = StrictCharField(
+        required=False, allow_null=True, max_length=_DETECTOR_TEXT_LIMITS["environment"]
+    )
+
+
+class UpdateDetectorActionSerializer(ActionPayloadSerializer):
+    detectorId = StrictCharField(max_length=32)
+    before = DetectorChangesSerializer()
+    changes = DetectorChangesSerializer()
+
+    def validate_detectorId(self, value: str) -> str:
+        if not (value.isascii() and value.isdigit()) or int(value) > I64_MAX:
+            raise serializers.ValidationError("Must be a numeric ID.")
+        return value
+
+    def validate(self, attrs: dict[str, Any]) -> dict[str, Any]:
+        before = attrs["before"]
+        changes = attrs["changes"]
+        if not changes:
+            raise serializers.ValidationError({"changes": "At least one change is required."})
+        if set(before) != set(changes):
+            raise serializers.ValidationError("before and changes must have the same keys.")
+        if any(changes[key] == before[key] for key in changes):
+            raise serializers.ValidationError({"changes": "Every change must differ from before."})
+        return attrs
+
+    @classmethod
+    def trim(cls, action: dict[str, Any]) -> dict[str, Any] | None:
+        """Drop changes we won't apply: text over our limits, or a null current aggregate.
+
+        No change left makes the request advice.
+        """
+
+        before = action.get("before")
+        changes = action.get("changes")
+        if (
+            not isinstance(before, dict)
+            or not isinstance(changes, dict)
+            or set(before) != set(changes)
+        ):
+            return action
+        dropped = {
+            key
+            for key, limit in _DETECTOR_TEXT_LIMITS.items()
+            for values in (before, changes)
+            if key in values
+            and (
+                values[key] is None
+                if key == "aggregate"
+                else isinstance(values[key], str) and len(values[key]) > limit
+            )
+        }
+        if not dropped:
+            return action
+        kept = [key for key in changes if key not in dropped]
+        if not kept:
+            return None
+        return {
+            **action,
+            "before": {key: before[key] for key in kept},
+            "changes": {key: changes[key] for key in kept},
+        }
+
+
+class UpdateDetectorKind(ActionKind):
+    kind = UPDATE_DETECTOR_ACTION
+    payload_serializer = UpdateDetectorActionSerializer

src/sentry/investigations/services/orchestration_events.py

@@ -138,7 +138,9 @@ def _validate_projection_project_scope(
         )
 
 
-def _validate_event_payload(event_type: str, payload: dict[str, Any]) -> dict[str, Any]:
+def _validate_event_payload(
+    event_type: str, payload: dict[str, Any], *, log_drops: bool = False
+) -> dict[str, Any]:
     """Check the payload against the schema for its event type and normalize it.
 
     Unknown fields survive validation; Seer owns the shape of these payloads and
@@ -149,7 +151,7 @@ def _validate_event_payload(event_type: str, payload: dict[str, Any]) -> dict[st
     schema = EVENT_PAYLOAD_SERIALIZERS.get(event_type)
     if schema is None:
         return payload
-    validator = schema(data=payload)
+    validator = schema(data=payload, context={"log_drops": log_drops})
     if not validator.is_valid():
         raise serializers.ValidationError({"payload": validator.errors})
     return validator.validated_data
@@ -811,7 +813,7 @@ def _apply_event(
     # Events are staged verbatim so that their identity does not move when Sentry's
     # schemas change. Normalizing here is what lets the apply path read coerced
     # values and defaults without re-checking any types.
-    payload = _validate_event_payload(event.type, _event_data(event))
+    payload = _validate_event_payload(event.type, _event_data(event), log_drops=True)
     generation = _event_generation(event)
     if generation < run.generation:
         return False, "stale_generation"

tests/sentry/investigations/services/test_connect_telemetry_contract.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+import pytest
+
+from sentry.investigations.services.action_kinds.connect_telemetry import (
+    ConnectTelemetryActionSerializer,
+)
+
+
+@pytest.mark.parametrize("provider", ["datadog", "gcp"])
+def test_accepts_supported_providers(provider: str) -> None:
+    action = {"kind": "connect_telemetry", "provider": provider}
+    serializer = ConnectTelemetryActionSerializer(data=action)
+
+    assert serializer.is_valid()
+    assert dict(serializer.validated_data) == action
+
+
+@pytest.mark.parametrize("provider", ["honeycomb", None, ""])
+def test_rejects_unknown_providers(provider: str | None) -> None:
+    serializer = ConnectTelemetryActionSerializer(
+        data={"kind": "connect_telemetry", "provider": provider}
+    )
+
+    assert not serializer.is_valid()