diff --git a/src/sentry/api/urls.py b/src/sentry/api/urls.py
index e15d16be4dfc..ac83761c2d05 100644
--- a/src/sentry/api/urls.py
+++ b/src/sentry/api/urls.py
@@ -296,6 +296,9 @@
 from sentry.integrations.api.endpoints.organization_repository_settings import (
     OrganizationRepositorySettingsEndpoint,
 )
+from sentry.investigations.endpoints.organization_investigation_action_request_apply import (
+    OrganizationInvestigationActionRequestApplyEndpoint,
+)
 from sentry.investigations.endpoints.organization_investigation_block_details import (
     OrganizationInvestigationBlockDetailsEndpoint,
 )
@@ -2452,6 +2455,11 @@ def create_group_urls(
         OrganizationInvestigationsDuplicateEndpoint.as_view(),
         name="sentry-api-0-organization-investigation-duplicate",
     ),
+    re_path(
+        r"^(?P<organization_id_or_slug>[^/]+)/investigations/(?P<investigation_id>[^/]+)/action-requests/(?P<action_request_id>[^/]+)/apply/$",
+        OrganizationInvestigationActionRequestApplyEndpoint.as_view(),
+        name="sentry-api-0-organization-investigation-action-request-apply",
+    ),
     re_path(
         r"^(?P<organization_id_or_slug>[^/]+)/investigations/(?P<investigation_id>[^/]+)/presence/$",
         OrganizationInvestigationPresenceEndpoint.as_view(),
diff --git a/src/sentry/apidocs/parameters.py b/src/sentry/apidocs/parameters.py
index 5be67561fdd7..62d972213b39 100644
--- a/src/sentry/apidocs/parameters.py
+++ b/src/sentry/apidocs/parameters.py
@@ -1536,6 +1536,13 @@ class InvestigationParams:
         type=str,
         description="The ID of an investigation in this organization.",
     )
+    ACTION_REQUEST_ID = OpenApiParameter(
+        name="action_request_id",
+        location="path",
+        required=True,
+        type=str,
+        description="The ID of an action request in this investigation.",
+    )
     CURSOR = OpenApiParameter(
         name="cursor",
         location="query",
diff --git a/src/sentry/investigations/endpoints/organization_investigation_action_request_apply.py b/src/sentry/investigations/endpoints/organization_investigation_action_request_apply.py
new file mode 100644
index 000000000000..effe8723deac
--- /dev/null
+++ b/src/sentry/investigations/endpoints/organization_investigation_action_request_apply.py
@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+from drf_spectacular.utils import extend_schema
+from rest_framework import status
+from rest_framework.request import Request
+from rest_framework.response import Response
+
+from sentry.api.api_owners import ApiOwner
+from sentry.api.api_publish_status import ApiPublishStatus
+from sentry.api.base import cell_silo_endpoint
+from sentry.api.serializers import serialize
+from sentry.apidocs import constants
+from sentry.apidocs.parameters import GlobalParams, InvestigationParams
+from sentry.apidocs.utils import inline_sentry_response_serializer
+from sentry.investigations.endpoints.base import (
+    OrganizationInvestigationEndpoint,
+    require_authenticated_user,
+)
+from sentry.investigations.endpoints.serializers import InvestigationActionRequestSerializer
+from sentry.investigations.endpoints.serializers.responses import (
+    InvestigationActionRequestApplyResponse,
+)
+from sentry.investigations.endpoints.validators import ActionRequestApplyValidator
+from sentry.investigations.models import Investigation
+from sentry.investigations.services.action_kinds.base import (
+    ApplyErrorCode,
+    ApplyFailed,
+    CannotApplyReason,
+)
+from sentry.investigations.services.action_requests import apply_action_request
+from sentry.models.organization import Organization
+
+_STATUS = {
+    CannotApplyReason.INVESTIGATION_ARCHIVED: status.HTTP_400_BAD_REQUEST,
+    CannotApplyReason.UNSUPPORTED_ACTION: status.HTTP_400_BAD_REQUEST,
+    ApplyErrorCode.VALIDATION_FAILED: status.HTTP_400_BAD_REQUEST,
+    CannotApplyReason.NO_ACCESS: status.HTTP_403_FORBIDDEN,
+    CannotApplyReason.PERMISSION_DENIED: status.HTTP_403_FORBIDDEN,
+}
+
+
+@extend_schema(tags=["Investigations"])
+@cell_silo_endpoint
+class OrganizationInvestigationActionRequestApplyEndpoint(OrganizationInvestigationEndpoint):
+    owner = ApiOwner.ML_AI
+    publish_status = {"POST": ApiPublishStatus.PRIVATE}
+
+    @extend_schema(
+        operation_id="Apply an Investigation Action Request",
+        parameters=[
+            GlobalParams.ORG_ID_OR_SLUG,
+            InvestigationParams.INVESTIGATION_ID,
+            InvestigationParams.ACTION_REQUEST_ID,
+        ],
+        request=ActionRequestApplyValidator,
+        responses={
+            200: inline_sentry_response_serializer(
+                "InvestigationActionRequestApplyResponse",
+                InvestigationActionRequestApplyResponse,
+            ),
+            400: constants.RESPONSE_BAD_REQUEST,
+            401: constants.RESPONSE_UNAUTHORIZED,
+            403: constants.RESPONSE_FORBIDDEN,
+            404: constants.RESPONSE_NOT_FOUND,
+            409: constants.RESPONSE_CONFLICT,
+        },
+    )
+    def post(
+        self,
+        request: Request,
+        organization: Organization,
+        investigation: Investigation,
+        action_request_id: str,
+    ) -> Response:
+        """
+        Apply a suggested action request as the current user. With `dryRun`, it runs the same
+        checks and changes nothing.
+
+        Errors about the request have a `code`. It's a 409 when the request was applied,
+        withdrawn or replaced, or its kind refuses it, and a 403 when the viewer can't apply it.
+        """
+        actor_id = require_authenticated_user(request)
+        validator = ActionRequestApplyValidator(data=request.data)
+        if not validator.is_valid():
+            return Response(validator.errors, status=status.HTTP_400_BAD_REQUEST)
+        try:
+            action_request = apply_action_request(
+                investigation=investigation,
+                action_request_id=action_request_id,
+                request=request,
+                actor_id=actor_id,
+                dry_run=validator.validated_data["dry_run"],
+            )
+        except ApplyFailed as failed:
+            return Response(failed.error, status=_STATUS.get(failed.error["code"], 409))
+        body: InvestigationActionRequestApplyResponse = {
+            "actionRequest": serialize(
+                action_request,
+                request.user,
+                InvestigationActionRequestSerializer(access=request.access),
+            ),
+        }
+        return Response(body)
diff --git a/src/sentry/investigations/endpoints/serializers/action_request.py b/src/sentry/investigations/endpoints/serializers/action_request.py
index 112039ba5f7b..722344cd5c5a 100644
--- a/src/sentry/investigations/endpoints/serializers/action_request.py
+++ b/src/sentry/investigations/endpoints/serializers/action_request.py
@@ -3,7 +3,7 @@
 from collections import defaultdict
 from collections.abc import Mapping, MutableMapping, Sequence
 from datetime import datetime
-from typing import Any, TypedDict, override
+from typing import Any, TypedDict, cast, override
 
 from django.contrib.auth.models import AnonymousUser
 
@@ -13,12 +13,13 @@
     Investigation,
     InvestigationActionRequest,
     InvestigationActionRequestAnchorType,
-    InvestigationActionRequestState,
     InvestigationBlock,
-    InvestigationStatus,
 )
