Closes #22770: Allow plugins to register Event Rule action handlers (#22793)

* Closes #22770: Allow plugins to register Event Rule action handlers

Introduces an EventRuleAction registration API (netbox.event_rules /
netbox.extras.event_rules) so plugins can add new EventRule action types
the same way they already register search indexes and event types,
replacing the hardcoded webhook/script/notification elif-chain. Core's
own three action types are refactored onto this mechanism.

An EventRule referencing an unregistered action (e.g. its providing
plugin is uninstalled) remains stored, is skipped during processing
without affecting other rules, is visibly marked unavailable in the
UI/API, and triggers a new extras.W001 system check, resuming
automatically once the plugin is reinstalled, with no need to re-save.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Fix CI failure: extras.W001 check must tolerate a not-yet-migrated database

check_event_rule_actions() queried EventRule unconditionally, which broke
`manage.py makemigrations --check` (and a fresh `manage.py migrate`) on a
database with no tables yet, since Django runs system checks before
verifying/applying migrations. Wrap the query and swallow DatabaseError,
matching the existing check_postgresql_version precedent for a database
that may not be ready. Verified against a fresh, unmigrated database that
makemigrations --check, migrate, and manage.py check all behave correctly,
and that the warning still fires once a qualifying EventRule exists.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Fix EventRuleForm action_type widget: HTMXSelect was silently ignored

Meta.widgets only applies to fields the ModelForm auto-generates; action_type
is an explicit class-level field, so its HTMXSelect assignment in Meta.widgets
never took effect, and switching Action type in the browser never refreshed
the action_choice field's label/queryset. Move the widget onto the field
declaration itself.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Address review feedback from Jeremy Stretch on PR #22793

- Revert action_object_type on_delete to CASCADE (was changed to SET_NULL)
- Make action_type choices dynamic via the model field's own callable
  choices=, simplifying EventRule.clean() and making any unavailable
  action_type invalid on save, whether new or unchanged
- Rename is_action_available to action_is_available
- Fold new dispatch tests into the existing RQQueueTestMixin test class to
  fix a flaky --parallel run (two such classes cross-flush each other's
  Redis queues)
- Only catch broad exceptions around plugin-provided actions in
  process_event_rules(); let a core action's own bugs propagate
- Add value_action_type() so table exports don't leak the "unavailable"
  badge's HTML markup
- Drop the frozen CSVChoiceField on action_type and make action_object
  optional at the field level, so bulk import of an object-less action
  works
- Map ValidationErrors on unexposed model fields to NON_FIELD_ERRORS in
  bulk import instead of letting them surface as a raw ValueError
- Restore EventRuleActionEnum/the enum-based GraphQL filter, built from
  the live action registry instead of the static EventRuleActionChoices
- Default EventRuleAction.object_required to False, matching
  object_model's default of None; set it explicitly on the three core
  actions
- Drop the unused request parameter on get_object_queryset()
- Fix action_object_type's serializer queryset, which incorrectly used
  the triggering object_types' feature flag
- Use .format() instead of % in get_action_type_display()
- Remove the extras.W001 system check (a DB query on every management
  command) in favor of an action_is_available field on the REST API
- Raise ValidationError instead of a bare Exception on duplicate action
  slug registration
- Shorten a couple of overly verbose inline comments
- Split EventRuleAction.validate() into an internal _validate() and a
  public no-op validate(), so a subclass's custom validation doesn't
  need to remember to call super()

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Trim verbose comments/docstrings added while addressing review feedback

Shortened a number of overly long inline comments and test docstrings
introduced across the previous commit's review-feedback fixes (the
EventRuleActionEnum comment, the _validate()/validate() docstrings, and
several test docstrings that restated context already given elsewhere).
Also drops the auto-generated "Generated by Django" header comment from
migration 0143, matching the rest of this app's hand-touched migrations.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Address findings from automated follow-up review of #22793

- Clear stale action_object_type/action_object_id when an action declares
  object_model but is left with no object selected (object_required=False);
  previously neither branch of the if/elif fired and the old value from
  before the edit silently persisted. Fixed in both EventRuleForm and
  EventRuleImportForm (the latter matters for CSV updates of an existing
  row). Also resolve the content type from the actual selected object
  rather than the action's declared object_model, correctly handling
  subclass/proxy instances.
- Validate action slugs at registration time (format, and collision via
  enum_key() with an already-registered slug) so a bad third-party slug
  is rejected immediately instead of crashing GraphQL schema assembly at
  startup.
- Strip the dead-code label/description duplication out of
  EventRuleActionChoices.CHOICES -- nothing reads it, and it risked
  drifting from WebhookAction/ScriptAction/NotificationAction's own.
- Record whether an action is plugin-provided at registration time
  instead of introspecting its module on every dispatch; core's three
  actions now register with is_plugin_provided=False explicitly.
- Add an action_is_available filter (API + UI) so event rules with a
  now-unavailable action can still be found in bulk, now that the
  extras.W001 system check is gone.
- Update the plugin dev docs: fix the OpenTicketAction example (it was
  missing object_required=True, the exact gap the action_object fix
  above addresses), note that an unavailable rule can't be saved at all
  (not just skipped), and move an internal-only note out of the
  published class docstring.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Address findings from second automated follow-up review of #22793

- Drop the ChoiceSet base from EventRuleActionChoices. With CHOICES=(),
  the previous version made ChoiceField(choices=EventRuleActionChoices)
  -- the idiomatic pattern used for every other ChoiceSet in this
  codebase, and reachable via `from extras.choices import *` -- silently
  reject every value instead of failing at first use.
- Reject slugs containing hyphens or a leading underscore at
  registration time: a hyphenated slug (plausible, since plugin
  distribution names are conventionally hyphenated) or a slug starting
  with an underscore both currently pass validation but produce a
  GraphQL-invalid or GraphQL-reserved enum member name once sanitized,
  crashing schema assembly at startup. Document the constraint in the
  plugin docs and the published slug docstring.
- Raise ImproperlyConfigured instead of ValidationError for all three
  registration-failure cases in register_event_rule_action() -- these
  are packaging/configuration mistakes surfaced from AppConfig.ready(),
  not user input, matching the convention ChoiceSetMeta already uses for
  the same class of error.
- Remove is_plugin_provided's class-level default; nothing reads it
  before an action is registered in any real code path, and the default
  masked a class-vs-instance inconsistency. Move its documentation out
  of the published Attributes docstring into a plain comment.
- Simplify EventRuleImportForm.clean()'s action_object_type/id
  assignment to match EventRuleForm.clean()'s approach (set both fields
  once, unconditionally, from the resolved object) rather than assigning
  via the GFK setter and then conditionally overwriting the content
  type.
- Split a dense doc sentence in eventrule.md onto its own line.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Address findings from third automated follow-up review of #22793

- Restore is_plugin_provided's class-level default of True. Its only
  read is inside process_event_rules()'s exception handler; without a
  default, an action reaching dispatch without going through
  register_event_rule_action() (e.g. inserted into the registry dict
  directly) raised AttributeError while already handling the real
  exception, masking it entirely instead of degrading gracefully.
- Move the slug/label presence checks out of __init_subclass__ (which
  fired at class-definition time, raising TypeError) and into
  register_event_rule_action() as ImproperlyConfigured, unifying them
  with the other three registration-time checks. This also resolves a
  still-open item from the very first automated review: an intermediate
  base class shared by several concrete plugin actions couldn't
  previously be defined without a placeholder slug/label of its own.
- Restore the GFK assignment (self.instance.action_object = obj) in
  EventRuleImportForm.clean() alongside the explicit content-type
  assignment, so EventRule.clean()'s later access to action_object hits
  the descriptor cache instead of an extra SELECT per imported row.
- Clarify the slug docstring/docs wording (leading underscore
  specifically, not underscores in general; tell authors to use an
  underscore instead of a hyphen) and document that intermediate base
  classes are now supported. Add a test for an uppercase slug.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

* Document the frozen-at-import tradeoff on EventRuleSerializer.action_type

The EventRuleActionAPITestCase docstring in test_api.py pointed here for
an explanation of why the choices are materialized once at module-import
time rather than dynamically, but the field itself had no such comment.

* Address findings from fourth automated follow-up review of #22793

- Restore the "must start with a letter" slug constraint dropped from
  the docs page and class docstring by the previous round's rewording;
  reword to "must begin with a lowercase letter", which covers the
  leading-digit case SLUG_RE actually rejects and matches the
  ImproperlyConfigured message an author will hit.
- Scope the GFK-cache-priming comment in EventRuleImportForm.clean() to
  the non-proxy case it actually holds for, rather than claiming it
  unconditionally.
- Trim comments and docstrings that had regrown into reviewer-facing
  rationale (why a prior finding was reverted, why a check isn't in
  __init_subclass__ anymore rather than API documentation, in
  event_rules.py and test_event_rules.py.
EOF
)

* Misc cleanup

* Misc cleanup

---------

Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
Co-authored-by: Jeremy Stretch <jstretch@netboxlabs.com>

docs/models/extras/eventrule.md

@@ -47,6 +47,11 @@ The type of action to take when the rule triggers. This must be one of the follo
 * Custom script
 * Notification
 
+!!! tip "Custom Action Types"
+    The above list includes only built-in action types. NetBox plugins can also [register their own custom action types](../../plugins/development/event-rule-actions.md).
+
+    If the plugin providing an event rule's action type is uninstalled or disabled, the event rule is not deleted, but it is marked as unavailable and will not run. It also cannot be saved -- even to edit an unrelated field -- until either the plugin is reinstalled or the action type is changed to a currently-available one.
+
 ### Action Data
 
 An optional dictionary of JSON data to pass when executing the rule. This can be useful to include additional context data, e.g. when transmitting a webhook.

docs/plugins/development/event-rule-actions.md

@@ -0,0 +1,62 @@
+# Event Rule Actions
+
+[Event rules](../../models/extras/eventrule.md) dispatch to an *action* when a matching event occurs, such as sending a webhook request or running a script. Plugins can register their own action types to extend the list of actions an event rule can perform, by subclassing NetBox's `EventRuleAction` class.
+
+```python title="event_rules.py"
+from django.utils.translation import gettext_lazy as _
+from netbox.event_rules import EventRuleAction
+
+from .models import Ticket
+
+class OpenTicketAction(EventRuleAction):
+    slug = 'my_plugin.open_ticket'
+    label = _('Open ticket')
+    description = _('Open a ticket in the external ticketing system')
+    object_model = Ticket
+    object_required = True
+
+    def enqueue(self, *, event_rule, event_context, action_object, action_data):
+        ...
+```
+
+To register one or more event rule actions with NetBox, define a list named `event_rule_actions` at the end of this file:
+
+```python title="event_rules.py"
+event_rule_actions = [OpenTicketAction]
+```
+
+!!! tip
+    The path to the list of event rule actions can be modified by setting `event_rule_actions` in the PluginConfig instance.
+
+A dotted namespace prefix (e.g. `my_plugin.open_ticket`) is strongly recommended for `slug` to avoid collisions with other plugins or with action types added to NetBox core in the future.
+
+`slug` must begin with a lowercase letter, and may contain only letters, digits, underscores, and dot-separated segments thereafter. **Hyphens are not allowed**, even though they're common in plugin/package names -- use an underscore instead, e.g. `my_plugin.open_ticket` as in the example above. `register_event_rule_action()` raises `ImproperlyConfigured` immediately for a slug outside this pattern, rather than allowing it to fail later during GraphQL schema assembly.
+
+`slug`/`label` are only required at registration time, not at class definition, so an intermediate base class shared by several concrete actions may leave them unset.
+
+!!! warning "Actions must be stateless"
+    Registration instantiates the class once, and that single instance serves every event rule, request, and background worker thread for the lifetime of the process. Do not stash per-event data on `self` in `enqueue()` or `validate()` -- concurrent dispatches would race over it. Everything an action needs is passed in as an argument.
+
+## Target Objects
+
+If an action operates against a specific object (e.g. a webhook targets a `Webhook` instance, and a script targets a `Script` instance), set `object_model` to the relevant model class. NetBox uses this to render the object-selection field on the event rule form and to validate the selected object's type. `object_required` defaults to `False` (matching `object_model`'s default of `None`); set it to `True` alongside `object_model` if the target object must always be selected. (Setting `object_required` *without* an `object_model` raises `ImproperlyConfigured` at registration, as it could never be satisfied.) Override `get_object_queryset()` to customize which objects are eligible for selection (e.g. to filter or further restrict the queryset).
+
+The object-selection field is labeled with `object_model`'s verbose name; set `object_label` to override it.
+
+If an action leaves `object_model` as `None`, event rules using it must not specify a target object: supplying one is rejected as a validation error rather than being silently stored.
+
+## Bulk Import
+
+To support resolving a target object from a CSV value during bulk import of event rules, override `resolve_import_object()`. Raise `django.core.exceptions.ObjectDoesNotExist` (or a subclass) if the supplied value doesn't resolve to an object. If this method is not overridden, event rules using this action type cannot be targeted at an object via bulk import.
+
+## Unregistered Actions
+
+An event rule's `action_type` is stored as a plain string, and is not validated against the set of currently-registered actions at the database level. This means an event rule can reference an action type provided by a plugin that is later uninstalled or disabled, without the row being deleted or corrupted. While its action type is unavailable:
+
+* The event rule is skipped during event processing (it does not raise an error, and does not prevent other event rules from being processed).
+* It is displayed with an "unavailable" indicator in the UI. `action_is_available` is exposed as a read-only field via the REST API, and as a filter (`?action_is_available=false`), so affected event rules can be found in bulk.
+* It cannot be saved via the UI or REST API -- even to edit an unrelated field -- until its `action_type` is changed to a currently-registered value.
+
+Reinstalling the plugin (and thereby re-registering the action type) automatically restores the event rule to working order, with no need to re-save it.
+
+::: netbox.event_rules.EventRuleAction