-from sentry.investigations.services.action_kinds import get_action_kind
-from sentry.investigations.services.action_kinds.base import CannotApplyReason
+from sentry.investigations.services.action_kinds import (
+    get_action_kind,
+    shared_cannot_apply_reason,
+)
+from sentry.investigations.services.action_kinds.base import ActionRequestError, CannotApplyReason
 from sentry.users.api.serializers.user import UserSerializerResponse
 from sentry.users.models.user import User
 from sentry.users.services.user.model import RpcUser
@@ -61,7 +62,7 @@ class InvestigationActionRequestSerializerResponse(TypedDict):
     appliedBy: UserSerializerResponse | None
     appliedDetails: dict[str, Any] | None
     dateApplied: datetime | None
-    error: dict[str, Any] | None
+    error: ActionRequestError | None
     dateCreated: datetime
 
 
@@ -99,12 +100,9 @@ def _cannot_apply_reasons(
         reasons: dict[int, CannotApplyReason | None] = {}
         by_kind: dict[str, list[InvestigationActionRequest]] = defaultdict(list)
         for item in item_list:
-            if investigations[item.investigation_id].status == InvestigationStatus.ARCHIVED:
-                reasons[item.id] = CannotApplyReason.INVESTIGATION_ARCHIVED
-            elif item.state != InvestigationActionRequestState.SUGGESTED:
-                reasons[item.id] = CannotApplyReason.NOT_SUGGESTED
-            elif get_action_kind(item.action_kind) is None:
-                reasons[item.id] = CannotApplyReason.UNSUPPORTED_ACTION
+            reason = shared_cannot_apply_reason(investigations[item.investigation_id], item)
+            if reason is not None:
+                reasons[item.id] = reason
             elif access is None:
                 reasons[item.id] = CannotApplyReason.NO_ACCESS
             else:
@@ -203,6 +201,7 @@ def serialize(
             "appliedBy": attrs["applied_by"],
             "appliedDetails": obj.applied_details,
             "dateApplied": obj.date_applied,
-            "error": obj.error,
+            # Only the shared apply path writes this column.
+            "error": cast(ActionRequestError | None, obj.error),
             "dateCreated": obj.date_added,
         }
diff --git a/src/sentry/investigations/endpoints/serializers/responses.py b/src/sentry/investigations/endpoints/serializers/responses.py
index 8f12171bc023..bb5095ff926c 100644
--- a/src/sentry/investigations/endpoints/serializers/responses.py
+++ b/src/sentry/investigations/endpoints/serializers/responses.py
@@ -1,6 +1,9 @@
 from datetime import datetime
 from typing import Any, NotRequired, TypedDict
 
+from sentry.investigations.endpoints.serializers.action_request import (
+    InvestigationActionRequestSerializerResponse,
+)
 from sentry.investigations.endpoints.serializers.investigation import (
     InvestigationOrchestrationSerializerResponse,
 )
@@ -258,3 +261,7 @@ class InvestigationPresenceResponse(TypedDict):
 class InvestigationTitlePreviewResponse(TypedDict):
     status: str | None
     preview: str | None
+
+
+class InvestigationActionRequestApplyResponse(TypedDict):
+    actionRequest: InvestigationActionRequestSerializerResponse
diff --git a/src/sentry/investigations/endpoints/validators/__init__.py b/src/sentry/investigations/endpoints/validators/__init__.py
index c69812b7a196..cdaafe722f38 100644
--- a/src/sentry/investigations/endpoints/validators/__init__.py
+++ b/src/sentry/investigations/endpoints/validators/__init__.py
@@ -1,4 +1,5 @@
 __all__ = (
+    "ActionRequestApplyValidator",
     "BlockCreateValidator",
     "BlockDeleteValidator",
     "BlockExecutionResumeValidator",
@@ -20,6 +21,7 @@
 )
 
 
+from .action_request import ActionRequestApplyValidator
 from .base import StrictCamelSnakeValidator
 from .block import (
     BlockCreateValidator,
diff --git a/src/sentry/investigations/endpoints/validators/action_request.py b/src/sentry/investigations/endpoints/validators/action_request.py
new file mode 100644
index 000000000000..138c3f481689
--- /dev/null
+++ b/src/sentry/investigations/endpoints/validators/action_request.py
@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from rest_framework import serializers
+
+from sentry.investigations.endpoints.validators.base import StrictCamelSnakeValidator
+
+
+class ActionRequestApplyValidator(StrictCamelSnakeValidator):
+    dry_run = serializers.BooleanField(
+        default=False,
+        help_text="Run every check without applying, to see whether the apply would work.",
+    )
diff --git a/src/sentry/investigations/services/action_kinds/__init__.py b/src/sentry/investigations/services/action_kinds/__init__.py
index 7fa4136191ba..c081123fa7e5 100644
--- a/src/sentry/investigations/services/action_kinds/__init__.py
+++ b/src/sentry/investigations/services/action_kinds/__init__.py
@@ -1,6 +1,12 @@
 from __future__ import annotations
 
-from sentry.investigations.services.action_kinds.base import ActionKind
+from sentry.investigations.models import (
+    Investigation,
+    InvestigationActionRequest,
+    InvestigationActionRequestState,
+    InvestigationStatus,
+)
+from sentry.investigations.services.action_kinds.base import ActionKind, CannotApplyReason
 from sentry.investigations.services.action_kinds.connect_telemetry import ConnectTelemetryKind
 from sentry.investigations.services.action_kinds.update_detector import UpdateDetectorKind
 
@@ -11,3 +17,16 @@
 
 def get_action_kind(kind: str | None) -> ActionKind | None:
     return ACTION_KINDS.get(kind) if kind is not None else None
+
+
+def shared_cannot_apply_reason(
+    investigation: Investigation, action_request: InvestigationActionRequest
+) -> CannotApplyReason | None:
+    """Why no one can apply this request, whatever its kind, or None if its kind decides."""
+    if investigation.status == InvestigationStatus.ARCHIVED:
+        return CannotApplyReason.INVESTIGATION_ARCHIVED
+    if action_request.state != InvestigationActionRequestState.SUGGESTED:
+        return CannotApplyReason.NOT_SUGGESTED
+    if get_action_kind(action_request.action_kind) is None:
+        return CannotApplyReason.UNSUPPORTED_ACTION
+    return None
diff --git a/src/sentry/investigations/services/action_kinds/base.py b/src/sentry/investigations/services/action_kinds/base.py
index 55efc2c27e5a..2f2539a45a44 100644
--- a/src/sentry/investigations/services/action_kinds/base.py
+++ b/src/sentry/investigations/services/action_kinds/base.py
@@ -1,14 +1,21 @@
 from __future__ import annotations
 
 import abc
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
 from enum import StrEnum
+from typing import NotRequired, TypedDict
 
 from django.contrib.auth.models import AnonymousUser
+from rest_framework.request import Request
 
 from sentry.auth.access import Access
 from sentry.investigations.contracts import ActionPayloadSerializer
-from sentry.investigations.models import Investigation, InvestigationActionRequest
+from sentry.investigations.models import (
+    Investigation,
+    InvestigationActionRequest,
+    InvestigationOrchestrationRun,
+)
 from sentry.users.models.user import User
 from sentry.users.services.user.model import RpcUser
 
@@ -26,6 +33,71 @@ class CannotApplyReason(StrEnum):
     DYNAMIC_DETECTOR = "dynamic_detector"
 
 
+# One setting an action reads or changes, as JSON.
+SettingValue = int | float | str | None
+
+
+class ApplyErrorCode(StrEnum):
+    """What else the apply endpoint can say went wrong. The UI switches on these."""
+
+    ALREADY_APPLIED = "already_applied"
+    WITHDRAWN = "withdrawn"
+    STALE = "stale"
+    VALIDATION_FAILED = "validation_failed"
+
+
+class ApplyError(TypedDict):
+    detail: str
+    code: CannotApplyReason | ApplyErrorCode
+    # Seer's reason for withdrawing it.
+    reason: NotRequired[str | None]
+    # The settings that moved since Seer suggested the change.
+    current: NotRequired[dict[str, SettingValue]]
+    # Why the change wouldn't save.
+    messages: NotRequired[list[str]]
+
+
+class ActionRequestError(TypedDict):
+    """What's stored on a request that went wrong, and shown with it."""
+
+    code: ApplyErrorCode
+    current: NotRequired[dict[str, SettingValue]]
+    messages: NotRequired[list[str]]
+
+
+class ApplyFailed(Exception):
+    def __init__(self, error: ApplyError) -> None:
+        super().__init__(error["detail"])
+        self.error = error
+
+
+@dataclass(frozen=True)
+class ApplyContext:
+    """When applying, the caller holds the investigation, run, and request row locks."""
+
+    investigation: Investigation
+    run: InvestigationOrchestrationRun
+    action_request: InvestigationActionRequest
+    request: Request
+    actor_id: int
+
+
+class ActionStale(Exception):
+    """What the action changes moved since Seer suggested it."""
+
+    def __init__(self, current: dict[str, SettingValue]) -> None:
+        super().__init__("The action is stale.")
+        self.current = current
+
+
+class ActionInvalid(Exception):
+    """The change wouldn't save, and why."""
+
+    def __init__(self, messages: list[str]) -> None:
+        super().__init__("The action is invalid.")
+        self.messages = messages
+
+
 class ActionKind(abc.ABC):
     """Everything Sentry needs to check, apply, and record one kind of action."""
 
@@ -43,6 +115,18 @@ def cannot_apply_reasons(
         """For each suggested request, why the viewer can't apply it, or None if they can."""
         return {item.id: CannotApplyReason.UNSUPPORTED_ACTION for item in action_requests}
 
+    def check(self, context: ApplyContext) -> None:
+        """Run the checks apply would, without changing anything. Raise ActionStale or
+        ActionInvalid where apply would. Kinds with nothing to check beyond
+        cannot_apply_reasons can leave this as is.
+        """
+
+    def apply(self, context: ApplyContext) -> Mapping[str, object]:
+        """Make the change, and return what the activity feed should show about it. Raise
+        ActionStale or ActionInvalid instead of updating the request; the caller records it.
+        """
+        raise NotImplementedError
+
     def after_upsert(self, action_request: InvestigationActionRequest) -> None:
         """Runs when Seer creates or changes a request, in the same transaction. Only requests
         that are still suggested once the whole event is handled get it.
diff --git a/src/sentry/investigations/services/action_kinds/update_detector.py b/src/sentry/investigations/services/action_kinds/update_detector.py
index 555e105e6341..b254427b41b4 100644
--- a/src/sentry/investigations/services/action_kinds/update_detector.py
+++ b/src/sentry/investigations/services/action_kinds/update_detector.py
@@ -27,7 +27,11 @@
     InvestigationActionRequest,
     InvestigationOrchestrationRun,
 )
-from sentry.investigations.services.action_kinds.base import ActionKind, CannotApplyReason
+from sentry.investigations.services.action_kinds.base import (
+    ActionKind,
+    CannotApplyReason,
+    SettingValue,
+)
 from sentry.models.organization import Organization
 from sentry.snuba.models import SnubaQuery
 from sentry.users.models.user import User
@@ -124,8 +128,6 @@ def trim(cls, action: dict[str, Any]) -> dict[str, Any] | None:
         }
 
 
-DetectorValue = int | float | str | None
-
 _THRESHOLD_RESULTS: dict[
     Literal["highThreshold", "mediumThreshold", "resolutionThreshold"], DetectorPriorityLevel
 ] = {
@@ -211,12 +213,12 @@ def load(cls, detector: Detector, snuba_query: SnubaQuery) -> DetectorState:
         )
         return cls(data_source=data_source, condition_group=condition_group)
 
-    def values(self) -> dict[str, DetectorValue]:
+    def values(self) -> dict[str, SettingValue]:
         """The monitor's current settings, keyed like Seer's changes. A threshold is left out
         unless exactly one condition sets it.
         """
 
-        values: dict[str, DetectorValue] = {
+        values: dict[str, SettingValue] = {
             "timeWindowSeconds": self.data_source["timeWindow"],
             "query": self.data_source["query"],
             "aggregate": self.data_source["aggregate"],
diff --git a/src/sentry/investigations/services/action_requests.py b/src/sentry/investigations/services/action_requests.py
index 3f747c2ffe4d..2c15eefcb20d 100644
--- a/src/sentry/investigations/services/action_requests.py
+++ b/src/sentry/investigations/services/action_requests.py
@@ -3,9 +3,12 @@
 import logging
 from collections.abc import Sequence
 
+from django.db import router, transaction
 from django.db.models import Q, QuerySet
 from django.utils import timezone
+from rest_framework.request import Request
 
+from sentry.api.exceptions import ResourceDoesNotExist
 from sentry.investigations.contracts import ActionRequestItem
 from sentry.investigations.models import (
     Investigation,
@@ -14,7 +17,18 @@
     InvestigationActionRequestState,
     InvestigationOrchestrationRun,
 )
-from sentry.investigations.services.action_kinds import get_action_kind
+from sentry.investigations.services.action_kinds import get_action_kind, shared_cannot_apply_reason
+from sentry.investigations.services.action_kinds.base import (
+    ActionInvalid,
+    ActionKind,
+    ActionRequestError,
+    ActionStale,
+    ApplyContext,
+    ApplyError,
+    ApplyErrorCode,
+    ApplyFailed,
+    CannotApplyReason,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -214,3 +228,159 @@ def supersede_block_action_requests(investigation: Investigation, block_keys: Se
             anchor_id__in=block_keys,
         )
     )
+
+
+_REASON_DETAILS = {
+    CannotApplyReason.INVESTIGATION_ARCHIVED: "Archived investigations are read-only.",
+    CannotApplyReason.UNSUPPORTED_ACTION: "This request has no action to apply.",
+}
+_DEFAULT_REASON_DETAIL = "This action can't be applied."
+
+
+def _not_suggested_error(action_request: InvestigationActionRequest) -> ApplyError:
+    if action_request.state == InvestigationActionRequestState.APPLIED:
+        return {
+            "detail": "This action was already applied.",
+            "code": ApplyErrorCode.ALREADY_APPLIED,
+        }
+    if action_request.state == InvestigationActionRequestState.WITHDRAWN:
+        return {
+            "detail": "Seer no longer suggests this.",
+            "code": ApplyErrorCode.WITHDRAWN,
+            "reason": action_request.reason,
+        }
+    return {
+        "detail": "This action can no longer be applied.",
+        "code": CannotApplyReason.NOT_SUGGESTED,
+    }
+
+
+def _check_can_apply(
+    investigation: Investigation, action_request: InvestigationActionRequest, request: Request
+) -> ActionKind:
+    """Return the request's kind, or raise ApplyFailed if the viewer can't apply it."""
+
+    reason = shared_cannot_apply_reason(investigation, action_request)
+    if reason == CannotApplyReason.NOT_SUGGESTED:
+        raise ApplyFailed(_not_suggested_error(action_request))
+    if reason is None:
+        kind = get_action_kind(action_request.action_kind)
+        assert kind is not None
+        # The same check that decides canApply guards the endpoint, so a kind can't forget it.
+        reason = kind.cannot_apply_reasons(
+            [action_request],
+            investigations={investigation.id: investigation},
+            access=request.access,
+            user=request.user,
+        )[action_request.id]
+        if reason is None:
+            return kind
+    raise ApplyFailed(
+        {"detail": _REASON_DETAILS.get(reason, _DEFAULT_REASON_DETAIL), "code": reason}
+    )
+
+
+def _apply(
+    *,
+    investigation: Investigation,
+    action_request: InvestigationActionRequest,
+    request: Request,
+    actor_id: int,
+    dry_run: bool,
+) -> InvestigationActionRequest:
+    investigations = Investigation.objects.select_related("organization")
+    runs = InvestigationOrchestrationRun.objects.all()
+    action_requests = InvestigationActionRequest.objects.all()
+    if not dry_run:
+        investigations = investigations.select_for_update(of=("self",))
+        runs = runs.select_for_update()
+        action_requests = action_requests.select_for_update()
+
+    locked = investigations.get(id=investigation.id)
+    run = runs.get(investigation=locked)
+    action_request = action_requests.get(id=action_request.id)
+    kind = _check_can_apply(locked, action_request, request)
+
+    context = ApplyContext(
+        investigation=locked,
+        run=run,
+        action_request=action_request,
+        request=request,
+        actor_id=actor_id,
+    )
+    try:
+        if dry_run:
+            kind.check(context)
+            return action_request
+        details = kind.apply(context)
+    except ActionStale as stale:
+        if not dry_run:
+            # Keep the error on the row, so Seer sending the same suggestion again can't bring
+            # it back.
+            action_request.update(
+                state=InvestigationActionRequestState.SUPERSEDED,
+                error=ActionRequestError(code=ApplyErrorCode.STALE, current=stale.current),
+            )
+        raise ApplyFailed(
+            {
+                "detail": "This changed since Seer suggested it.",
+                "code": ApplyErrorCode.STALE,
+                "current": stale.current,
+            }
+        )
+    except ActionInvalid as invalid:
+        # The request stays suggested, so the user can try again.
+        raise ApplyFailed(
+            {
+                "detail": "The change wasn't valid.",
+                "code": ApplyErrorCode.VALIDATION_FAILED,
+                "messages": invalid.messages,
+            }
+        )
+    action_request.update(
+        state=InvestigationActionRequestState.APPLIED,
+        applied_by_id=actor_id,
+        date_applied=timezone.now(),
+        applied_details=details,
+        error=None,
+    )
+    return action_request
+
+
+def apply_action_request(
+    *,
+    investigation: Investigation,
+    action_request_id: str,
+    request: Request,
+    actor_id: int,
+    dry_run: bool = False,
+) -> InvestigationActionRequest:
+    """Apply a suggested action as the requesting user, or raise ApplyFailed. A dry run only
+    says whether it would work, and doesn't take locks or change anything.
+    """
+
+    try:
+        action_request = InvestigationActionRequest.objects.get(
+            id=action_request_id, investigation=investigation
+        )
+    except (InvestigationActionRequest.DoesNotExist, ValueError):
+        raise ResourceDoesNotExist
+
+    using = router.db_for_write(InvestigationActionRequest)
+    with transaction.atomic(using=using):
+        try:
+            return _apply(
+                investigation=investigation,
+                action_request=action_request,
+                request=request,
+                actor_id=actor_id,
+                dry_run=dry_run,
+            )
+        except ApplyFailed as failed:
+            error = failed
+        finally:
+            if dry_run:
+                # The detector validator can create environments while validating.
+                transaction.set_rollback(True, using=using)
+    # Raise once the transaction commits, so a stale request stays superseded.
+    raise error
diff --git a/static/app/utils/api/knownSentryApiUrls.generated.ts b/static/app/utils/api/knownSentryApiUrls.generated.ts
index 93165daf8764..cdcd9e3b7c81 100644
--- a/static/app/utils/api/knownSentryApiUrls.generated.ts
+++ b/static/app/utils/api/knownSentryApiUrls.generated.ts
@@ -194,6 +194,7 @@ export type KnownSentryApiUrls =
   | '/organizations/$organizationIdOrSlug/intercom-jwt/'
   | '/organizations/$organizationIdOrSlug/investigations/'
   | '/organizations/$organizationIdOrSlug/investigations/$investigationId/'
+  | '/organizations/$organizationIdOrSlug/investigations/$investigationId/action-requests/$actionRequestId/apply/'
   | '/organizations/$organizationIdOrSlug/investigations/$investigationId/blocks/'
   | '/organizations/$organizationIdOrSlug/investigations/$investigationId/blocks/$blockId/'
   | '/organizations/$organizationIdOrSlug/investigations/$investigationId/blocks/$blockId/executions/'
diff --git a/tests/sentry/investigations/endpoints/test_organization_investigation_action_request_apply.py b/tests/sentry/investigations/endpoints/test_organization_investigation_action_request_apply.py
index ce3ca7d501cb..8344cd8e3e0b 100644
--- a/tests/sentry/investigations/endpoints/test_organization_investigation_action_request_apply.py
+++ b/tests/sentry/investigations/endpoints/test_organization_investigation_action_request_apply.py
@@ -11,7 +11,11 @@
 from sentry.incidents.utils.subscription_limits import METRIC_SUBSCRIPTION_FEATURE_FLAGS
 from sentry.investigations.models import (
     InvestigationActionRequest,
+    InvestigationActionRequestState,
+    InvestigationStatus,
 )
+from sentry.investigations.services.action_kinds import ACTION_KINDS
+from sentry.investigations.services.action_kinds.base import ActionInvalid, ActionStale
 from sentry.seer.models.run import SeerRunMirrorStatus, SeerRunType
 from sentry.snuba.dataset import Dataset
 from sentry.snuba.models import ExtrapolationMode, SnubaQuery, SnubaQueryEventType
@@ -22,6 +26,7 @@
 from sentry.workflow_engine.models import DataConditionGroup, Detector
 from sentry.workflow_engine.models.data_condition import Condition
 from sentry.workflow_engine.types import DetectorPriorityLevel
+from tests.sentry.investigations.test_contracts import RecordingKind, fake_action, fake_action_kind
 
 pytestmark = [requires_snuba, requires_kafka]
 
@@ -131,6 +136,23 @@ def suggest(
             investigation=self.investigation, key="widen-window", **values
         )
 
+    def apply_url(self, action_request_id: int | str, investigation_id: int | None = None) -> str:
+        return reverse(
+            "sentry-api-0-organization-investigation-action-request-apply",
+            kwargs={
+                "organization_id_or_slug": self.organization.slug,
+                "investigation_id": investigation_id or self.investigation.id,
+                "action_request_id": action_request_id,
+            },
+        )
+
+    def apply(self, action_request: InvestigationActionRequest, **data: Any) -> Any:
+        return self.client.post(self.apply_url(action_request.id), data=data, format="json")
+
+    def snuba_query(self, detector: Detector | None = None) -> SnubaQuery:
+        data_source = (detector or self.detector).data_sources.get()
+        return SnubaQuery.objects.get(subscriptions__id=int(data_source.source_id))
+
     def details_request(self) -> dict[str, Any]:
         response = self.client.get(
             reverse(
@@ -168,3 +190,203 @@ def test_a_dataset_the_plan_no_longer_allows_cannot_apply(self) -> None:
             False,
             "detector_not_found",
         )
+
+    def test_requests_that_are_no_longer_suggested_conflict(self) -> None:
+        action_request = self.suggest(state=InvestigationActionRequestState.SUPERSEDED)
+
+        response = self.apply(action_request)
+
+        assert response.status_code == 409
+        assert response.data["code"] == "not_suggested"
+
+    def test_a_dry_run_runs_the_same_checks(self) -> None:
+        action_request = self.suggest(state=InvestigationActionRequestState.SUPERSEDED)
+
+        response = self.apply(action_request, dryRun=True)
+
+        assert response.status_code == 409
+        assert response.data["code"] == "not_suggested"
+
+    def test_rejects_an_unknown_field(self) -> None:
+        response = self.apply(self.suggest(), dryRn=True)
+
+        assert response.status_code == 400
+        assert response.data == {"dryRn": "Unknown field."}
+
+    def test_archived_investigations_are_read_only(self) -> None:
+        action_request = self.suggest()
+        self.investigation.update(status=InvestigationStatus.ARCHIVED)
+
+        response = self.apply(action_request)
+
+        assert response.status_code == 400
+        assert response.data == {
+            "detail": "Archived investigations are read-only.",
+            "code": "investigation_archived",
+        }
+
+    def test_the_kind_check_that_hides_the_button_also_guards_apply(self) -> None:
+        self.organization.update_option("sentry:alerts_member_write", False)
+        member = self.create_user()
+        self.create_member(user=member, organization=self.organization, role="member")
+        self.login_as(member)
+        action_request = self.suggest()
+
+        response = self.apply(action_request)
+
+        assert response.status_code == 403
+        assert response.data["code"] == "permission_denied"
+        action_request.refresh_from_db()
+        assert action_request.state == InvestigationActionRequestState.SUGGESTED
+
+    def test_requests_without_a_supported_action_cannot_apply(self) -> None:
+        for key, action in [("no-action", None), ("task", {"kind": "seer_task", "prompt": "Go"})]:
+            action_request = self.create_investigation_action_request(
+                investigation=self.investigation,
+                key=key,
+                action_kind=action["kind"] if action else None,
+                action_payload=action,
+            )
+
+            response = self.apply(action_request)
+
+            assert response.status_code == 400
+            assert response.data == {
+                "detail": "This request has no action to apply.",
+                "code": "unsupported_action",
+            }
+
+    def test_a_withdrawn_request_is_no_longer_suggested(self) -> None:
+        action_request = self.suggest(
+            state=InvestigationActionRequestState.WITHDRAWN, reason="The spikes are real."
+        )
+
+        response = self.apply(action_request)
+
+        assert response.status_code == 409
+        assert response.data == {
+            "detail": "Seer no longer suggests this.",
+            "code": "withdrawn",
+            "reason": "The spikes are real.",
+        }
+        assert self.snuba_query().time_window == 300
+
+    def test_requests_from_other_investigations_are_not_found(self) -> None:
+        other = self.create_investigation(organization=self.organization, created_by=self.user)
+        self.create_investigation_orchestration_run(investigation=other)
+        action_request = self.suggest()
+
+        response = self.client.post(
+            self.apply_url(action_request.id, investigation_id=other.id), data={}, format="json"
+        )
+        assert response.status_code == 404
+        assert self.client.post(self.apply_url("abc"), data={}, format="json").status_code == 404
+
+    def test_requires_the_feature(self) -> None:
+        action_request = self.suggest()
+
+        with self.feature({FEATURE: False}):
+            response = self.client.post(self.apply_url(action_request.id), data={}, format="json")
+
+        assert response.status_code == 404
+
+
+@with_feature(FEATURE)
+class SharedApplyPathTest(APITestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.login_as(self.user)
+        kind = fake_action_kind()
+        kind.start()
+        self.addCleanup(kind.stop)
+        recording = ACTION_KINDS["test_kind"]
+        assert isinstance(recording, RecordingKind)
+        self.kind = recording
+        self.investigation = self.create_investigation(
+            organization=self.organization, created_by=self.user
+        )
+        self.create_investigation_orchestration_run(investigation=self.investigation)
+        self.action_request = self.create_investigation_action_request(
+            investigation=self.investigation,
+            key="test",
+            action_kind="test_kind",
+            action_payload=fake_action(),
+        )
+
+    def post(self, **data: Any) -> Any:
+        url = reverse(
+            "sentry-api-0-organization-investigation-action-request-apply",
+            kwargs={
+                "organization_id_or_slug": self.organization.slug,
+                "investigation_id": self.investigation.id,
+                "action_request_id": self.action_request.id,
+            },
+        )
+        return self.client.post(url, data=data, format="json")
+
+    def test_a_dry_run_only_checks(self) -> None:
+        response = self.post(dryRun=True)
+
+        assert response.status_code == 200, response.data
+        assert self.kind.calls == ["check"]
+        self.action_request.refresh_from_db()
+        assert self.action_request.state == InvestigationActionRequestState.SUGGESTED
+
+    def test_a_dry_run_keeps_nothing_a_check_writes(self) -> None:
+        title = self.action_request.title
+        self.kind.check_title = "Changed by the check"
+
+        response = self.post(dryRun=True)
+
+        assert response.status_code == 200, response.data
+        self.action_request.refresh_from_db()
+        assert self.action_request.title == title
+
+    def test_applying_records_what_the_kind_returns(self) -> None:
+        response = self.post()
+
+        assert response.status_code == 200, response.data
+        assert self.kind.calls == ["apply"]
+        self.action_request.refresh_from_db()
+        assert self.action_request.state == InvestigationActionRequestState.APPLIED
+        assert self.action_request.applied_by_id == self.user.id
+        assert self.action_request.applied_details == {"target": "a"}
+
+    def test_a_stale_action_is_superseded_with_what_moved(self) -> None:
+        self.kind.raises = ActionStale({"window": 600})
+
+        response = self.post()
+
+        assert response.status_code == 409
+        assert response.data == {
+            "detail": "This changed since Seer suggested it.",
+            "code": "stale",
+            "current": {"window": 600},
+        }
+        self.action_request.refresh_from_db()
+        assert self.action_request.state == InvestigationActionRequestState.SUPERSEDED
+        assert self.action_request.error == {"code": "stale", "current": {"window": 600}}
+
+    def test_a_stale_dry_run_changes_nothing(self) -> None:
+        self.kind.raises = ActionStale({"window": 600})
+
+        response = self.post(dryRun=True)
+
+        assert response.status_code == 409
+        self.action_request.refresh_from_db()
+        assert self.action_request.state == InvestigationActionRequestState.SUGGESTED
+        assert self.action_request.error is None
+
+    def test_an_invalid_action_stays_suggested(self) -> None:
+        self.kind.raises = ActionInvalid(["Pick a supported window."])
+
+        response = self.post()
+
+        assert response.status_code == 400
+        assert response.data == {
+            "detail": "The change wasn't valid.",
+            "code": "validation_failed",
+            "messages": ["Pick a supported window."],
+        }
+        self.action_request.refresh_from_db()
+        assert self.action_request.state == InvestigationActionRequestState.SUGGESTED
diff --git a/tests/sentry/investigations/test_contracts.py b/tests/sentry/investigations/test_contracts.py
index 3e29a4a2fbc3..19e7df0ca0ea 100644
--- a/tests/sentry/investigations/test_contracts.py
+++ b/tests/sentry/investigations/test_contracts.py
@@ -34,7 +34,11 @@
 )
 from sentry.investigations.models import Investigation, InvestigationActionRequest
 from sentry.investigations.services.action_kinds import ACTION_KINDS
-from sentry.investigations.services.action_kinds.base import ActionKind, CannotApplyReason
+from sentry.investigations.services.action_kinds.base import (
+    ActionKind,
+    ApplyContext,
+    CannotApplyReason,
+)
 from sentry.users.models.user import User
 from sentry.users.services.user.model import RpcUser
 from sentry.utils import json
@@ -654,6 +658,12 @@ def __init__(self) -> None:
         self.upserted: list[InvestigationActionRequest] = []
         # The arguments from each cannot_apply_reasons call.
         self.checked: list[tuple[list[int], list[int], Access, User | RpcUser | AnonymousUser]] = []
+        # Which of check and apply ran, in order.
+        self.calls: list[str] = []
+        # What check and apply raise, if anything.
+        self.raises: Exception | None = None
+        # A title for check to write on the request.
+        self.check_title: str | None = None
 
     def after_upsert(self, action_request: InvestigationActionRequest) -> None:
         self.upserted.append(action_request)
@@ -671,6 +681,21 @@ def cannot_apply_reasons(
         )
         return {item.id: None for item in action_requests}
 
+    def check(self, context: ApplyContext) -> None:
+        self.calls.append("check")
+        if self.check_title is not None:
+            context.action_request.update(title=self.check_title)
+        if self.raises is not None:
+            raise self.raises
+
+    def apply(self, context: ApplyContext) -> dict[str, object]:
+        self.calls.append("apply")
+        if self.raises is not None:
+            raise self.raises
+        payload = context.action_request.action_payload
+        assert payload is not None
+        return {"target": payload["target"]}
+
 
 def fake_action(target: str = "a") -> dict[str, Any]:
     return {"kind": "test_kind", "target": target}