docs/plugins/development/index.md

@@ -118,6 +118,7 @@ NetBox looks for the `config` variable within a plugin's `__init__.py` to load i
 | `events_pipeline`     | A list of handlers to add to [`EVENTS_PIPELINE`](../../configuration/miscellaneous.md#events_pipeline), identified by dotted paths |
 | `search_indexes`      | The dotted path to the list of search index classes (default: `search.indexes`)                                                    |
 | `data_backends`       | The dotted path to the list of data source backend classes (default: `data_backends.backends`)                                     |
+| `event_rule_actions`  | The dotted path to the list of event rule action classes (default: `event_rules.event_rule_actions`)                               |
 | `template_extensions` | The dotted path to the list of template extension classes (default: `template_content.template_extensions`)                        |
 | `jinja_filters`       | The dotted path to a dict of custom Jinja filter functions for use in config templates (default: `jinja_env.filters`)              |
 | `menu`                | The dotted path to a top-level navigation menu provided by the plugin (default: `navigation.menu`)                                 |

mkdocs.yml

@@ -154,6 +154,7 @@ nav:
             - Filters & Filter Sets: 'plugins/development/filtersets.md'
             - Search: 'plugins/development/search.md'
             - Event Types: 'plugins/development/event-types.md'
+            - Event Rule Actions: 'plugins/development/event-rule-actions.md'
             - Permissions: 'plugins/development/permissions.md'
             - Data Backends: 'plugins/development/data-backends.md'
             - Webhooks: 'plugins/development/webhooks.md'

netbox/extras/api/serializers_/events.py

@@ -1,9 +1,12 @@
+from rest_framework import serializers
+
 from core.models import ObjectType
 from extras.choices import *
 from extras.models import EventRule, Webhook
 from netbox.api.fields import ChoiceField, ContentTypeField
 from netbox.api.gfk_fields import GFKSerializerField
 from netbox.api.serializers import NetBoxModelSerializer
+from netbox.event_rules import get_event_rule_action_choices
 from users.api.serializers_.mixins import OwnerMixin
 
 __all__ = (
@@ -21,21 +24,34 @@ class EventRuleSerializer(OwnerMixin, NetBoxModelSerializer):
         queryset=ObjectType.objects.with_feature('event_rules'),
         many=True
     )
-    action_type = ChoiceField(choices=EventRuleActionChoices)
+    action_type = ChoiceField(choices=[])  # Choices are set by get_fields()
     action_object_type = ContentTypeField(
-        queryset=ObjectType.objects.with_feature('event_rules'),
+        queryset=ObjectType.objects.all(),
+        required=False,
+        allow_null=True,
     )
     action_object = GFKSerializerField(read_only=True)
+    action_is_available = serializers.BooleanField(read_only=True)
 
     class Meta:
         model = EventRule
         fields = [
             'id', 'url', 'display_url', 'display', 'object_types', 'name', 'enabled', 'event_types', 'conditions',
-            'action_type', 'action_object_type', 'action_object_id', 'action_object', 'description', 'custom_fields',
-            'owner', 'tags', 'created', 'last_updated',
+            'action_type', 'action_object_type', 'action_object_id', 'action_object', 'action_is_available',
+            'description', 'custom_fields', 'owner', 'tags', 'created', 'last_updated',
         ]
         brief_fields = ('id', 'url', 'display', 'name', 'description')
 
+    def get_fields(self):
+        fields = super().get_fields()
+
+        # Rebuild action_type from the live registry on each instantiation to ensure all registered
+        # actions are captured as choices.
+        if 'action_type' in fields:
+            fields['action_type'] = ChoiceField(choices=get_event_rule_action_choices())
+
+        return fields
+
 
 #
 # Webhooks

netbox/extras/apps.py

@@ -5,9 +5,16 @@ class ExtrasConfig(AppConfig):
     name = "extras"
 
     def ready(self):
+        from netbox.event_rules import register_event_rule_action
         from netbox.models.features import register_models
 
         from . import dashboard, lookups, search, signals  # noqa: F401
+        from .event_rules import NotificationAction, ScriptAction, WebhookAction
 
         # Register models
         register_models(*self.get_models())
+
+        # Register core event rule actions
+        register_event_rule_action(WebhookAction, is_plugin_provided=False)
+        register_event_rule_action(ScriptAction, is_plugin_provided=False)
+        register_event_rule_action(NotificationAction, is_plugin_provided=False)

netbox/extras/choices.py

@@ -276,18 +276,14 @@ class DashboardWidgetColorChoices(ChoiceSet):
 # Event Rules
 #
 
-class EventRuleActionChoices(ChoiceSet):
+class EventRuleActionChoices:
+    """
+    The slugs of NetBox's built-in event rule actions. Not a ChoiceSet: the full, current set of
+    valid action_type values is plugin-extensible and lives in the netbox.event_rules registry,
+    not here -- see get_event_rule_action_choices(). Use these constants for the three built-in
+    actions; do not pass this class itself as a Django/DRF field's `choices=`.
+    """
 
     WEBHOOK = 'webhook'
     SCRIPT = 'script'
     NOTIFICATION = 'notification'
-
-    CHOICES = (
-        Choice(WEBHOOK, _('Webhook'), description=_('Send an outgoing HTTP request to a remote endpoint')),
-        Choice(SCRIPT, _('Script'), description=_('Execute a custom script')),
-        Choice(
-            NOTIFICATION,
-            _('Notification'),
-            description=_('Generate a notification for one or more users or groups')
-        ),
-    )

netbox/extras/event_rules.py

@@ -0,0 +1,106 @@
+from django.utils import timezone
+from django.utils.translation import gettext_lazy as _
+from django_rq import get_queue
+
+from netbox.config import get_config
+from netbox.constants import RQ_QUEUE_DEFAULT
+from netbox.event_rules import EventRuleAction
+from utilities.request import copy_safe_request
+from utilities.rqworker import get_rq_retry
+
+from .choices import EventRuleActionChoices
+from .models import NotificationGroup, Script, Webhook
+
+__all__ = (
+    'NotificationAction',
+    'ScriptAction',
+    'WebhookAction',
+)
+
+
+class WebhookAction(EventRuleAction):
+    slug = EventRuleActionChoices.WEBHOOK
+    label = _('Webhook')
+    description = _('Send an outgoing HTTP request to a remote endpoint')
+    object_model = Webhook
+    object_required = True
+
+    def enqueue(self, *, event_rule, event_context, action_object, action_data):
+        # Select the appropriate RQ queue
+        queue_name = get_config().QUEUE_MAPPINGS.get('webhook', RQ_QUEUE_DEFAULT)
+        rq_queue = get_queue(queue_name)
+
+        # Compile the task parameters
+        params = {
+            'event_rule': event_rule,
+            'object_type': event_context['object_type'],
+            'event_type': event_context['event_type'],
+            'data': action_data,
+            'snapshots': event_context.get('snapshots'),
+            'timestamp': timezone.now().isoformat(),
+            'retry': get_rq_retry(),
+        }
+        if 'request' in event_context:
+            # Exclude FILES - webhooks don't need uploaded files,
+            # which can cause pickle errors with Pillow.
+            params['request'] = copy_safe_request(event_context['request'], include_files=False)
+
+        # Enqueue the task
+        rq_queue.enqueue('extras.webhooks.send_webhook', **params)
+
+    def resolve_import_object(self, value):
+        return Webhook.objects.get(name=value)
+
+
+class ScriptAction(EventRuleAction):
+    slug = EventRuleActionChoices.SCRIPT
+    label = _('Script')
+    description = _('Execute a custom script')
+    object_model = Script
+    object_required = True
+
+    def enqueue(self, *, event_rule, event_context, action_object, action_data):
+        # Resolve the script from action parameters
+        script = action_object.python_class()
+
+        # Enqueue a Job to record the script's execution
+        from extras.jobs import ScriptJob
+
+        params = {
+            'instance': action_object,
+            'name': script.name,
+            'user': event_context['user'],
+            'data': action_data,
+        }
+        if 'snapshots' in event_context:
+            params['snapshots'] = event_context['snapshots']
+        if 'request' in event_context:
+            params['request'] = copy_safe_request(event_context['request'], include_files=False)
+
+        # Enqueue the job
+        ScriptJob.enqueue(**params)
+
+    def resolve_import_object(self, value):
+        from extras.scripts import get_module_and_script
+        module_name, script_name = value.split('.', 1)
+        return get_module_and_script(module_name, script_name)[1]
+
+
+class NotificationAction(EventRuleAction):
+    slug = EventRuleActionChoices.NOTIFICATION
+    label = _('Notification')
+    description = _('Generate a notification for one or more users or groups')
+    object_model = NotificationGroup
+    object_required = True
+
+    def enqueue(self, *, event_rule, event_context, action_object, action_data):
+        # Bulk-create notifications for all members of the notification group
+        action_object.notify(
+            object_type=event_context['object_type'],
+            object_id=action_data['id'],
+            object_repr=action_data.get('display'),
+            event_type=event_context['event_type'],
+        )
+
+    def resolve_import_object(self, value):
+        return NotificationGroup.objects.get(name=value)

netbox/extras/events.py

@@ -2,22 +2,15 @@ import logging
 from collections import UserDict, defaultdict
 
 from django.conf import settings
-from django.utils import timezone
 from django.utils.module_loading import import_string
 from django.utils.translation import gettext as _
-from django_rq import get_queue
 
 from core.events import *
 from core.models import ObjectType
-from netbox.config import get_config
-from netbox.constants import RQ_QUEUE_DEFAULT
 from netbox.models.features import has_feature
 from utilities.api import get_serializer_for_model
-from utilities.request import copy_safe_request
-from utilities.rqworker import get_rq_retry
 from utilities.serialization import serialize_object
 
-from .choices import EventRuleActionChoices
 from .models import EventRule
 
 logger = logging.getLogger('netbox.events_processor')
@@ -174,6 +167,10 @@ def process_event_rules(event_rules, object_type, event):
       that a request is always present.
     """
 
+    # Normalize object_type onto the event context so that an action's enqueue() can always read
+    # event_context['object_type']: job-lifecycle events pass it only as this parameter.
+    event['object_type'] = object_type
+
     for event_rule in event_rules:
 
         # Evaluate event rule conditions (if any).
@@ -206,67 +203,35 @@ def process_event_rules(event_rules, object_type, event):
         # Copy to avoid mutating the rule's stored action_data dict.
         event_data = {**action_data, **event['data']}
 
-        # Webhooks
-        if event_rule.action_type == EventRuleActionChoices.WEBHOOK:
-
-            # Select the appropriate RQ queue
-            queue_name = get_config().QUEUE_MAPPINGS.get('webhook', RQ_QUEUE_DEFAULT)
-            rq_queue = get_queue(queue_name)
-
-            # Compile the task parameters
-            params = {
-                'event_rule': event_rule,
-                'object_type': object_type,
-                'event_type': event['event_type'],
-                'data': event_data,
-                'snapshots': event.get('snapshots'),
-                'timestamp': timezone.now().isoformat(),
-                'retry': get_rq_retry(),
-            }
-            if 'request' in event:
-                # Exclude FILES - webhooks don't need uploaded files,
-                # which can cause pickle errors with Pillow.
-                params['request'] = copy_safe_request(event['request'], include_files=False)
-
-            # Enqueue the task
-            rq_queue.enqueue('extras.webhooks.send_webhook', **params)
-
-        # Scripts
-        elif event_rule.action_type == EventRuleActionChoices.SCRIPT:
-            # Resolve the script from action parameters
-            script = event_rule.action_object.python_class()
-
-            # Enqueue a Job to record the script's execution
-            from extras.jobs import ScriptJob
-
-            params = {
-                'instance': event_rule.action_object,
-                'name': script.name,
-                'user': event['user'],
-                'data': event_data,
-            }
-            if 'snapshots' in event:
-                params['snapshots'] = event['snapshots']
-            if 'request' in event:
-                params['request'] = copy_safe_request(event['request'], include_files=False)
-
-            # Enqueue the job
-            ScriptJob.enqueue(**params)
-
-        # Notification groups
-        elif event_rule.action_type == EventRuleActionChoices.NOTIFICATION:
-            # Bulk-create notifications for all members of the notification group
-            event_rule.action_object.notify(
-                object_type=object_type,
-                object_id=event_data['id'],
-                object_repr=event_data.get('display'),
-                event_type=event['event_type'],
+        action = event_rule.action_provider
+        if action is None:
+            # The plugin providing this action type may not be installed. Log and move on to the
+            # next rule rather than raising: one rule's unavailable action must not prevent any
+            # other rule in this batch from being processed.
+            logger.warning(
+                _('Skipping event rule "{rule}": action type "{action_type}" is not registered '
+                  '(the providing plugin may not be installed).').format(
+                    rule=event_rule, action_type=event_rule.action_type,
+                )
             )
+            continue
 
-        else:
-            raise ValueError(_("Unknown action type for an event rule: {action_type}").format(
-                action_type=event_rule.action_type
-            ))
+        try:
+            action.enqueue(
+                event_rule=event_rule,
+                event_context=event,
+                action_object=event_rule.action_object,
+                action_data=event_data,
+            )
+        except Exception:
+            # Isolate third-party bugs; a core action's own bugs should propagate instead.
+            if not action.is_plugin_provided:
+                raise
+            logger.exception(
+                _('Error processing event rule "{rule}" (action: {action_type})').format(
+                    rule=event_rule, action_type=event_rule.action_type,
+                )
+            )
 
 
 def process_event_queue(events):

netbox/extras/filtersets.py

@@ -5,6 +5,7 @@ from django.utils.translation import gettext as _
 
 from core.models import DataSource, ObjectType
 from dcim.models import DeviceRole, DeviceType, Location, Platform, Region, Site, SiteGroup
+from netbox.event_rules import get_event_rule_action_choices, get_event_rule_action_slugs
 from netbox.filtersets import BaseFilterSet, ChangeLoggedModelFilterSet, NetBoxModelFilterSet, PrimaryModelFilterSet
 from tenancy.models import Tenant, TenantGroup
 from users.filterset_mixins import OwnerFilterMixin
@@ -112,9 +113,13 @@ class EventRuleFilterSet(OwnerFilterMixin, NetBoxModelFilterSet):
         method='filter_event_type'
     )
     action_type = django_filters.MultipleChoiceFilter(
-        choices=EventRuleActionChoices,
+        choices=get_event_rule_action_choices,
         distinct=False,
     )
+    action_is_available = django_filters.BooleanFilter(
+        method='filter_action_is_available',
+        label=_('Action available'),
+    )
     action_object_type = MultiValueContentTypeFilter()
     action_object_id = MultiValueNumberFilter()
 
@@ -136,6 +141,12 @@ class EventRuleFilterSet(OwnerFilterMixin, NetBoxModelFilterSet):
     def filter_event_type(self, queryset, name, value):
         return queryset.filter(event_types__overlap=value)
 
+    def filter_action_is_available(self, queryset, name, value):
+        registered_slugs = get_event_rule_action_slugs()
+        if value:
+            return queryset.filter(action_type__in=registered_slugs)
+        return queryset.exclude(action_type__in=registered_slugs)
+
 
 @register_filterset
 class CustomFieldFilterSet(OwnerFilterMixin, ChangeLoggedModelFilterSet):

netbox/extras/forms/bulk_import.py

@@ -2,12 +2,13 @@ import re
 
 from django import forms
 from django.contrib.postgres.forms import SimpleArrayField
-from django.core.exceptions import ObjectDoesNotExist
+from django.core.exceptions import NON_FIELD_ERRORS, ObjectDoesNotExist, ValidationError
 from django.utils.translation import gettext_lazy as _
 
 from core.models import DataFile, DataSource, ObjectType
 from extras.choices import *
 from extras.models import *
+from netbox.event_rules import get_event_rule_action
 from netbox.events import get_event_type_choices
 from netbox.forms import NetBoxModelImportForm, OwnerCSVMixin, PrimaryModelImportForm
 from users.models import Group, User
@@ -271,8 +272,11 @@ class EventRuleImportForm(OwnerCSVMixin, NetBoxModelImportForm):
     )
     action_object = forms.CharField(
         label=_('Action object'),
-        required=True,
-        help_text=_('Webhook name or script as dotted path module.Class')
+        required=False,
+        help_text=_(
+            'The target object for the action, if it requires one. The expected format depends on the action type '
+            '(e.g. a webhook or notification group name, or a script as dotted path module.Class).'
+        )
     )
 
     class Meta:
@@ -287,24 +291,58 @@ class EventRuleImportForm(OwnerCSVMixin, NetBoxModelImportForm):
 
         action_object = self.cleaned_data.get('action_object')
         action_type = self.cleaned_data.get('action_type')
-        if action_object and action_type:
-            # Webhook
-            if action_type == EventRuleActionChoices.WEBHOOK:
-                try:
-                    webhook = Webhook.objects.get(name=action_object)
-                except Webhook.DoesNotExist:
-                    raise forms.ValidationError(_("Webhook {name} not found").format(name=action_object))
-                self.instance.action_object = webhook
-            # Script
-            elif action_type == EventRuleActionChoices.SCRIPT:
-                from extras.scripts import get_module_and_script
-                module_name, script_name = action_object.split('.', 1)
-                try:
-                    script = get_module_and_script(module_name, script_name)[1]
-                except ObjectDoesNotExist:
-                    raise forms.ValidationError(_("Script {name} not found").format(name=action_object))
-                self.instance.action_object = script
-                self.instance.action_object_type = ObjectType.objects.get_for_model(script, for_concrete_model=False)
+        if not action_type:
+            return
+
+        action = get_event_rule_action(action_type)
+        if action is None:
+            raise forms.ValidationError({
+                'action_type': _('"{action_type}" is not a registered action type.').format(action_type=action_type)
+            })
+
+        if not action_object:
+            if action.object_required:
+                raise forms.ValidationError({
+                    'action_object': _("This action type requires a target object."),
+                })
+            # Clear any action_object this instance previously had (relevant for a CSV row that
+            # updates an existing rule, matched by id, to a now-object-less action_type).
+            self.instance.action_object_type = None
+            self.instance.action_object_id = None
+            return
+
+        if action.object_model is None:
+            raise forms.ValidationError({
+                'action_object': _("This action type does not operate against a target object."),
+            })
+
+        try:
+            obj = action.resolve_import_object(action_object)
+        except ObjectDoesNotExist:
+            raise forms.ValidationError({
+                'action_object': _("{name} not found").format(name=action_object)
+            })
+        if obj is None:
+            raise forms.ValidationError({
+                'action_object': _("This action type does not support bulk import.")
+            })
+
+        # Assign the GFK itself (not just action_object_type/id) so EventRule.clean()'s later
+        # access to self.action_object hits the descriptor cache instead of a fresh SELECT --
+        # for a non-proxy object_model, where the concrete and non-concrete content types match.
+        self.instance.action_object = obj
+        self.instance.action_object_type = ObjectType.objects.get_for_model(obj, for_concrete_model=False)
+
+    def _update_errors(self, errors):
+        # Remap errors keyed by fields this form doesn't expose (e.g. action_object_id) to
+        # NON_FIELD_ERRORS; otherwise Django's add_error() raises ValueError instead of failing validation normally.
+        if hasattr(errors, 'error_dict'):
+            remapped = {}
+            for field, messages in errors.error_dict.items():
+                key = field if field == NON_FIELD_ERRORS or field in self.fields else NON_FIELD_ERRORS
+                remapped.setdefault(key, []).extend(messages)
+            errors = ValidationError(remapped)
+        super()._update_errors(errors)
 
 
 class TagImportForm(OwnerCSVMixin, CSVModelForm):

netbox/extras/forms/filtersets.py

@@ -5,6 +5,7 @@ from core.models import DataFile, DataSource, ObjectType
 from dcim.models import DeviceRole, DeviceType, Location, Platform, Region, Site, SiteGroup
 from extras.choices import *
 from extras.models import *
+from netbox.event_rules import get_event_rule_action_choices
 from netbox.events import get_event_type_choices
 from netbox.forms import NetBoxModelFilterSetForm, PrimaryModelFilterSetForm
 from netbox.forms.mixins import OwnerFilterMixin, SavedFiltersMixin
@@ -338,7 +339,9 @@ class EventRuleFilterForm(OwnerFilterMixin, NetBoxModelFilterSetForm):
     model = EventRule
     fieldsets = (
         FieldSet('q', 'filter_id', 'tag'),
-        FieldSet('object_type_id', 'event_type', 'action_type', 'enabled', name=_('Attributes')),
+        FieldSet(
+            'object_type_id', 'event_type', 'action_type', 'action_is_available', 'enabled', name=_('Attributes')
+        ),
         FieldSet('owner_group_id', 'owner_id', name=_('Ownership')),
     )
     object_type_id = ContentTypeMultipleChoiceField(
@@ -352,10 +355,19 @@ class EventRuleFilterForm(OwnerFilterMixin, NetBoxModelFilterSetForm):
         label=_('Event type')
     )
     action_type = forms.ChoiceField(
-        choices=add_blank_choice(EventRuleActionChoices),
+        # Wrapped in a callable so the registry is read on each access, rather than frozen at the
+        # time this module is first imported (see EventRule.action_type).
+        choices=lambda: add_blank_choice(get_event_rule_action_choices()),
         required=False,
         label=_('Action type')
     )
+    action_is_available = forms.NullBooleanField(
+        label=_('Action available'),
+        required=False,
+        widget=forms.Select(
+            choices=BOOLEAN_WITH_BLANK_CHOICES
+        )
+    )
     enabled = forms.NullBooleanField(
         label=_('Enabled'),
         required=False,

netbox/extras/forms/model_forms.py

@@ -12,6 +12,7 @@ from dcim.models import DeviceRole, DeviceType, Location, Platform, Region, Site
 from extras.choices import *
 from extras.constants import IMAGE_ATTACHMENT_IMAGE_FORMATS
 from extras.models import *
+from netbox.event_rules import get_event_rule_action, get_event_rule_action_choices
 from netbox.events import get_event_type_choices
 from netbox.forms import NetBoxModelForm, PrimaryModelForm
 from netbox.forms.mixins import ChangelogMessageMixin, OwnerMixin
@@ -601,8 +602,9 @@ class WebhookForm(OwnerMixin, NetBoxModelForm):
 class EventRuleForm(OwnerMixin, NetBoxModelForm):
     action_type = ChoiceField(
         label=_('Action type'),
-        choices=EventRuleActionChoices,
+        choices=get_event_rule_action_choices,
         initial=EventRuleActionChoices.WEBHOOK,
+        widget=HTMXSelect(hx_target_id='event-rule-action'),
     )
     object_types = ContentTypeMultipleChoiceField(
         label=_('Object types'),
@@ -640,45 +642,31 @@ class EventRuleForm(OwnerMixin, NetBoxModelForm):
         )
         widgets = {
             'conditions': forms.Textarea(attrs={'class': 'font-monospace'}),
-            'action_type': HTMXSelect(hx_target_id='event-rule-action'),
             'action_object_type': forms.HiddenInput,
             'action_object_id': forms.HiddenInput,
         }
 
-    def init_script_choice(self):
-        initial = None
-        if self.instance.action_type == EventRuleActionChoices.SCRIPT:
-            script_id = get_field_value(self, 'action_object_id')
-            initial = Script.objects.get(pk=script_id) if script_id else None
-        self.fields['action_choice'] = DynamicModelChoiceField(
-            label=_('Script'),
-            queryset=Script.objects.all(),
-            required=True,
-            initial=initial
-        )
+    def init_action_choice(self):
+        action_type = get_field_value(self, 'action_type')
+        action = get_event_rule_action(action_type)
 
-    def init_webhook_choice(self):
-        initial = None
-        if self.instance.action_type == EventRuleActionChoices.WEBHOOK:
-            webhook_id = get_field_value(self, 'action_object_id')
-            initial = Webhook.objects.get(pk=webhook_id) if webhook_id else None
-        self.fields['action_choice'] = DynamicModelChoiceField(
-            label=_('Webhook'),
-            queryset=Webhook.objects.all(),
-            required=True,
-            initial=initial
-        )
+        if action is None or action.object_model is None:
+            # Either an unregistered action_type (e.g. the providing plugin is not installed), or
+            # an action that doesn't operate on a target object at all -- no object picker needed.
+            self.fields.pop('action_choice', None)
+            return
 
-    def init_notificationgroup_choice(self):
         initial = None
-        if self.instance.action_type == EventRuleActionChoices.NOTIFICATION:
-            notificationgroup_id = get_field_value(self, 'action_object_id')
-            initial = NotificationGroup.objects.get(pk=notificationgroup_id) if notificationgroup_id else None
+        if self.instance.action_type == action_type:
+            object_id = get_field_value(self, 'action_object_id')
+            if object_id:
+                initial = action.get_object_queryset().filter(pk=object_id).first()
+
         self.fields['action_choice'] = DynamicModelChoiceField(
-            label=_('Notification group'),
-            queryset=NotificationGroup.objects.all(),
-            required=True,
-            initial=initial
+            label=action.get_object_label(),
+            queryset=action.get_object_queryset(),
+            required=action.object_required,
+            initial=initial,
         )
 
     def __init__(self, *args, **kwargs):
@@ -686,35 +674,24 @@ class EventRuleForm(OwnerMixin, NetBoxModelForm):
         self.fields['action_object_type'].required = False
         self.fields['action_object_id'].required = False
 
-        # Determine the action type
-        action_type = get_field_value(self, 'action_type')
-
-        if action_type == EventRuleActionChoices.WEBHOOK:
-            self.init_webhook_choice()
-        elif action_type == EventRuleActionChoices.SCRIPT:
-            self.init_script_choice()
-        elif action_type == EventRuleActionChoices.NOTIFICATION:
-            self.init_notificationgroup_choice()
+        self.init_action_choice()
 
     def clean(self):
         super().clean()
 
+        action = get_event_rule_action(self.cleaned_data.get('action_type'))
         action_choice = self.cleaned_data.get('action_choice')
-        # Webhook
-        if self.cleaned_data.get('action_type') == EventRuleActionChoices.WEBHOOK:
-            self.cleaned_data['action_object_type'] = ObjectType.objects.get_for_model(action_choice)
-            self.cleaned_data['action_object_id'] = action_choice.id
-        # Script
-        elif self.cleaned_data.get('action_type') == EventRuleActionChoices.SCRIPT:
+
+        if action and action.object_model and action_choice:
             self.cleaned_data['action_object_type'] = ObjectType.objects.get_for_model(
-                Script,
-                for_concrete_model=False
+                action_choice, for_concrete_model=False
             )
-            self.cleaned_data['action_object_id'] = action_choice.id
-        # Notification
-        elif self.cleaned_data.get('action_type') == EventRuleActionChoices.NOTIFICATION:
-            self.cleaned_data['action_object_type'] = ObjectType.objects.get_for_model(action_choice)
-            self.cleaned_data['action_object_id'] = action_choice.id
+            self.cleaned_data['action_object_id'] = action_choice.pk
+        elif action:
+            # A no-object action, or an optional object left unselected: store no action_object
+            self.cleaned_data['action_object_type'] = None
+            self.cleaned_data['action_object_id'] = None
+        # An unregistered action_type leaves the stored action_object untouched
 
         return self.cleaned_data
 

netbox/extras/graphql/enums.py

@@ -1,6 +1,10 @@
+import enum
+
 import strawberry
 
 from extras.choices import *
+from netbox.event_rules import get_event_rule_action_choices
+from utilities.string import enum_key
 
 __all__ = (
     'CustomFieldChoiceColorEnum',
@@ -23,6 +27,10 @@ CustomFieldTypeEnum = strawberry.enum(CustomFieldTypeChoices.as_enum(prefix='typ
 CustomFieldUIEditableEnum = strawberry.enum(CustomFieldUIEditableChoices.as_enum())
 CustomFieldUIVisibleEnum = strawberry.enum(CustomFieldUIVisibleChoices.as_enum())
 CustomLinkButtonClassEnum = strawberry.enum(CustomLinkButtonClassChoices.as_enum())
-EventRuleActionEnum = strawberry.enum(EventRuleActionChoices.as_enum())
+# Built from the event_rule_actions registry, which is fully populated by the time the schema is
+# assembled. Fixed for the process's lifetime, as any Strawberry enum is.
+EventRuleActionEnum = strawberry.enum(enum.Enum('EventRuleActionEnum', {
+    enum_key(choice.value): choice.value for choice in get_event_rule_action_choices()
+}))
 JournalEntryKindEnum = strawberry.enum(JournalEntryKindChoices.as_enum(prefix='kind'))
 WebhookHttpMethodEnum = strawberry.enum(WebhookHttpMethodChoices.as_enum())

netbox/extras/migrations/0143_event_rule_action_registry.py

@@ -0,0 +1,29 @@
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('contenttypes', '0002_remove_content_type_name'),
+        ('extras', '0142_webhook_timeout'),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name='eventrule',
+            name='action_object_type',
+            field=models.ForeignKey(
+                blank=True,
+                null=True,
+                on_delete=django.db.models.deletion.CASCADE,
+                related_name='eventrule_actions',
+                to='contenttypes.contenttype',
+            ),
+        ),
+        migrations.AlterField(
+            model_name='eventrule',
+            name='action_type',
+            field=models.CharField(default='webhook', max_length=100),
+        ),
+    ]

netbox/extras/models/models.py

@@ -21,6 +21,7 @@ from extras.models.mixins import RenderTemplateMixin
 from extras.querysets import SharedObjectQuerySet
 from extras.utils import image_upload
 from netbox.config import get_config
+from netbox.event_rules import get_event_rule_action, get_event_rule_action_choices
 from netbox.events import get_event_type_choices
 from netbox.models import ChangeLoggedModel
 from netbox.models.features import (
@@ -91,15 +92,19 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, OwnerMixin, TagsMixin,
 
     # Action to take
     action_type = models.CharField(
-        max_length=30,
-        choices=EventRuleActionChoices,
+        max_length=100,
+        # Bare callable, re-evaluated fresh on each access via Django's CallableChoiceIterator,
+        # so a plugin action registered after this module was first imported is still reflected.
+        choices=get_event_rule_action_choices,
         default=EventRuleActionChoices.WEBHOOK,
         verbose_name=_('action type')
     )
     action_object_type = models.ForeignKey(
         to='contenttypes.ContentType',
         related_name='eventrule_actions',
-        on_delete=models.CASCADE
+        on_delete=models.CASCADE,
+        blank=True,
+        null=True,
     )
     action_object_id = models.PositiveBigIntegerField(
         blank=True,
@@ -134,6 +139,26 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, OwnerMixin, TagsMixin,
     def get_absolute_url(self):
         return reverse('extras:eventrule', args=[self.pk])
 
+    @property
+    def action_provider(self):
+        """
+        Return the registered EventRuleAction instance for this rule's action_type, or None if it
+        is not currently registered (e.g. the providing plugin is not installed).
+        """
+        return get_event_rule_action(self.action_type)
+
+    @property
+    def action_is_available(self):
+        return self.action_provider is not None
+
+    def get_action_type_display(self):
+        if action := self.action_provider:
+            return action.label
+        return _('{slug} (unavailable)').format(slug=self.action_type)
+
+    def get_action_type_color(self):
+        return None if self.action_is_available else 'red'
+
     def clean(self):
         super().clean()
 
@@ -148,6 +173,11 @@ class EventRule(CustomFieldsMixin, ExportTemplatesMixin, OwnerMixin, TagsMixin,
         if self.action_data is not None and not isinstance(self.action_data, dict):
             raise ValidationError({'action_data': _('Action data must be a JSON object or null.')})
 
+        # action_type's own validity is already enforced by the field's dynamic choices= (Field.
+        # validate(), earlier in full_clean()); guard here only in case clean() ran standalone.
+        if self.action_is_available:
+            self.action_provider._validate(action_object=self.action_object, action_data=self.action_data)
+
     def eval_conditions(self, data):
         """
         Test whether the given data meets the conditions of the event rule (if any). Return True

netbox/extras/tables/tables.py

@@ -552,6 +552,19 @@ class EventRuleTable(NetBoxTable):
             'pk', 'name', 'enabled', 'action_type', 'action_object', 'object_types', 'event_types',
         )
 
+    def render_action_type(self, record):
+        # Render explicitly (rather than relying on django-tables2's built-in choices-driven
+        # get_FOO_display() auto-rendering) so an unavailable action type gets a red badge.
+        label = record.get_action_type_display()
+        if not record.action_is_available:
+            return format_html('<span class="badge text-bg-red">{}</span>', label)
+        return label
+
+    def value_action_type(self, record):
+        # Raw value for non-HTML output (e.g. CSV/table-config export), so the badge's HTML
+        # markup from render_action_type() above isn't leaked into it.
+        return record.get_action_type_display()
+
 
 class TagTable(NetBoxTable):
     name = tables.Column(

netbox/extras/tests/test_api.py

@@ -16,10 +16,13 @@ from core.choices import ManagedFileRootPathChoices
 from core.events import *
 from core.models import DataFile, DataSource, ObjectType
 from dcim.models import Device, DeviceRole, DeviceType, Location, Manufacturer, Rack, RackRole, Site
+from extras.api.serializers import EventRuleSerializer
 from extras.choices import *
 from extras.models import *
 from extras.scripts import BooleanVar, IntegerVar, StringVar
 from extras.scripts import Script as PythonClass
+from netbox.event_rules import EventRuleAction, register_event_rule_action
+from netbox.registry import registry
 from users.constants import TOKEN_PREFIX
 from users.models import Group, ObjectPermission, Token, User
 from utilities.tables import get_table_for_model
@@ -157,6 +160,129 @@ class EventRuleTestCase(APIViewTestCases.APIViewTestCase):
         ]
 
 
+class EventRuleActionAPITestCase(APITestCase):
+    """
+    REST API tests for EventRule's registry-driven action_type.
+    """
+
+    def test_create_event_rule_with_unregistered_action_type_fails(self):
+        self.add_permissions('extras.add_eventrule')
+        url = reverse('extras-api:eventrule-list')
+        data = {
+            'name': 'API Bad Action Rule',
+            'object_types': ['dcim.site'],
+            'event_types': [OBJECT_CREATED],
+            'action_type': 'this.is.not.registered',
+        }
+        response = self.client.post(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('action_type', response.data)
+
+    def test_update_unrelated_field_on_unavailable_action_rule_fails(self):
+        """PATCHing a rule with an unavailable action_type is rejected, even for an unrelated field."""
+        rule = EventRule.objects.create(
+            name='API Unavailable Rule',
+            event_types=[OBJECT_CREATED],
+            action_type='some.plugin.not_installed',
+        )
+        rule.object_types.set([ObjectType.objects.get_for_model(Site)])
+
+        self.add_permissions('extras.change_eventrule')
+        url = reverse('extras-api:eventrule-detail', kwargs={'pk': rule.pk})
+        response = self.client.patch(url, {'enabled': False}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('action_type', response.data)
+
+        rule.refresh_from_db()
+        self.assertTrue(rule.enabled)
+
+    def test_action_is_available_exposed_via_api(self):
+        """action_is_available is exposed as a read-only field, so unavailable rules can be found in bulk."""
+        available_rule = EventRule.objects.create(
+            name='API Available Rule', event_types=[OBJECT_CREATED], action_type='webhook',
+        )
+        available_rule.object_types.set([ObjectType.objects.get_for_model(Site)])
+
+        unavailable_rule = EventRule.objects.create(
+            name='API Unavailable Flag Rule',
+            event_types=[OBJECT_CREATED],
+            action_type='some.plugin.not_installed_flag_test',
+        )
+        unavailable_rule.object_types.set([ObjectType.objects.get_for_model(Site)])
+
+        self.add_permissions('extras.view_eventrule')
+        url = reverse('extras-api:eventrule-detail', kwargs={'pk': available_rule.pk})
+        response = self.client.get(url, **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertTrue(response.data['action_is_available'])
+
+        url = reverse('extras-api:eventrule-detail', kwargs={'pk': unavailable_rule.pk})
+        response = self.client.get(url, **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+        self.assertFalse(response.data['action_is_available'])
+
+    def test_create_event_rule_with_runtime_registered_action(self):
+        """An action registered after this serializer's module was imported must still be a valid action_type."""
+        class NoObjectAction(EventRuleAction):
+            slug = 'test.api_no_object_action'
+            label = 'API No-Object Action'
+
+        register_event_rule_action(NoObjectAction)
+        self.addCleanup(registry['event_rule_actions'].pop, NoObjectAction.slug, None)
+
+        self.add_permissions('extras.add_eventrule')
+        url = reverse('extras-api:eventrule-list')
+        data = {
+            'name': 'API No-Object Rule',
+            'object_types': ['dcim.site'],
+            'event_types': [OBJECT_CREATED],
+            'action_type': NoObjectAction.slug,
+        }
+        response = self.client.post(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+
+        rule = EventRule.objects.get(pk=response.data['id'])
+        self.assertEqual(rule.action_type, NoObjectAction.slug)
+        self.assertIsNone(rule.action_object_type)
+        self.assertIsNone(rule.action_object_id)
+
+    def test_create_event_rule_with_object_for_no_object_action_fails(self):
+        """An action with no object_model must reject a target object rather than storing it."""
+        class NoObjectAction(EventRuleAction):
+            slug = 'test.api_no_object_action'
+            label = 'API No-Object Action'
+
+        register_event_rule_action(NoObjectAction)
+        self.addCleanup(registry['event_rule_actions'].pop, NoObjectAction.slug, None)
+
+        site = Site.objects.create(name='Action Object Site', slug='action-object-site')
+
+        self.add_permissions('extras.add_eventrule')
+        url = reverse('extras-api:eventrule-list')
+        data = {
+            'name': 'API Bogus Object Rule',
+            'object_types': ['dcim.site'],
+            'event_types': [OBJECT_CREATED],
+            'action_type': NoObjectAction.slug,
+            'action_object_type': 'dcim.site',
+            'action_object_id': site.pk,
+        }
+        response = self.client.post(url, data, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_400_BAD_REQUEST)
+        self.assertIn('action_object_id', response.data)
+        self.assertFalse(EventRule.objects.filter(name='API Bogus Object Rule').exists())
+
+    def test_action_object_type_field_accepts_any_content_type(self):
+        """action_object_type's queryset must not be restricted to the with_feature('event_rules') set."""
+        field = EventRuleSerializer().fields['action_object_type']
+        user_ct = ObjectType.objects.get_for_model(User)
+        self.assertFalse(
+            ObjectType.objects.with_feature('event_rules').filter(pk=user_ct.pk).exists(),
+            "auth.user must not support event_rules for this test to be meaningful; pick another type.",
+        )
+        self.assertIn(user_ct, field.queryset)
+
+
 class CustomFieldTestCase(APIViewTestCases.APIViewTestCase):
     model = CustomField
     brief_fields = ['description', 'display', 'id', 'name', 'url']

netbox/extras/tests/test_event_rules.py

@@ -7,6 +7,7 @@ from unittest.mock import Mock, patch
 
 import django_rq
 from django.conf import settings
+from django.core.exceptions import ImproperlyConfigured, ValidationError
 from django.http import HttpResponse
 from django.test import RequestFactory, TestCase, tag
 from django.urls import reverse
@@ -20,12 +21,20 @@ from core.models import Job, ObjectType
 from dcim.choices import SiteStatusChoices
 from dcim.models import DeviceType, Interface, Manufacturer, Site
 from extras.choices import EventRuleActionChoices
-from extras.events import enqueue_event, flush_events, serialize_for_event
+from extras.events import enqueue_event, flush_events, process_event_rules, serialize_for_event
 from extras.models import EventRule, Script, ScriptModule, Tag, Webhook
 from extras.scripts import Script as ScriptBase
 from extras.signals import process_job_end_event_rules
 from extras.webhooks import generate_signature, send_webhook
 from netbox.context_managers import event_tracking
+from netbox.event_rules import (
+    EventRuleAction,
+    get_event_rule_action,
+    get_event_rule_action_choices,
+    register_event_rule_action,
+)
+from netbox.registry import registry
+from netbox.tests.dummy_plugin.event_rules import DummyRaisingAction
 from utilities.testing import APITestCase, create_test_device
 from utilities.testing.mixins import RQQueueTestMixin
 
@@ -882,6 +891,464 @@ class EventRuleTestCase(RQQueueTestMixin, APITestCase):
         self.assertEqual(job.kwargs['event_rule'], event_rule)
         self.assertEqual(job.kwargs['event_type'], OBJECT_UPDATED)
 
+    def test_unregistered_action_type_does_not_block_other_rules(self):
+        """
+        An unregistered action_type must not block other EventRules for the same event. Kept on
+        this class, not a separate RQQueueTestMixin one, since two such classes in different
+        `--parallel` subsuites cross-flush each other's Redis queue.
+        """
+        site_type = ObjectType.objects.get_for_model(Site)
+        webhook = Webhook.objects.create(name='Dispatch Test Webhook', payload_url='http://localhost:9000/')
+        webhook_type = ObjectType.objects.get_for_model(Webhook)
+
+        good_rule = EventRule.objects.create(
+            name='Good Rule',
+            event_types=[OBJECT_CREATED],
+            action_type=EventRuleActionChoices.WEBHOOK,
+            action_object_type=webhook_type,
+            action_object_id=webhook.pk,
+        )
+        good_rule.object_types.set([site_type])
+
+        bad_rule = EventRule.objects.create(
+            name='Bad Rule',
+            event_types=[OBJECT_CREATED],
+            action_type='someplugin.not_installed',
+        )
+        bad_rule.object_types.set([site_type])
+
+        url = reverse('dcim-api:site-list')
+        self.add_permissions('dcim.add_site')
+        with self.assertLogs('netbox.events_processor', level='WARNING') as cm:
+            response = self.client.post(
+                url, {'name': 'Dispatch Site', 'slug': 'dispatch-site'}, format='json', **self.header
+            )
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+        self.assertTrue(any('someplugin.not_installed' in message for message in cm.output))
+
+        # The good rule's webhook must still have been enqueued despite the bad rule.
+        rule_jobs = [j for j in self.queue.jobs if j.kwargs['event_rule'] == good_rule]
+        self.assertEqual(len(rule_jobs), 1)
+
+    def test_raising_enqueue_does_not_block_other_rules(self):
+        """A raising action registered as plugin-provided (the default) must not block other rules."""
+        register_event_rule_action(DummyRaisingAction)
+        self.addCleanup(registry['event_rule_actions'].pop, DummyRaisingAction.slug, None)
+
+        site_type = ObjectType.objects.get_for_model(Site)
+        webhook = Webhook.objects.create(name='Dispatch Test Webhook 2', payload_url='http://localhost:9000/')
+        webhook_type = ObjectType.objects.get_for_model(Webhook)
+
+        good_rule = EventRule.objects.create(
+            name='Good Rule 2',
+            event_types=[OBJECT_CREATED],
+            action_type=EventRuleActionChoices.WEBHOOK,
+            action_object_type=webhook_type,
+            action_object_id=webhook.pk,
+        )
+        good_rule.object_types.set([site_type])
+
+        raising_rule = EventRule.objects.create(
+            name='Raising Rule',
+            event_types=[OBJECT_CREATED],
+            action_type=DummyRaisingAction.slug,
+        )
+        raising_rule.object_types.set([site_type])
+
+        url = reverse('dcim-api:site-list')
+        self.add_permissions('dcim.add_site')
+        with self.assertLogs('netbox.events_processor', level='ERROR') as cm:
+            response = self.client.post(
+                url, {'name': 'Dispatch Site 2', 'slug': 'dispatch-site-2'}, format='json', **self.header
+            )
+        self.assertHttpStatus(response, status.HTTP_201_CREATED)
+        self.assertTrue(any('Raising Rule' in message for message in cm.output))
+
+        # The good rule's webhook must still have been enqueued despite the raising rule.
+        rule_jobs = [j for j in self.queue.jobs if j.kwargs['event_rule'] == good_rule]
+        self.assertEqual(len(rule_jobs), 1)
+
+    def test_raising_action_registered_as_non_plugin_propagates(self):
+        """A raising action registered with is_plugin_provided=False (as core actions are) must propagate."""
+        class RaisingCoreLikeAction(EventRuleAction):
+            slug = 'test.raising_core_like_action'
+            label = 'Raising Core-Like Action'
+            object_required = False
+
+            def enqueue(self, **kwargs):
+                raise RuntimeError("intentional failure for test")
+
+        register_event_rule_action(RaisingCoreLikeAction, is_plugin_provided=False)
+        self.addCleanup(registry['event_rule_actions'].pop, 'test.raising_core_like_action', None)
+
+        site_type = ObjectType.objects.get_for_model(Site)
+        rule = EventRule.objects.create(
+            name='Raising Core-Like Rule',
+            event_types=[OBJECT_CREATED],
+            action_type='test.raising_core_like_action',
+        )
+        rule.object_types.set([site_type])
+
+        with self.assertRaises(RuntimeError):
+            process_event_rules([rule], object_type=site_type, event={'data': {}, 'event_type': OBJECT_CREATED})
+
+
+class EventRuleActionRegistrationTestCase(TestCase):
+    """
+    Unit tests for the EventRuleAction registry (netbox.event_rules).
+    """
+
+    def tearDown(self):
+        super().tearDown()
+        # The registry is a global dict; test-registered actions must not leak into other tests.
+        for slug in ('test.dummy_action', 'test.duplicate_action'):
+            registry['event_rule_actions'].pop(slug, None)
+
+    def test_register_event_rule_action(self):
+        class DummyAction(EventRuleAction):
+            slug = 'test.dummy_action'
+            label = 'Dummy Action'
+            description = 'A dummy action for testing'
+
+        register_event_rule_action(DummyAction)
+
+        action = get_event_rule_action('test.dummy_action')
+        self.assertIsInstance(action, DummyAction)
+
+        choices = {choice.value: choice.label for choice in get_event_rule_action_choices()}
+        self.assertEqual(choices.get('test.dummy_action'), 'Dummy Action')
+
+    def test_register_event_rule_action_as_decorator(self):
+        @register_event_rule_action
+        class DummyAction(EventRuleAction):
+            slug = 'test.dummy_action'
+            label = 'Dummy Action'
+
+        self.assertIsInstance(get_event_rule_action('test.dummy_action'), DummyAction)
+
+    def test_duplicate_slug_raises(self):
+        class FirstAction(EventRuleAction):
+            slug = 'test.duplicate_action'
+            label = 'First'
+
+        class SecondAction(EventRuleAction):
+            slug = 'test.duplicate_action'
+            label = 'Second'
+
+        register_event_rule_action(FirstAction)
+        with self.assertRaises(ImproperlyConfigured):
+            register_event_rule_action(SecondAction)
+
+    def test_slug_starting_with_digit_rejected(self):
+        """A slug starting with a digit would sanitize into a GraphQL-invalid enum member name."""
+        class DigitSlugAction(EventRuleAction):
+            slug = '2fa.notify'
+            label = 'Digit Slug Action'
+
+        with self.assertRaises(ImproperlyConfigured):
+            register_event_rule_action(DigitSlugAction)
+        self.assertIsNone(get_event_rule_action('2fa.notify'))
+
+    def test_slug_with_hyphen_rejected(self):
+        """Hyphens are not permitted, though plugin distribution names conventionally use them."""
+        class HyphenSlugAction(EventRuleAction):
+            slug = 'my-plugin.open_ticket'
+            label = 'Hyphen Slug Action'
+
+        with self.assertRaises(ImproperlyConfigured):
+            register_event_rule_action(HyphenSlugAction)
+        self.assertIsNone(get_event_rule_action('my-plugin.open_ticket'))
+
+    def test_slug_with_leading_underscore_rejected(self):
+        """A leading underscore sanitizes into a "__"-prefixed name, which GraphQL reserves for introspection."""
+        class LeadingUnderscoreAction(EventRuleAction):
+            slug = '_internal.foo'
+            label = 'Leading Underscore Action'
+
+        with self.assertRaises(ImproperlyConfigured):
+            register_event_rule_action(LeadingUnderscoreAction)
+        self.assertIsNone(get_event_rule_action('_internal.foo'))
+
+    def test_slug_with_uppercase_rejected(self):
+        """Slugs must be lowercase, though plugin/class names conventionally are not."""
+        class UppercaseSlugAction(EventRuleAction):
+            slug = 'MyPlugin.action'
+            label = 'Uppercase Slug Action'
+
+        with self.assertRaises(ImproperlyConfigured):
+            register_event_rule_action(UppercaseSlugAction)
+        self.assertIsNone(get_event_rule_action('MyPlugin.action'))
+
+    def test_slug_enum_key_collision_rejected(self):
+        """Two distinct slugs that sanitize to the same GraphQL enum member name must not both register."""
+        class DotAction(EventRuleAction):
+            slug = 'test.collision_action'
+            label = 'Dot Action'
+
+        class UnderscoreAction(EventRuleAction):
+            slug = 'test_collision_action'
+            label = 'Underscore Action'
+
+        register_event_rule_action(DotAction)
+        self.addCleanup(registry['event_rule_actions'].pop, 'test.collision_action', None)
+        with self.assertRaises(ImproperlyConfigured):
+            register_event_rule_action(UnderscoreAction)
+        self.assertIsNone(get_event_rule_action('test_collision_action'))
+
+    def test_missing_slug_raises_at_registration(self):
+        # Class definition itself must succeed; only registration checks slug/label.
+        class NoSlugAction(EventRuleAction):
+            label = 'No Slug'
+
+        with self.assertRaises(ImproperlyConfigured):
+            register_event_rule_action(NoSlugAction)
+
+    def test_missing_label_raises_at_registration(self):
+        class NoLabelAction(EventRuleAction):
+            slug = 'test.no_label'
+
+        with self.assertRaises(ImproperlyConfigured):
+            register_event_rule_action(NoLabelAction)
+
+    def test_intermediate_base_class_without_slug_or_label_is_definable(self):
+        """
+        slug/label are checked at registration, not class definition, so several concrete actions
+        can share an intermediate base class which sets neither.
+        """
+        class PluginActionBase(EventRuleAction):
+            object_required = False
+
+            def enqueue(self, **kwargs):
+                pass
+
+        class ConcreteAction(PluginActionBase):
+            slug = 'test.intermediate_base_concrete_action'
+            label = 'Concrete Action'
+
+        register_event_rule_action(ConcreteAction)
+        self.addCleanup(registry['event_rule_actions'].pop, 'test.intermediate_base_concrete_action', None)
+
+        self.assertIsInstance(get_event_rule_action('test.intermediate_base_concrete_action'), ConcreteAction)
+
+    def test_unregistered_slug_returns_none(self):
+        self.assertIsNone(get_event_rule_action('this.does.not.exist'))
+
+    def test_core_actions_are_registered(self):
+        """WebhookAction/ScriptAction/NotificationAction are registered at app startup."""
+        core_slugs = (
+            EventRuleActionChoices.WEBHOOK, EventRuleActionChoices.SCRIPT, EventRuleActionChoices.NOTIFICATION,
+        )
+        for slug in core_slugs:
+            self.assertIsNotNone(get_event_rule_action(slug))
+
+    def test_get_object_queryset_returns_none_without_object_model(self):
+        action = EventRuleAction()
+        self.assertIsNone(action.get_object_queryset())
+
+    def test_internal_validate_requires_object_when_object_required(self):
+        action = EventRuleAction()
+        action.object_required = True
+        with self.assertRaises(ValidationError):
+            action._validate(action_object=None, action_data={})
+
+    def test_internal_validate_passes_when_object_not_required(self):
+        action = EventRuleAction()
+        action.object_required = False
+        # Must not raise
+        action._validate(action_object=None, action_data={})
+
+    def test_internal_validate_rejects_wrong_object_type(self):
+        action = EventRuleAction()
+        action.object_model = Webhook
+        action.object_required = True
+        site = Site(name='Not A Webhook')
+        with self.assertRaises(ValidationError):
+            action._validate(action_object=site, action_data={})
+
+    def test_internal_validate_rejects_object_for_action_without_object_model(self):
+        """An action which declares no object_model must reject a target object outright."""
+        action = EventRuleAction()
+        with self.assertRaises(ValidationError):
+            action._validate(action_object=Webhook(), action_data={})
+
+    def test_object_required_without_object_model_rejected_at_registration(self):
+        """object_required with no object_model could never be satisfied, so it's caught early."""
+        class ImpossibleAction(EventRuleAction):
+            slug = 'test.impossible_action'
+            label = 'Impossible Action'
+            object_required = True
+
+        with self.assertRaises(ImproperlyConfigured):
+            register_event_rule_action(ImpossibleAction)
+        self.assertIsNone(get_event_rule_action('test.impossible_action'))
+
+    def test_get_object_label_defaults_to_object_model_verbose_name(self):
+        """The object picker's label defaults to the model's verbose name, capitalized."""
+        self.assertEqual(get_event_rule_action(EventRuleActionChoices.NOTIFICATION).get_object_label(),
+                         'Notification group')
+        self.assertEqual(get_event_rule_action(EventRuleActionChoices.WEBHOOK).get_object_label(), 'Webhook')
+
+    def test_get_object_label_honors_explicit_override(self):
+        class LabeledAction(EventRuleAction):
+            object_model = Webhook
+            object_label = 'Destination'
+
+        self.assertEqual(LabeledAction().get_object_label(), 'Destination')
+
+    def test_get_object_label_is_none_without_object_model(self):
+        self.assertIsNone(EventRuleAction().get_object_label())
+
+    def test_validate_is_noop_by_default(self):
+        action = EventRuleAction()
+        # Must not raise
+        action.validate(action_object=None, action_data={})
+
+    def test_validate_override_does_not_need_super(self):
+        """A subclass overriding validate() gets the base object_required check for free, no super() needed."""
+        class CustomValidatingAction(EventRuleAction):
+            slug = 'test.custom_validating_action'
+            label = 'Custom Validating Action'
+            object_model = Webhook
+            object_required = True
+
+            def validate(self, *, action_object, action_data):
+                if action_data.get('bad'):
+                    raise ValidationError({'action_data': 'bad action_data for test'})
+
+        action = CustomValidatingAction()
+
+        # The subclass's own check fires
+        with self.assertRaises(ValidationError):
+            action._validate(action_object=Webhook(), action_data={'bad': True})
+
+        # ...as does the base object_required check
+        with self.assertRaises(ValidationError):
+            action._validate(action_object=None, action_data={})
+
+        action._validate(action_object=Webhook(), action_data={})  # must not raise
+
+    def test_enqueue_not_implemented_by_default(self):
+        action = EventRuleAction()
+        with self.assertRaises(NotImplementedError):
+            action.enqueue(event_rule=None, event_context={}, action_object=None, action_data={})
+
+    def test_is_plugin_provided_defaults_true_before_registration(self):
+        """is_plugin_provided is True on an instance that never goes through registration."""
+        self.assertTrue(EventRuleAction().is_plugin_provided)
+
+
+class EventRuleActionAvailabilityTestCase(TestCase):
+    """
+    An EventRule with an unregistered action_type must remain loadable, skip gracefully during
+    processing, and display as "unavailable" -- but reject full_clean() until action_type changes.
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        site_type = ObjectType.objects.get_for_model(Site)
+        webhook = Webhook.objects.create(name='Availability Test Webhook', payload_url='http://localhost:9000/')
+        webhook_type = ObjectType.objects.get_for_model(Webhook)
+
+        cls.healthy_rule = EventRule.objects.create(
+            name='Healthy Rule',
+            event_types=[OBJECT_CREATED],
+            action_type=EventRuleActionChoices.WEBHOOK,
+            action_object_type=webhook_type,
+            action_object_id=webhook.pk,
+        )
+        cls.healthy_rule.object_types.set([site_type])
+
+        # .objects.create() calls save(), not full_clean(), so an unregistered action_type can be
+        # persisted directly, matching the state of a row whose providing plugin was uninstalled.
+        cls.unavailable_rule = EventRule.objects.create(
+            name='Unavailable Rule',
+            event_types=[OBJECT_CREATED],
+            action_type='someplugin.not_installed',
+        )
+        cls.unavailable_rule.object_types.set([site_type])
+
+    def test_action_is_available_true_for_registered_action(self):
+        self.assertTrue(self.healthy_rule.action_is_available)
+        self.assertIsNotNone(self.healthy_rule.action_provider)
+
+    def test_action_is_available_false_for_unregistered_action(self):
+        self.assertFalse(self.unavailable_rule.action_is_available)
+        self.assertIsNone(self.unavailable_rule.action_provider)
+
+    def test_get_action_type_display_for_registered_action(self):
+        self.assertEqual(self.healthy_rule.get_action_type_display(), 'Webhook')
+
+    def test_get_action_type_display_for_unregistered_action(self):
+        self.assertEqual(
+            self.unavailable_rule.get_action_type_display(),
+            'someplugin.not_installed (unavailable)',
+        )
+
+    def test_get_action_type_color_for_registered_action(self):
+        self.assertIsNone(self.healthy_rule.get_action_type_color())
+
+    def test_get_action_type_color_for_unregistered_action(self):
+        self.assertEqual(self.unavailable_rule.get_action_type_color(), 'red')
+
+    def test_clean_rejects_unchanged_unavailable_action_type(self):
+        """A persisted-but-unavailable action_type is rejected by full_clean() even when left unchanged."""
+        rule = EventRule.objects.get(pk=self.unavailable_rule.pk)
+        rule.enabled = False
+        with self.assertRaises(ValidationError):
+            rule.full_clean()
+
+    def test_clean_rejects_new_row_with_unregistered_action_type(self):
+        rule = EventRule(
+            name='New Unregistered Rule',
+            event_types=[OBJECT_CREATED],
+            action_type='someplugin.also_not_installed',
+        )
+        with self.assertRaises(ValidationError):
+            rule.full_clean()
+
+    def test_clean_rejects_changing_to_unregistered_action_type(self):
+        rule = EventRule.objects.get(pk=self.healthy_rule.pk)
+        rule.action_type = 'someplugin.newly_unregistered'
+        with self.assertRaises(ValidationError):
+            rule.full_clean()
+
+    def test_clean_accepts_registered_action_with_valid_object(self):
+        rule = EventRule.objects.get(pk=self.healthy_rule.pk)
+        rule.full_clean()  # must not raise
+
+
+class EventRuleNoObjectActionTestCase(TestCase):
+    """
+    Model-layer tests for an EventRuleAction which declares object_model=None (no target object).
+    """
+
+    def tearDown(self):
+        super().tearDown()
+        registry['event_rule_actions'].pop('test.model_no_object_action', None)
+
+    def test_full_clean_and_save_with_no_object_action(self):
+        class NoObjectAction(EventRuleAction):
+            slug = 'test.model_no_object_action'
+            label = 'Model No-Object Action'
+            object_required = False
+
+        register_event_rule_action(NoObjectAction)
+
+        site_type = ObjectType.objects.get_for_model(Site)
+        rule = EventRule(
+            name='Model No-Object Rule',
+            event_types=[OBJECT_CREATED],
+            action_type='test.model_no_object_action',
+        )
+        rule.full_clean()  # must not raise: no action_object required or supplied
+        rule.save()
+        rule.object_types.set([site_type])
+
+        rule.refresh_from_db()
+        self.assertIsNone(rule.action_object_type)
+        self.assertIsNone(rule.action_object_id)
+        self.assertIsNone(rule.action_object)
+
 
 class WebhookRenderHeadersTest(TestCase):
 

netbox/extras/tests/test_filtersets.py

@@ -396,6 +396,49 @@ class EventRuleTestCase(TestCase, BaseFilterSetTests):
         params = {'action_type': [EventRuleActionChoices.SCRIPT]}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)
 
+    def test_action_is_available(self):
+        unavailable_rule = EventRule.objects.create(
+            name='Unavailable Filterset Rule',
+            event_types=[OBJECT_CREATED],
+            action_type='someplugin.not_installed_filterset_test',
+        )
+        unavailable_rule.object_types.set([ObjectType.objects.get_for_model(Site)])
+
+        params = {'action_is_available': True}
+        qs = self.filterset(params, EventRule.objects.all()).qs
+        self.assertEqual(qs.count(), 5)
+        self.assertNotIn(unavailable_rule, qs)
+
+        params = {'action_is_available': False}
+        qs = self.filterset(params, EventRule.objects.all()).qs
+        self.assertEqual(qs.count(), 1)
+        self.assertEqual(qs.first(), unavailable_rule)
+
+    def test_action_type_registered_plugin_style_slug(self):
+        """A plugin-registered action slug is a valid action_type filter value, not just the core actions."""
+        from netbox.event_rules import EventRuleAction, register_event_rule_action
+        from netbox.registry import registry
+
+        class FilterTestAction(EventRuleAction):
+            slug = 'test.filterset_registered_action'
+            label = 'Filterset Test Action'
+            object_required = False
+
+        register_event_rule_action(FilterTestAction)
+        self.addCleanup(registry['event_rule_actions'].pop, FilterTestAction.slug, None)
+
+        rule = EventRule.objects.create(
+            name='Filterset Registered Action Rule',
+            event_types=[OBJECT_CREATED],
+            action_type=FilterTestAction.slug,
+        )
+        rule.object_types.set([ObjectType.objects.get_for_model(Site)])
+
+        params = {'action_type': [FilterTestAction.slug]}
+        qs = self.filterset(params, EventRule.objects.all()).qs
+        self.assertEqual(qs.count(), 1)
+        self.assertEqual(qs.first(), rule)
+
     def test_enabled(self):
         params = {'enabled': True}
         self.assertEqual(self.filterset(params, self.queryset).qs.count(), 2)

netbox/extras/tests/test_forms.py

@@ -1,19 +1,25 @@
 import tempfile
 from pathlib import Path
 
-from django.core.exceptions import NON_FIELD_ERRORS
+from django.core.exceptions import NON_FIELD_ERRORS, ValidationError
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.test import TestCase
 
 from core.choices import ManagedFileRootPathChoices
+from core.events import OBJECT_CREATED
 from core.models import DataSource, ObjectType
 from dcim.forms import SiteForm
 from dcim.models import Site
-from extras.choices import CustomFieldTypeChoices
+from extras.choices import CustomFieldTypeChoices, EventRuleActionChoices
 from extras.forms import SavedFilterForm, TableConfigBulkEditForm, TableConfigForm
-from extras.forms.model_forms import CustomFieldChoiceSetForm
+from extras.forms.bulk_import import EventRuleImportForm
+from extras.forms.filtersets import EventRuleFilterForm
+from extras.forms.model_forms import CustomFieldChoiceSetForm, EventRuleForm
 from extras.forms.scripts import ScriptFileForm
-from extras.models import CustomField, CustomFieldChoiceSet, ScriptModule
+from extras.models import CustomField, CustomFieldChoiceSet, EventRule, NotificationGroup, Script, ScriptModule, Webhook
+from netbox.event_rules import EventRuleAction, register_event_rule_action
+from netbox.registry import registry
+from utilities.forms.widgets import HTMXSelect
 
 
 class CustomFieldModelFormTestCase(TestCase):
@@ -337,3 +343,353 @@ class TableConfigFormTestCase(TestCase):
         form = TableConfigBulkEditForm()
         self.assertIn('changelog_message', form.fields)
         self.assertIn('changelog_message', form.meta_fields)
+
+
+class EventRuleFormTestCase(TestCase):
+    """
+    EventRuleForm's action_choice field is built dynamically from the EventRuleAction registry,
+    for both core actions and those registered by a plugin.
+    """
+
+    def tearDown(self):
+        super().tearDown()
+        registry['event_rule_actions'].pop('test.form_no_object_action', None)
+
+    def test_action_type_widget_is_htmx_select(self):
+        """
+        action_choice refreshes via HTMX when action_type changes. The widget must be set on the
+        field itself: Meta.widgets applies only to fields the ModelForm generates from the model,
+        and action_type is declared explicitly.
+        """
+        form = EventRuleForm()
+        widget = form.fields['action_type'].widget
+        self.assertIsInstance(widget, HTMXSelect)
+        self.assertEqual(widget.attrs.get('hx-target'), '#event-rule-action')
+
+    def test_action_choice_field_for_webhook(self):
+        webhook = Webhook.objects.create(name='Form Test Webhook', payload_url='http://localhost:9000/')
+        form = EventRuleForm(data={'action_type': EventRuleActionChoices.WEBHOOK})
+        self.assertIn('action_choice', form.fields)
+        self.assertIn(webhook, form.fields['action_choice'].queryset)
+
+    def test_action_choice_field_for_script(self):
+        form = EventRuleForm(data={'action_type': EventRuleActionChoices.SCRIPT})
+        self.assertIn('action_choice', form.fields)
+        self.assertEqual(form.fields['action_choice'].queryset.model, Script)
+
+    def test_action_choice_field_for_notification(self):
+        form = EventRuleForm(data={'action_type': EventRuleActionChoices.NOTIFICATION})
+        self.assertIn('action_choice', form.fields)
+        self.assertEqual(form.fields['action_choice'].queryset.model, NotificationGroup)
+
+    def test_action_choice_field_labels(self):
+        """The object picker is labeled for the object being selected, not for the action itself."""
+        for action_type, label in (
+            (EventRuleActionChoices.WEBHOOK, 'Webhook'),
+            (EventRuleActionChoices.SCRIPT, 'Script'),
+            (EventRuleActionChoices.NOTIFICATION, 'Notification group'),
+        ):
+            form = EventRuleForm(data={'action_type': action_type})
+            self.assertEqual(form.fields['action_choice'].label, label)
+
+    def test_action_choice_field_honors_object_label(self):
+        class LabeledObjectAction(EventRuleAction):
+            slug = 'test.form_labeled_object_action'
+            label = 'Form Labeled Object Action'
+            object_model = Webhook
+            object_label = 'Destination'
+
+        register_event_rule_action(LabeledObjectAction)
+        self.addCleanup(registry['event_rule_actions'].pop, LabeledObjectAction.slug, None)
+
+        form = EventRuleForm(data={'action_type': LabeledObjectAction.slug})
+        self.assertEqual(form.fields['action_choice'].label, 'Destination')
+
+    def test_action_choice_field_omitted_for_registered_no_object_action(self):
+        class NoObjectAction(EventRuleAction):
+            slug = 'test.form_no_object_action'
+            label = 'Form No-Object Action'
+            object_required = False
+
+        register_event_rule_action(NoObjectAction)
+
+        form = EventRuleForm(data={'action_type': 'test.form_no_object_action'})
+        self.assertNotIn('action_choice', form.fields)
+
+    def test_action_choice_field_falls_back_to_initial_for_unregistered_action(self):
+        """
+        get_field_value() falls back to the field's own initial (webhook) for an unregistered
+        action_type, so init_action_choice() still builds a usable picker.
+        """
+        form = EventRuleForm(data={'action_type': 'not.a.registered.action'})
+        self.assertIn('action_choice', form.fields)
+        self.assertEqual(form.fields['action_choice'].queryset.model, Webhook)
+
+    def test_submit_and_save_with_registered_no_object_action(self):
+        """A runtime-registered action can be submitted and saved end-to-end through the form."""
+        class NoObjectAction(EventRuleAction):
+            slug = 'test.form_no_object_action'
+            label = 'Form No-Object Action'
+            object_required = False
+
+            def enqueue(self, **kwargs):
+                pass
+
+        register_event_rule_action(NoObjectAction)
+
+        object_type = ObjectType.objects.get_for_model(Site)
+        form = EventRuleForm(data={
+            'name': 'Form No-Object Rule',
+            'object_types': [object_type.pk],
+            'event_types': [OBJECT_CREATED],
+            'action_type': 'test.form_no_object_action',
+        })
+        self.assertTrue(form.is_valid(), form.errors)
+        rule = form.save()
+        self.assertIsNone(rule.action_object_type)
+        self.assertIsNone(rule.action_object_id)
+
+    def test_submit_and_save_webhook_action(self):
+        """The generalized form still saves a core Webhook action correctly."""
+        webhook = Webhook.objects.create(name='Form Submit Webhook', payload_url='http://localhost:9000/')
+        object_type = ObjectType.objects.get_for_model(Site)
+        form = EventRuleForm(data={
+            'name': 'Form Webhook Rule',
+            'object_types': [object_type.pk],
+            'event_types': [OBJECT_CREATED],
+            'action_type': EventRuleActionChoices.WEBHOOK,
+            'action_choice': webhook.pk,
+        })
+        self.assertTrue(form.is_valid(), form.errors)
+        rule = form.save()
+        self.assertEqual(rule.action_object, webhook)
+
+    def test_switching_to_optional_object_action_clears_stale_action_object(self):
+        """
+        Switching an existing rule to an action which declares object_model but not
+        object_required, leaving the picker blank, must clear the old action_object.
+        """
+        class OptionalObjectAction(EventRuleAction):
+            slug = 'test.optional_object_action'
+            label = 'Optional Object Action'
+            object_model = Webhook
+            object_required = False
+
+        register_event_rule_action(OptionalObjectAction)
+        self.addCleanup(registry['event_rule_actions'].pop, 'test.optional_object_action', None)
+
+        webhook = Webhook.objects.create(name='Stale Webhook', payload_url='http://localhost:9000/')
+        webhook_type = ObjectType.objects.get_for_model(Webhook)
+        object_type = ObjectType.objects.get_for_model(Site)
+        rule = EventRule.objects.create(
+            name='Stale Object Rule',
+            event_types=[OBJECT_CREATED],
+            action_type=EventRuleActionChoices.WEBHOOK,
+            action_object_type=webhook_type,
+            action_object_id=webhook.pk,
+        )
+        rule.object_types.set([object_type])
+
+        form = EventRuleForm(data={
+            'name': rule.name,
+            'object_types': [object_type.pk],
+            'event_types': [OBJECT_CREATED],
+            'action_type': 'test.optional_object_action',
+            # action_choice omitted: the user left the picker blank
+        }, instance=rule)
+        self.assertTrue(form.is_valid(), form.errors)
+        saved = form.save()
+        self.assertIsNone(saved.action_object_type)
+        self.assertIsNone(saved.action_object_id)
+
+
+class EventRuleFilterFormTestCase(TestCase):
+
+    def test_action_type_choices_reflect_the_live_registry(self):
+        """
+        The filter form's action_type choices must be read from the registry on access, not frozen
+        when this module was first imported.
+        """
+        class FilterFormAction(EventRuleAction):
+            slug = 'test.filter_form_action'
+            label = 'Filter Form Action'
+
+        register_event_rule_action(FilterFormAction)
+        self.addCleanup(registry['event_rule_actions'].pop, FilterFormAction.slug, None)
+
+        choices = dict(EventRuleFilterForm().fields['action_type'].choices)
+        self.assertEqual(choices.get(FilterFormAction.slug), 'Filter Form Action')
+        self.assertIn(None, choices)  # The blank choice is retained
+
+
+class EventRuleImportFormTestCase(TestCase):
+    """
+    EventRuleImportForm resolves action_object via each registered action's resolve_import_object()
+    hook, and treats it as optional (an action need not operate against a target object).
+    """
+
+    def tearDown(self):
+        super().tearDown()
+        registry['event_rule_actions'].pop('test.import_no_object_action', None)
+
+    def test_resolves_webhook_by_name(self):
+        webhook = Webhook.objects.create(name='Import Test Webhook', payload_url='http://localhost:9000/')
+        form = EventRuleImportForm(data={
+            'name': 'Import Webhook Rule',
+            'object_types': 'dcim.site',
+            'event_types': 'object_created',
+            'action_type': EventRuleActionChoices.WEBHOOK,
+            'action_object': webhook.name,
+        })
+        self.assertTrue(form.is_valid(), form.errors)
+        self.assertEqual(form.instance.action_object, webhook)
+
+    def test_resolves_notification_group_by_name(self):
+        """The import form resolves a notification group, not just webhooks and scripts."""
+        group = NotificationGroup.objects.create(name='Import Test Group')
+        form = EventRuleImportForm(data={
+            'name': 'Import Notification Rule',
+            'object_types': 'dcim.site',
+            'event_types': 'object_created',
+            'action_type': EventRuleActionChoices.NOTIFICATION,
+            'action_object': group.name,
+        })
+        self.assertTrue(form.is_valid(), form.errors)
+        self.assertEqual(form.instance.action_object, group)
+
+    def test_unresolvable_webhook_name_is_rejected(self):
+        form = EventRuleImportForm(data={
+            'name': 'Import Bad Webhook Rule',
+            'object_types': 'dcim.site',
+            'event_types': 'object_created',
+            'action_type': EventRuleActionChoices.WEBHOOK,
+            'action_object': 'Does Not Exist',
+        })
+        self.assertFalse(form.is_valid())
+        self.assertIn('action_object', form.errors)
+
+    def test_unregistered_action_type_is_rejected(self):
+        form = EventRuleImportForm(data={
+            'name': 'Import Bad Type Rule',
+            'object_types': 'dcim.site',
+            'event_types': 'object_created',
+            'action_type': 'not.a.registered.action',
+            'action_object': 'whatever',
+        })
+        self.assertFalse(form.is_valid())
+        self.assertIn('action_type', form.errors)
+
+    def test_submit_no_object_action_with_blank_action_object_succeeds(self):
+        """A blank action_object must be accepted for bulk-importing a no-object action."""
+        class NoObjectAction(EventRuleAction):
+            slug = 'test.import_no_object_action'
+            label = 'Import No-Object Action'
+            object_required = False
+
+        register_event_rule_action(NoObjectAction)
+
+        form = EventRuleImportForm(data={
+            'name': 'Import No-Object Rule',
+            'object_types': 'dcim.site',
+            'event_types': 'object_created',
+            'action_type': 'test.import_no_object_action',
+            'action_object': '',
+        })
+        self.assertTrue(form.is_valid(), form.errors)
+        rule = form.save()
+        self.assertIsNone(rule.action_object_type)
+        self.assertIsNone(rule.action_object_id)
+
+    def test_blank_action_object_rejected_for_object_required_action(self):
+        """A blank action_object must be rejected cleanly (not raise) for an action which requires one."""
+        form = EventRuleImportForm(data={
+            'name': 'Import Webhook No Object',
+            'object_types': 'dcim.site',
+            'event_types': 'object_created',
+            'action_type': EventRuleActionChoices.WEBHOOK,
+            'action_object': '',
+        })
+        self.assertFalse(form.is_valid())
+        self.assertIn('action_object', form.errors)
+
+    def test_action_object_rejected_for_action_without_object_model(self):
+        """
+        An action declaring no object_model rejects a supplied action_object as inapplicable,
+        rather than reporting it as an unsupported bulk import.
+        """
+        class NoObjectAction(EventRuleAction):
+            slug = 'test.import_no_object_action'
+            label = 'Import No-Object Action'
+            object_required = False
+
+        register_event_rule_action(NoObjectAction)
+
+        form = EventRuleImportForm(data={
+            'name': 'Import No-Object Rule With Object',
+            'object_types': 'dcim.site',
+            'event_types': 'object_created',
+            'action_type': 'test.import_no_object_action',
+            'action_object': 'Some Object',
+        })
+        self.assertFalse(form.is_valid())
+        self.assertIn('does not operate against a target object', str(form.errors['action_object']))
+
+    def test_csv_update_to_optional_object_action_clears_stale_action_object(self):
+        """
+        A CSV row updating an existing rule to an action_type which declares object_model but not
+        object_required, with action_object left blank, must clear the previous action_object.
+        """
+        class OptionalObjectAction(EventRuleAction):
+            slug = 'test.import_optional_object_action'
+            label = 'Import Optional Object Action'
+            object_model = Webhook
+            object_required = False
+
+        register_event_rule_action(OptionalObjectAction)
+        self.addCleanup(registry['event_rule_actions'].pop, 'test.import_optional_object_action', None)
+
+        webhook = Webhook.objects.create(name='CSV Stale Webhook', payload_url='http://localhost:9000/')
+        webhook_type = ObjectType.objects.get_for_model(Webhook)
+        rule = EventRule.objects.create(
+            name='CSV Stale Object Rule',
+            event_types=[OBJECT_CREATED],
+            action_type=EventRuleActionChoices.WEBHOOK,
+            action_object_type=webhook_type,
+            action_object_id=webhook.pk,
+        )
+
+        form = EventRuleImportForm(data={
+            'name': rule.name,
+            'object_types': 'dcim.site',
+            'event_types': 'object_created',
+            'action_type': 'test.import_optional_object_action',
+            'action_object': '',
+        }, instance=rule)
+        self.assertTrue(form.is_valid(), form.errors)
+        saved = form.save()
+        self.assertIsNone(saved.action_object_type)
+        self.assertIsNone(saved.action_object_id)
+
+    def test_action_validate_error_on_unexposed_field_becomes_non_field_error(self):
+        """A validate() error keyed by a field this form doesn't expose (e.g. action_data) must not raise."""
+        class ActionDataValidatingAction(EventRuleAction):
+            slug = 'test.import_action_data_validating'
+            label = 'Import Action Data Validating'
+            object_required = False
+
+            def validate(self, *, action_object, action_data):
+                raise ValidationError({'action_data': 'Bad action_data for test'})
+
+        register_event_rule_action(ActionDataValidatingAction)
+        self.addCleanup(registry['event_rule_actions'].pop, 'test.import_action_data_validating', None)
+
+        form = EventRuleImportForm(data={
+            'name': 'Import Bad Action Data Rule',
+            'object_types': 'dcim.site',
+            'event_types': 'object_created',
+            'action_type': 'test.import_action_data_validating',
+            'action_object': '',
+        })
+        self.assertFalse(form.is_valid())
+        self.assertIn(NON_FIELD_ERRORS, form.errors)
+        self.assertIn('Bad action_data for test', form.errors[NON_FIELD_ERRORS])

netbox/extras/tests/test_graphql.py

@@ -0,0 +1,56 @@
+import json
+
+from django.urls import reverse
+from rest_framework import status
+
+from core.events import OBJECT_CREATED
+from core.models import ObjectType
+from dcim.models import Site
+from extras.choices import EventRuleActionChoices
+from extras.graphql.enums import EventRuleActionEnum
+from extras.models import EventRule, Webhook
+from utilities.testing import APITestCase
+
+
+class EventRuleActionEnumTestCase(APITestCase):
+    """EventRuleActionEnum must reflect the live action registry, and the filter must use it."""
+
+    def test_enum_contains_core_actions(self):
+        # A subset check, since an installed plugin may register actions of its own
+        values = {member.value for member in EventRuleActionEnum}
+        core_slugs = {
+            EventRuleActionChoices.WEBHOOK, EventRuleActionChoices.SCRIPT, EventRuleActionChoices.NOTIFICATION,
+        }
+        self.assertLessEqual(core_slugs, values)
+
+    def test_filter_event_rules_by_action_type(self):
+        webhook = Webhook.objects.create(name='GraphQL Enum Test Webhook', payload_url='http://localhost:9000/')
+        webhook_type = ObjectType.objects.get_for_model(Webhook)
+        site_type = ObjectType.objects.get_for_model(Site)
+
+        webhook_rule = EventRule.objects.create(
+            name='GraphQL Enum Webhook Rule',
+            event_types=[OBJECT_CREATED],
+            action_type=EventRuleActionChoices.WEBHOOK,
+            action_object_type=webhook_type,
+            action_object_id=webhook.pk,
+        )
+        webhook_rule.object_types.set([site_type])
+
+        script_rule = EventRule.objects.create(
+            name='GraphQL Enum Script Rule',
+            event_types=[OBJECT_CREATED],
+            action_type=EventRuleActionChoices.SCRIPT,
+        )
+        script_rule.object_types.set([site_type])
+
+        self.add_permissions('extras.view_eventrule')
+        url = reverse('graphql')
+        query = '{event_rule_list(filters: {action_type: {exact: WEBHOOK}}) {name action_type}}'
+        response = self.client.post(url, data={'query': query}, format='json', **self.header)
+        self.assertHttpStatus(response, status.HTTP_200_OK)
+
+        data = json.loads(response.content)
+        self.assertNotIn('errors', data)
+        names = {rule['name'] for rule in data['data']['event_rule_list']}
+        self.assertEqual(names, {'GraphQL Enum Webhook Rule'})

netbox/extras/tests/test_models.py

@@ -32,6 +32,7 @@ from extras.models import (
     TableConfig,
     Tag,
     TaggedItem,
+    Webhook,
 )
 from extras.models.mixins import RenderTemplateMixin
 from tenancy.models import Tenant, TenantGroup
@@ -1586,8 +1587,16 @@ class EventRuleTestCase(TestCase):
         """
         clean() should accept a JSON object (or null) as action_data.
         """
+        webhook = Webhook.objects.create(name='Action Data Test Webhook', payload_url='http://localhost:9000/')
+        webhook_type = ObjectType.objects.get_for_model(Webhook)
         for value in ({'key': 'value'}, None):
-            rule = EventRule(name='test', event_types=[OBJECT_CREATED], action_data=value)
+            rule = EventRule(
+                name='test',
+                event_types=[OBJECT_CREATED],
+                action_data=value,
+                action_object_type=webhook_type,
+                action_object_id=webhook.pk,
+            )
             rule.clean()
 
     def test_action_data_clean_rejects_non_dict(self):

netbox/extras/tests/test_tables.py

@@ -1,4 +1,9 @@
-from extras.models import Bookmark, Notification, Subscription
+from django.test import TestCase
+
+from core.events import OBJECT_CREATED
+from core.models import ObjectType
+from dcim.models import Site
+from extras.models import Bookmark, EventRule, Notification, Subscription
 from extras.tables import *
 from utilities.testing import TableTestCases
 
@@ -69,6 +74,39 @@ class EventRuleTableTestCase(TableTestCases.StandardTableTestCase):
     table = EventRuleTable
 
 
+class EventRuleTableActionTypeRenderingTestCase(TestCase):
+    """
+    render_action_type() badges an unregistered action as unavailable; value_action_type() carries
+    the same label for non-HTML output (e.g. CSV export), without the markup.
+    """
+
+    def test_render_action_type_for_registered_action(self):
+        rule = EventRule.objects.create(name='Render Test Rule', event_types=[OBJECT_CREATED], action_type='webhook')
+        rule.object_types.set([ObjectType.objects.get_for_model(Site)])
+
+        table = EventRuleTable(EventRule.objects.filter(pk=rule.pk))
+        self.assertEqual(table.render_action_type(rule), 'Webhook')
+        self.assertEqual(table.value_action_type(rule), 'Webhook')
+
+    def test_render_action_type_for_unregistered_action(self):
+        rule = EventRule.objects.create(
+            name='Render Test Unavailable Rule',
+            event_types=[OBJECT_CREATED],
+            action_type='someplugin.not_installed_render_test',
+        )
+        rule.object_types.set([ObjectType.objects.get_for_model(Site)])
+
+        table = EventRuleTable(EventRule.objects.filter(pk=rule.pk))
+        rendered = table.render_action_type(rule)
+        self.assertIn('someplugin.not_installed_render_test (unavailable)', rendered)
+        self.assertIn('badge text-bg-red', rendered)
+
+        # The same label, without markup
+        value = table.value_action_type(rule)
+        self.assertEqual(value, 'someplugin.not_installed_render_test (unavailable)')
+        self.assertNotIn('<span', value)
+
+
 class TagTableTestCase(TableTestCases.StandardTableTestCase):
     table = TagTable
 

netbox/netbox/event_rules.py

@@ -0,0 +1,204 @@
+import re
+
+from django.core.exceptions import ImproperlyConfigured, ValidationError
+from django.utils.text import capfirst
+from django.utils.translation import gettext_lazy as _
+
+from netbox.registry import registry
+from utilities.choices import Choice
+from utilities.string import enum_key
+
+__all__ = (
+    'EventRuleAction',
+    'get_event_rule_action',
+    'get_event_rule_action_choices',
+    'get_event_rule_action_slugs',
+    'register_event_rule_action',
+)
+
+# A slug must sanitize (via enum_key()) into a valid GraphQL enum member name, which rules out a
+# leading digit or underscore. Hyphens are rejected outright rather than sanitized away, so that a
+# slug always reads as it was written.
+SLUG_RE = re.compile(r'^[a-z][a-z0-9_]*(\.[a-z0-9_]+)*$')
+
+
+# This module must not import any concrete Django models: it's imported by netbox.plugins (itself
+# imported by netbox.settings, before the app registry is populated), so only registry-level
+# bookkeeping belongs here. Subclasses that reference real models (e.g. NetBox's own
+# WebhookAction/ScriptAction/NotificationAction) live in netbox.extras.event_rules instead, and are
+# registered from ExtrasConfig.ready() once the app registry is available.
+class EventRuleAction:
+    """
+    Base class for a registered Event Rule action. Subclass this to add a new action type that an
+    EventRule can dispatch to, whether defined in NetBox core or in a plugin.
+
+    Registration instantiates the class once; that single instance serves every event rule, request,
+    and background worker thread for the lifetime of the process. Implementations must therefore be
+    stateless: enqueue() and validate() must not stash per-event data on self, as concurrent
+    dispatches would race over it. Everything an action needs is passed in as arguments.
+
+    Attributes:
+        slug: A unique identifier for this action (e.g. "webhook", or "myplugin.run_check" for a
+            plugin-provided action). Must begin with a lowercase letter, and may contain only
+            letters, digits, underscores, and dot-separated segments thereafter -- no hyphens (use
+            an underscore instead, e.g. "my_plugin.open_ticket"). A dotted namespace prefix is
+            strongly recommended for plugin-provided actions to avoid collisions with other plugins
+            or future core actions.
+        label: The human-friendly name shown in the UI/API.
+        description: An optional, longer description shown alongside the label (e.g. as a tooltip
+            in the action_type dropdown).
+        object_model: The model class (if any) which EventRule.action_object must be an instance of.
+            May be left as None if this action never operates against a target object, in which
+            case supplying an action_object is treated as a validation error.
+        object_required: Whether an action_object must be supplied for this action to be usable.
+            Defaults to False; set True (alongside object_model) when a target object is mandatory.
+            An action may declare object_model but still treat the object as optional.
+        object_label: The label for the object selection field on the event rule form. Defaults to
+            object_model's verbose name.
+    """
+    slug = None
+    label = None
+    description = None
+    object_model = None
+    object_required = False
+    object_label = None
+
+    # Set per-instance by register_event_rule_action(); a subclass override is ignored. Determines
+    # whether a dispatch-time exception from this action is isolated or propagates (see
+    # process_event_rules() in extras.events).
+    is_plugin_provided = True
+
+    def __repr__(self):
+        return f"<{self.__class__.__name__}: {self.slug}>"
+
+    def get_object_queryset(self):
+        """
+        Return the queryset of objects eligible for selection as this action's action_object, or
+        None if object_model is not set.
+        """
+        if self.object_model is None:
+            return None
+        return self.object_model.objects.all()
+
+    def get_object_label(self):
+        """
+        Return the label for this action's object selection field, or None if object_model is not
+        set. Defaults to object_model's verbose name, unless object_label has been set explicitly.
+        """
+        if self.object_label:
+            return self.object_label
+        if self.object_model is None:
+            return None
+        return capfirst(self.object_model._meta.verbose_name)
+
+    def resolve_import_object(self, value):
+        """
+        Optional hook: resolve a CSV/bulk-import "action object" string to a model instance. Raise
+        django.core.exceptions.ObjectDoesNotExist (or a subclass) if the value doesn't resolve.
+        Return None (the default) if this action doesn't support bulk import.
+        """
+        return
+
+    def _validate(self, *, action_object, action_data):
+        """
+        Entry point called from EventRule.clean(). Enforces the base object_required/object_model
+        checks, then delegates to validate() for any action-specific validation.
+        """
+        if self.object_required and action_object is None:
+            raise ValidationError({
+                'action_object_id': _("This action requires a target object to be selected."),
+            })
+        if action_object is not None:
+            if self.object_model is None:
+                raise ValidationError({
+                    'action_object_id': _("This action does not operate against a target object."),
+                })
+            if not isinstance(action_object, self.object_model):
+                raise ValidationError({
+                    'action_object_id': _("Selected object is not a valid {model}.").format(
+                        model=self.object_model._meta.verbose_name
+                    ),
+                })
+        self.validate(action_object=action_object, action_data=action_data)
+
+    def validate(self, *, action_object, action_data):
+        """
+        Optional hook: add custom validation, raising ValidationError on failure. No-op by
+        default; no need to call super() -- _validate() above runs the base checks regardless.
+        """
+        pass
+
+    def enqueue(self, *, event_rule, event_context, action_object, action_data):
+        """
+        Perform (or schedule) this action in response to a queued event. Implementations should
+        not raise for conditions that are the fault of this EventRule's own configuration alone;
+        log and return instead, so that other EventRules processed in the same batch are
+        unaffected.
+        """
+        raise NotImplementedError(f"{self.__class__.__name__} must implement enqueue()")
+
+
+def register_event_rule_action(cls, *, is_plugin_provided=True):
+    """
+    Register an EventRuleAction subclass. Can be used as a decorator, or called directly (e.g. when
+    iterating a plugin's declared event_rule_actions):
+
+        @register_event_rule_action
+        class MyAction(EventRuleAction):
+            slug = 'myplugin.my_action'
+            ...
+
+    Raises ImproperlyConfigured if slug/label are missing, the slug is malformed, already
+    registered, collides via enum_key() with another registered slug once both feed the GraphQL
+    EventRuleActionEnum (see extras.graphql.enums), or object_required is set without an
+    object_model to validate the object against. Checking slug/label here rather than at class
+    definition means an intermediate base class shared by several concrete actions in a plugin can
+    leave them unset.
+
+    is_plugin_provided determines whether a dispatch-time exception from this action is isolated
+    or propagates (see process_event_rules() in extras.events); defaults to True. NetBox's own
+    core registrations (extras.apps.ExtrasConfig) pass False explicitly.
+    """
+    instance = cls()
+    if not instance.slug:
+        raise ImproperlyConfigured(f"{cls.__name__} must define a non-empty 'slug' attribute.")
+    if not instance.label:
+        raise ImproperlyConfigured(f"{cls.__name__} must define a 'label' attribute.")
+    if instance.object_required and instance.object_model is None:
+        # Unsatisfiable: an object is mandatory, yet any object supplied is rejected by _validate().
+        raise ImproperlyConfigured(
+            f"{cls.__name__} sets object_required but no object_model; a target object cannot be "
+            f"required for an action which does not operate against one."
+        )
+    if not SLUG_RE.fullmatch(instance.slug):
+        raise ImproperlyConfigured(
+            f"Invalid event rule action slug {instance.slug!r}: must be lowercase, start with a "
+            f"letter, and use only letters, digits, underscores, and dot-separated segments."
+        )
+    if instance.slug in registry['event_rule_actions']:
+        raise ImproperlyConfigured(f"An event rule action named {instance.slug} has already been registered!")
+    new_key = enum_key(instance.slug)
+    for existing in registry['event_rule_actions'].values():
+        if enum_key(existing.slug) == new_key:
+            raise ImproperlyConfigured(
+                f"Event rule action slug {instance.slug!r} collides with the already-registered "
+                f"{existing.slug!r} once both are sanitized into a GraphQL enum member name."
+            )
+    instance.is_plugin_provided = is_plugin_provided
+    registry['event_rule_actions'][instance.slug] = instance
+    return cls
+
+
+def get_event_rule_action(slug):
+    return registry['event_rule_actions'].get(slug)
+
+
+def get_event_rule_action_choices():
+    return [
+        Choice(action.slug, action.label, description=action.description)
+        for action in registry['event_rule_actions'].values()
+    ]
+
+
+def get_event_rule_action_slugs():
+    return list(registry['event_rule_actions'].keys())

netbox/netbox/plugins/__init__.py

@@ -7,6 +7,7 @@ from django.utils.module_loading import import_string
 from packaging import version
 
 from core.exceptions import IncompatiblePluginError
+from netbox.event_rules import register_event_rule_action
 from netbox.registry import registry
 from netbox.search import register_search
 from netbox.utils import register_data_backend
@@ -35,6 +36,7 @@ registry['plugins'].update({
 DEFAULT_RESOURCE_PATHS = {
     'search_indexes': 'search.indexes',
     'data_backends': 'data_backends.backends',
+    'event_rule_actions': 'event_rules.event_rule_actions',
     'graphql_schema': 'graphql.schema',
     'jinja_filters': 'jinja_env.filters',
     'graphql_type_extensions': 'graphql.type_extensions',
@@ -86,6 +88,7 @@ class PluginConfig(AppConfig):
     # Optional plugin resources
     search_indexes = None
     data_backends = None
+    event_rule_actions = None
     graphql_schema = None
     jinja_filters = None
     graphql_type_extensions = None
@@ -142,6 +145,11 @@ class PluginConfig(AppConfig):
         for backend in data_backends:
             register_data_backend()(backend)
 
+        # Register event rule actions (if defined)
+        event_rule_actions = self._load_resource('event_rule_actions') or []
+        for action in event_rule_actions:
+            register_event_rule_action(action)
+
         # Register Jinja filters (if defined)
         if jinja_filters := self._load_resource('jinja_filters'):
             register_jinja_filters(jinja_filters)

netbox/netbox/registry.py

@@ -25,6 +25,7 @@ class Registry(dict):
 registry = Registry({
     'counter_fields': collections.defaultdict(dict),
     'data_backends': dict(),
+    'event_rule_actions': dict(),
     'event_types': dict(),
     'filtersets': dict(),
     'model_actions': collections.defaultdict(set),

netbox/netbox/tests/dummy_plugin/event_rules.py

@@ -0,0 +1,15 @@
+from netbox.event_rules import EventRuleAction
+
+__all__ = (
+    'DummyRaisingAction',
+)
+
+
+class DummyRaisingAction(EventRuleAction):
+    """A plugin action for testing process_event_rules() exception handling. Registered per-test, not on load."""
+    slug = 'dummy_plugin.raising_action'
+    label = 'Dummy Raising Action'
+    object_required = False
+
+    def enqueue(self, **kwargs):
+        raise RuntimeError("intentional failure for test")