diff --git a/packages/forest_admin_datasource_zendesk/Gemfile b/packages/forest_admin_datasource_zendesk/Gemfile
index 56cf384f3..c229ff1d5 100644
--- a/packages/forest_admin_datasource_zendesk/Gemfile
+++ b/packages/forest_admin_datasource_zendesk/Gemfile
@@ -2,6 +2,7 @@ source 'https://rubygems.org'
 
 gemspec
 
+gem 'forest_admin_datasource_customizer'
 gem 'forest_admin_datasource_toolkit'
 gem 'rake', '~> 13.0'
 gem 'rubocop', '1.86.1'
diff --git a/packages/forest_admin_datasource_zendesk/Gemfile-test b/packages/forest_admin_datasource_zendesk/Gemfile-test
index 126a451a7..64fab449a 100644
--- a/packages/forest_admin_datasource_zendesk/Gemfile-test
+++ b/packages/forest_admin_datasource_zendesk/Gemfile-test
@@ -9,6 +9,7 @@ gem 'rubocop-performance', '1.26.1'
 gem 'rubocop-rspec', '3.9.0'
 
 group :development, :test do
+  gem 'forest_admin_datasource_customizer', path: '../forest_admin_datasource_customizer'
   gem 'forest_admin_datasource_toolkit', path: '../forest_admin_datasource_toolkit'
   gem 'rspec', '~> 3.0'
   gem 'simplecov', '~> 0.22', require: false
diff --git a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/ticket.rb b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/ticket.rb
index 4475d1662..ac9d7e6a2 100644
--- a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/ticket.rb
+++ b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/ticket.rb
@@ -8,10 +8,6 @@ class Ticket < BaseCollection
 
       ManyToOneSchema = ForestAdminDatasourceToolkit::Schema::Relations::ManyToOneSchema
 
-      ENUM_STATUS   = %w[new open pending hold solved closed].freeze
-      ENUM_PRIORITY = %w[low normal high urgent].freeze
-      ENUM_TYPE     = %w[problem incident question task].freeze
-
       ZENDESK_SORTABLE = {
         'updated_at' => 'updated_at',
         'created_at' => 'created_at',
diff --git a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/ticket/schema_definition.rb b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/ticket/schema_definition.rb
index b1a76d438..cb21ebb51 100644
--- a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/ticket/schema_definition.rb
+++ b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/ticket/schema_definition.rb
@@ -18,11 +18,14 @@ def define_schema
           add_field('description', ColumnSchema.new(column_type: 'String', filter_operators: [],
                                                     is_read_only: false, is_sortable: false))
           add_field('status', ColumnSchema.new(column_type: 'Enum', filter_operators: STRING_OPS,
-                                               enum_values: ENUM_STATUS, is_read_only: false, is_sortable: true))
+                                               enum_values: TicketEnums::STATUS, is_read_only: false,
+                                               is_sortable: true))
           add_field('priority', ColumnSchema.new(column_type: 'Enum', filter_operators: STRING_OPS,
-                                                 enum_values: ENUM_PRIORITY, is_read_only: false, is_sortable: true))
+                                                 enum_values: TicketEnums::PRIORITY, is_read_only: false,
+                                                 is_sortable: true))
           add_field('ticket_type', ColumnSchema.new(column_type: 'Enum', filter_operators: STRING_OPS,
-                                                    enum_values: ENUM_TYPE, is_read_only: false, is_sortable: true))
+                                                    enum_values: TicketEnums::TYPE, is_read_only: false,
+                                                    is_sortable: true))
           add_field('requester_id', ColumnSchema.new(column_type: 'Number', filter_operators: NUMBER_OPS,
                                                      is_read_only: false, is_sortable: true))
           add_field('assignee_id', ColumnSchema.new(column_type: 'Number', filter_operators: NUMBER_OPS,
diff --git a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/user.rb b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/user.rb
index f85407f55..bb20d9f92 100644
--- a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/user.rb
+++ b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/collections/user.rb
@@ -6,6 +6,8 @@ class User < BaseCollection
       ManyToOneSchema = ForestAdminDatasourceToolkit::Schema::Relations::ManyToOneSchema
       OneToManySchema = ForestAdminDatasourceToolkit::Schema::Relations::OneToManySchema
       ENUM_ROLE       = %w[end-user agent admin].freeze
+      BASE_ATTR_KEYS  = %w[id email name role phone organization_id time_zone locale verified suspended
+                           created_at updated_at].freeze
 
       ZENDESK_SORTABLE = {
         'created_at' => 'created_at',
@@ -100,22 +102,11 @@ def define_relations
 
       def serialize(user)
         attrs = attrs_of(user)
-        result = base_attributes(attrs)
+        result = BASE_ATTR_KEYS.to_h { |k| [k, attrs[k]] }
         user_fields = attrs['user_fields'] || {}
         @custom_fields.each { |cf| result[cf[:column_name]] = user_fields[cf[:zendesk_key]] }
         result
       end
-
-      def base_attributes(attrs)
-        {
-          'id' => attrs['id'], 'email' => attrs['email'], 'name' => attrs['name'],
-          'role' => attrs['role'], 'phone' => attrs['phone'],
-          'organization_id' => attrs['organization_id'],
-          'time_zone' => attrs['time_zone'], 'locale' => attrs['locale'],
-          'verified' => attrs['verified'], 'suspended' => attrs['suspended'],
-          'created_at' => attrs['created_at'], 'updated_at' => attrs['updated_at']
-        }
-      end
     end
   end
 end
diff --git a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/close_ticket.rb b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/close_ticket.rb
new file mode 100644
index 000000000..5b6da151b
--- /dev/null
+++ b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/close_ticket.rb
@@ -0,0 +1,133 @@
+module ForestAdminDatasourceZendesk
+  module Plugins
+    # The Zendesk ticket id is read from a configurable column on the host
+    # record(s); Zendesk sometimes rejects the direct `open -> closed`
+    # transition so failures are surfaced per-id rather than retried.
+    class CloseTicket < ForestAdminDatasourceCustomizer::Plugins::Plugin
+      BaseAction  = ForestAdminDatasourceCustomizer::Decorators::Action::BaseAction
+      ActionScope = ForestAdminDatasourceCustomizer::Decorators::Action::Types::ActionScope
+
+      STATUSES = %w[solved closed].freeze
+      SCOPE_KEYS = %i[single bulk].freeze
+
+      # Zendesk refuses any update on a closed ticket with this exact
+      # wording — detected so we can swap the raw stack for a clean message.
+      ALREADY_CLOSED_PATTERN = 'closed prevents ticket update'.freeze
+
+      NAMES = {
+        'solved' => { single: 'Mark Zendesk ticket as solved',
+                      bulk: 'Mark selected Zendesk tickets as solved' }.freeze,
+        'closed' => { single: 'Mark Zendesk ticket as closed',
+                      bulk: 'Mark selected Zendesk tickets as closed' }.freeze
+      }.freeze
+
+      SCOPES = { single: ActionScope::SINGLE, bulk: ActionScope::BULK }.freeze
+
+      def run(_datasource_customizer, collection_customizer = nil, options = {})
+        datasource = options[:datasource]
+        ticket_id_field = options[:ticket_id_field]
+        raise ArgumentError, 'CloseTicket plugin requires :datasource' unless datasource
+        raise ArgumentError, 'CloseTicket plugin requires :ticket_id_field' unless ticket_id_field
+        raise ArgumentError, 'CloseTicket plugin requires a collection' unless collection_customizer
+
+        statuses = normalize_statuses(options[:statuses])
+        scopes   = normalize_scopes(options[:scopes])
+
+        variants(statuses, scopes).each do |name, status, scope|
+          collection_customizer.add_action(name, build_action(datasource, status, scope, ticket_id_field))
+        end
+      end
+
+      private
+
+      def normalize_statuses(value)
+        list = Array(value).map(&:to_s).uniq
+        list = STATUSES if list.empty?
+        unknown = list - STATUSES
+        return list if unknown.empty?
+
+        raise ForestAdminDatasourceToolkit::Exceptions::ForestException,
+              "Unknown CloseTicket statuses: #{unknown.join(", ")}. Allowed: #{STATUSES.join(", ")}."
+      end
+
+      def normalize_scopes(value)
+        list = Array(value).map(&:to_sym).uniq
+        list = SCOPE_KEYS if list.empty?
+        unknown = list - SCOPE_KEYS
+        return list if unknown.empty?
+
+        raise ForestAdminDatasourceToolkit::Exceptions::ForestException,
+              "Unknown CloseTicket scopes: #{unknown.join(", ")}. Allowed: #{SCOPE_KEYS.join(", ")}."
+      end
+
+      def variants(statuses, scopes)
+        statuses.flat_map do |status|
+          scopes.map { |scope_key| [NAMES[status][scope_key], status, SCOPES[scope_key]] }
+        end
+      end
+
+      def build_action(datasource, status, scope, ticket_id_field)
+        BaseAction.new(scope: scope, &executor(datasource, status, ticket_id_field))
+      end
+
+      def executor(datasource, status, ticket_id_field)
+        lambda do |context, result_builder|
+          ids = resolve_ticket_ids(context, ticket_id_field)
+          next result_builder.error(message: "No Zendesk ticket id found in '#{ticket_id_field}'.") if ids.empty?
+
+          succeeded, already_closed, failed = apply_status(datasource, ids, status)
+
+          # Closed tickets can't be reopened to 'solved'; fold into failures.
+          if status == 'solved'
+            failed += already_closed.map { |id| [id, 'ticket is already closed (cannot reopen to mark as solved)'] }
+            already_closed = []
+          end
+
+          if succeeded.empty? && already_closed.empty?
+            result_builder.error(message: Messages.error(failed, status))
+          else
+            result_builder.success(message: Messages.success(succeeded, already_closed, failed, status))
+          end
+        end
+      end
+
+      def resolve_ticket_ids(context, ticket_id_field)
+        records = context.get_records([ticket_id_field])
+        records = [records].compact unless records.is_a?(Array)
+        records.filter_map { |r| r[ticket_id_field] || r[ticket_id_field.to_sym] }
+      rescue StandardError => e
+        ForestAdminDatasourceZendesk.logger.warn(
+          "[forest_admin_datasource_zendesk] failed to resolve ticket ids from '#{ticket_id_field}': " \
+          "#{e.class}: #{e.message}"
+        )
+        []
+      end
+
+      # Per-id rescue so a single transition rejection doesn't abort bulk.
+      def apply_status(datasource, ids, status)
+        succeeded = []
+        already_closed = []
+        failed = []
+        ids.each do |id|
+          datasource.client.update_ticket(id, 'status' => status)
+          succeeded << id
+        rescue StandardError => e
+          if already_closed?(e)
+            already_closed << id
+          else
+            ForestAdminDatasourceZendesk.logger.warn(
+              "[forest_admin_datasource_zendesk] failed to set ticket ##{id} to '#{status}': " \
+              "#{e.class}: #{e.message}"
+            )
+            failed << [id, "#{e.class}: #{e.message}"]
+          end
+        end
+        [succeeded, already_closed, failed]
+      end
+
+      def already_closed?(error)
+        error.message.to_s.include?(ALREADY_CLOSED_PATTERN)
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/close_ticket/messages.rb b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/close_ticket/messages.rb
new file mode 100644
index 000000000..9b777225a
--- /dev/null
+++ b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/close_ticket/messages.rb
@@ -0,0 +1,41 @@
+module ForestAdminDatasourceZendesk
+  module Plugins
+    class CloseTicket
+      module Messages
+        module_function
+
+        def success(succeeded, already_closed, failed, status)
+          [succeeded_phrase(succeeded, status), already_closed_phrase(already_closed),
+           failed_phrase(failed)].compact.join(' ')
+        end
+
+        def error(failed, status)
+          verb = status == 'closed' ? 'close' : 'mark as solved'
+          return "Failed to #{verb} ticket ##{failed.first.first}: #{failed.first.last}" if failed.size == 1
+
+          "Failed to #{verb} all #{failed.size} tickets. First error: #{failed.first.last}"
+        end
+
+        def succeeded_phrase(succeeded, status)
+          return nil if succeeded.empty?
+
+          verb = status == 'closed' ? 'closed' : 'marked as solved'
+          succeeded.size == 1 ? "Ticket ##{succeeded.first} #{verb}." : "#{succeeded.size} tickets #{verb}."
+        end
+
+        def already_closed_phrase(already_closed)
+          return nil if already_closed.empty?
+          return "Ticket ##{already_closed.first} was already closed." if already_closed.size == 1
+
+          "#{already_closed.size} tickets were already closed: #{already_closed.join(", ")}."
+        end
+
+        def failed_phrase(failed)
+          return nil if failed.empty?
+
+          "#{failed.size} failed: #{failed.map(&:first).join(", ")}."
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/create_ticket_with_notification.rb b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/create_ticket_with_notification.rb
new file mode 100644
index 000000000..b65b0b8f8
--- /dev/null
+++ b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/create_ticket_with_notification.rb
@@ -0,0 +1,107 @@
+module ForestAdminDatasourceZendesk
+  module Plugins
+    # Zendesk creates the requester user on the fly from the form's email,
+    # so the action can be registered on any host collection — no relation
+    # to Zendesk needed.
+    class CreateTicketWithNotification < ForestAdminDatasourceCustomizer::Plugins::Plugin
+      BaseAction  = ForestAdminDatasourceCustomizer::Decorators::Action::BaseAction
+      ActionScope = ForestAdminDatasourceCustomizer::Decorators::Action::Types::ActionScope
+
+      NAME = 'Create ticket and notify'.freeze
+
+      def run(_datasource_customizer, collection_customizer = nil, options = {})
+        datasource = options[:datasource]
+        raise ArgumentError, 'CreateTicketWithNotification plugin requires :datasource' unless datasource
+        raise ArgumentError, 'CreateTicketWithNotification plugin requires a collection' unless collection_customizer
+
+        opts = options.except(:datasource)
+        opts[:email_templates] = Array(opts[:email_templates]).compact
+        collection_customizer.add_action(opts[:action_name] || NAME, build_action(datasource, opts))
+      end
+
+      private
+
+      def build_action(datasource, opts)
+        BaseAction.new(scope: ActionScope::SINGLE, form: FormBuilder.build(opts), &executor(datasource, opts))
+      end
+
+      def executor(datasource, opts)
+        lambda do |context, result_builder|
+          values = context.form_values
+          email  = values['Requester email']
+          next result_builder.error(message: 'Requester email is required.') unless present?(email)
+
+          payload = build_payload(values, email, opts)
+          ticket = datasource.client.create_ticket(payload)
+          ticket_id = ticket.respond_to?(:[]) ? ticket['id'] : nil
+
+          writeback = write_back_ticket_id(context, opts[:ticket_id_field], ticket_id)
+          result_builder.success(message: success_message(ticket_id, values, writeback))
+        end
+      end
+
+      def build_payload(values, email, opts)
+        internal_note = truthy?(values['Send as internal note'])
+        payload = {
+          # Zendesk's create-user-on-the-fly requires a non-empty `name`;
+          # derive from the email's local-part. Ignored if the user exists.
+          'requester' => { 'email' => email, 'name' => derive_requester_name(email) },
+          'subject' => values['Subject'],
+          'comment' => { 'html_body' => values['Message'], 'public' => !internal_note }
+        }
+        priority = present?(opts[:priority_override]) ? opts[:priority_override] : values['Priority']
+        type     = present?(opts[:type_override])     ? opts[:type_override]     : values['Type']
+        payload['priority']  = priority             if present?(priority)
+        payload['type']      = type                 if present?(type)
+        # Zendesk's `recipient` = the support address replies come FROM.
+        payload['recipient'] = opts[:sender_email] if present?(opts[:sender_email])
+        payload
+      end
+
+      def derive_requester_name(email)
+        local = email.to_s.split('@').first.to_s
+        local.empty? ? email.to_s : local
+      end
+
+      # Best-effort: a writeback failure mustn't roll back the Zendesk ticket.
+      def write_back_ticket_id(context, field, ticket_id)
+        return :skipped if field.nil? || ticket_id.nil?
+
+        context.collection.update(context.filter, { field => ticket_id })
+        :ok
+      rescue StandardError => e
+        ForestAdminDatasourceZendesk.logger.warn(
+          "[forest_admin_datasource_zendesk] failed to store ticket id in '#{field}': #{e.class}: #{e.message}"
+        )
+        [:failed, "#{e.class}: #{e.message}"]
+      end
+
+      def success_message(ticket_id, values, writeback = :skipped)
+        base = base_success_message(ticket_id, values)
+        return base unless writeback.is_a?(Array) && writeback.first == :failed
+
+        "#{base} (warning: could not store the ticket id on the record: #{writeback.last})"
+      end
+
+      def base_success_message(ticket_id, values)
+        if truthy?(values['Send as internal note'])
+          return 'Ticket created (internal note).' unless ticket_id
+
+          "Ticket ##{ticket_id} created (internal note, no email)."
+        else
+          return 'Ticket created and requester notified.' unless ticket_id
+
+          "Ticket ##{ticket_id} created and requester notified."
+        end
+      end
+
+      def truthy?(value)
+        value == true || value.to_s.casecmp('true').zero?
+      end
+
+      def present?(value)
+        !value.nil? && value.to_s != ''
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/create_ticket_with_notification/form_builder.rb b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/create_ticket_with_notification/form_builder.rb
new file mode 100644
index 000000000..527ef985a
--- /dev/null
+++ b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/plugins/create_ticket_with_notification/form_builder.rb
@@ -0,0 +1,149 @@
+require 'cgi'
+
+module ForestAdminDatasourceZendesk
+  module Plugins
+    class CreateTicketWithNotification
+      module FormBuilder
+        FieldType = ForestAdminDatasourceCustomizer::Decorators::Action::Types::FieldType
+
+        NO_TEMPLATE = 'No template'.freeze
+        TOKEN_RE = /\{\{\s*record\.([a-zA-Z_][a-zA-Z0-9_]*)\s*\}\}/
+
+        module_function
+
+        # ActionCollectionDecorator rejects forms that mix Page elements with
+        # non-Page elements, so each mode (flat / wizard) stays homogeneous.
+        def build(opts)
+          body = body_fields(opts)
+          return body if opts[:email_templates].empty?
+
+          [
+            { type: 'Layout', component: 'Page', next_button_label: 'Continue',
+              elements: [template_field(opts[:email_templates])] },
+            { type: 'Layout', component: 'Page', previous_button_label: 'Back',
+              elements: body }
+          ]
+        end
+
+        def body_fields(opts)
+          fields = [requester_field(opts[:requester_email_default]),
+                    subject_field(opts[:default_subject]),
+                    message_field(opts[:default_message], opts[:email_templates])]
+          fields << priority_field unless present?(opts[:priority_override])
+          fields << type_field unless present?(opts[:type_override])
+          fields << internal_note_field
+          fields
+        end
+
+        def requester_field(default)
+          { type: FieldType::STRING, label: 'Requester email', is_required: true,
+            description: 'Email of the Zendesk requester. Pre-filled from the selected record when available.',
+            default_value: requester_default(default) }
+        end
+
+        def template_field(templates)
+          { type: FieldType::ENUM, label: 'Template', is_required: true,
+            enum_values: [NO_TEMPLATE] + templates.map { |t| t[:title] },
+            default_value: NO_TEMPLATE,
+            description: 'Pick a template to pre-fill the Message on the next page.' }
+        end
+
+        def subject_field(default_subject)
+          { type: FieldType::STRING, label: 'Subject', is_required: true,
+            default_value: template_default(default_subject, escape_html: false) }
+        end
+
+        def message_field(default_message, templates)
+          field = { type: FieldType::STRING, label: 'Message', widget: 'RichText', is_required: true,
+                    description: 'Sent as the ticket\'s first comment (HTML). Public comments trigger the ' \
+                                 'default Zendesk notification email to the requester.' }
+          return field.merge(default_value: template_default(default_message, escape_html: true)) if templates.empty?
+
+          # `value:` (not `default_value:`) — drop_default runs once (data
+          # key sticks after the first render); drop_deferred re-evaluates
+          # on every fetch, so Template changes re-fire the message proc.
+          field.merge(value: message_value(templates))
+        end
+
+        def priority_field
+          { type: FieldType::ENUM, label: 'Priority',
+            enum_values: TicketEnums::PRIORITY, default_value: 'normal' }
+        end
+
+        def type_field
+          { type: FieldType::ENUM, label: 'Type', enum_values: TicketEnums::TYPE }
+        end
+
+        def internal_note_field
+          { type: FieldType::BOOLEAN, label: 'Send as internal note',
+            description: 'When checked, the first comment is private and no email is sent to the requester.',
+            default_value: false }
+        end
+
+        def requester_default(value)
+          return nil if value.nil?
+          return value if value.is_a?(String)
+
+          lambda do |context|
+            record = fetch_record(context)
+            record.empty? ? nil : value.call(record)
+          rescue StandardError => e
+            ForestAdminDatasourceZendesk.logger.warn(
+              "[forest_admin_datasource_zendesk] requester_email_default resolver raised: #{e.class}: #{e.message}"
+            )
+            nil
+          end
+        end
+
+        def template_default(template, escape_html:)
+          return nil unless present?(template)
+          return template unless template.match?(TOKEN_RE)
+
+          ->(context) { interpolate(template, fetch_record(context), escape_html: escape_html) }
+        end
+
+        # Returns nil unless Template was just changed, so set_watch_changes
+        # carries over the user's current Message edits between renders.
+        def message_value(templates)
+          by_title = templates.to_h { |t| [t[:title], t[:content].to_s] }
+          lambda do |context|
+            return nil unless context.field_changed?('Template')
+
+            title = context.get_form_value('Template')
+            return '' if title == NO_TEMPLATE
+
+            content = by_title[title].to_s
+            return content unless content.match?(TOKEN_RE)
+
+            interpolate(content, fetch_record(context), escape_html: true)
+          end
+        end
+
+        def fetch_record(context)
+          context.get_record([]) || {}
+        rescue StandardError => e
+          ForestAdminDatasourceZendesk.logger.warn(
+            "[forest_admin_datasource_zendesk] failed to fetch record for token interpolation: #{e.class}: #{e.message}"
+          )
+          {}
+        end
+
+        # Message ships as html_body — unescaped `<` or `&` from a record
+        # value would break the outbound email or smuggle markup into it.
+        def interpolate(template, record, escape_html:)
+          template.gsub(TOKEN_RE) do
+            key = ::Regexp.last_match(1)
+            value = record[key] || record[key.to_sym]
+            next '' if value.nil?
+
+            escape_html ? CGI.escapeHTML(value.to_s) : value.to_s
+          end
+        end
+
+        def present?(value)
+          !value.nil? && value.to_s != ''
+        end
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/ticket_enums.rb b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/ticket_enums.rb
new file mode 100644
index 000000000..f34776664
--- /dev/null
+++ b/packages/forest_admin_datasource_zendesk/lib/forest_admin_datasource_zendesk/ticket_enums.rb
@@ -0,0 +1,8 @@
+module ForestAdminDatasourceZendesk
+  # Shared between the Ticket schema and plugins that build ticket forms.
+  module TicketEnums
+    STATUS   = %w[new open pending hold solved closed].freeze
+    PRIORITY = %w[low normal high urgent].freeze
+    TYPE     = %w[problem incident question task].freeze
+  end
+end
diff --git a/packages/forest_admin_datasource_zendesk/spec/forest_admin_datasource_zendesk/datasource_spec.rb b/packages/forest_admin_datasource_zendesk/spec/forest_admin_datasource_zendesk/datasource_spec.rb
index c72afda36..6e0f47871 100644
--- a/packages/forest_admin_datasource_zendesk/spec/forest_admin_datasource_zendesk/datasource_spec.rb
+++ b/packages/forest_admin_datasource_zendesk/spec/forest_admin_datasource_zendesk/datasource_spec.rb
@@ -29,6 +29,12 @@
     )
   end
 
+  it 'registers no smart actions by default (actions are opt-in via plugins)' do
+    ds = described_class.new(**valid_args)
+    expect(ds.get_collection('ZendeskTicket').schema[:actions]).to be_empty
+    expect(ds.get_collection('ZendeskUser').schema[:actions]).to be_empty
+  end
+
   it 'forwards discovered ticket custom fields into the Ticket schema' do
     stub_request(:get, "#{base}/ticket_fields")
       .to_return(status: 200,
diff --git a/packages/forest_admin_datasource_zendesk/spec/forest_admin_datasource_zendesk/plugins/close_ticket_spec.rb b/packages/forest_admin_datasource_zendesk/spec/forest_admin_datasource_zendesk/plugins/close_ticket_spec.rb
new file mode 100644
index 000000000..8f9fc1c2c
--- /dev/null
+++ b/packages/forest_admin_datasource_zendesk/spec/forest_admin_datasource_zendesk/plugins/close_ticket_spec.rb
@@ -0,0 +1,270 @@
+module ForestAdminDatasourceZendesk
+  # Stand-in for an action context. The executor only reads
+  # `get_records(fields)` so we don't need the full ActionContext class here.
+  class FakeCloseContext
+    def initialize(records: [])
+      @records = records
+    end
+
+    def get_records(_fields = [])
+      @records
+    end
+  end
+
+  RSpec.describe Plugins::CloseTicket do
+    let(:client) { instance_double(ForestAdminDatasourceZendesk::Client) }
+    let(:datasource) do
+      instance_double(ForestAdminDatasourceZendesk::Datasource, client: client, custom_field_mapping: {})
+    end
+    let(:result_builder) { ForestAdminDatasourceCustomizer::Decorators::Action::ResultBuilder.new }
+    let(:action_scope) { ForestAdminDatasourceCustomizer::Decorators::Action::Types::ActionScope }
+    let(:ticket_id_field) { 'last_zendesk_ticket_id' }
+    let(:collection_customizer) do
+      Class.new do
+        attr_reader :registered
+
+        def initialize = @registered = {}
+        def add_action(name, action) = @registered[name] = action
+      end.new
+    end
+
+    def register(opts = {})
+      described_class.new.run(nil, collection_customizer,
+                              { datasource: datasource, ticket_id_field: ticket_id_field }.merge(opts))
+      collection_customizer.registered
+    end
+
+    describe '#run' do
+      it 'registers all four variants by default (solved/closed × single/bulk)' do
+        register
+
+        expect(collection_customizer.registered.keys).to contain_exactly(
+          'Mark Zendesk ticket as solved', 'Mark selected Zendesk tickets as solved',
+          'Mark Zendesk ticket as closed', 'Mark selected Zendesk tickets as closed'
+        )
+      end
+
+      it 'honors :statuses to keep only the requested status family' do
+        register(statuses: %w[solved])
+        expect(collection_customizer.registered.keys).to contain_exactly(
+          'Mark Zendesk ticket as solved', 'Mark selected Zendesk tickets as solved'
+        )
+      end
+
+      it 'honors :scopes to keep only the requested scopes' do
+        register(scopes: %i[single])
+        expect(collection_customizer.registered.keys).to contain_exactly(
+          'Mark Zendesk ticket as solved', 'Mark Zendesk ticket as closed'
+        )
+      end
+
+      it 'composes both :statuses and :scopes to a Cartesian subset (one action)' do
+        register(statuses: %w[closed], scopes: %i[bulk])
+        expect(collection_customizer.registered.keys).to contain_exactly('Mark selected Zendesk tickets as closed')
+      end
+
+      it 'accepts symbol statuses and string scopes interchangeably' do
+        register(statuses: %i[solved], scopes: %w[bulk])
+        expect(collection_customizer.registered.keys).to contain_exactly('Mark selected Zendesk tickets as solved')
+      end
+
+      it 'binds the right ActionScope to each registered action' do
+        register
+        registered = collection_customizer.registered
+        expect(registered['Mark Zendesk ticket as solved'].scope).to eq(action_scope::SINGLE)
+        expect(registered['Mark selected Zendesk tickets as solved'].scope).to eq(action_scope::BULK)
+        expect(registered['Mark Zendesk ticket as closed'].scope).to eq(action_scope::SINGLE)
+        expect(registered['Mark selected Zendesk tickets as closed'].scope).to eq(action_scope::BULK)
+      end
+
+      it 'raises a ForestException on an unknown status' do
+        expect { register(statuses: %w[solved unknown]) }
+          .to raise_error(ForestAdminDatasourceToolkit::Exceptions::ForestException, /Unknown.*unknown/)
+      end
+
+      it 'raises a ForestException on an unknown scope' do
+        expect { register(scopes: %i[single weird]) }
+          .to raise_error(ForestAdminDatasourceToolkit::Exceptions::ForestException, /Unknown.*weird/)
+      end
+
+      it 'raises ArgumentError without :datasource' do
+        expect { described_class.new.run(nil, collection_customizer, ticket_id_field: 'id') }
+          .to raise_error(ArgumentError, /datasource/)
+      end
+
+      it 'raises ArgumentError without :ticket_id_field' do
+        expect { described_class.new.run(nil, collection_customizer, datasource: datasource) }
+          .to raise_error(ArgumentError, /ticket_id_field/)
+      end
+
+      it 'raises ArgumentError without a collection_customizer' do
+        expect do
+          described_class.new.run(nil, nil, datasource: datasource, ticket_id_field: 'id')
+        end.to raise_error(ArgumentError, /collection/)
+      end
+    end
+
+    describe 'executor' do
+      let(:solved_single) do
+        register(statuses: %w[solved], scopes: %i[single])['Mark Zendesk ticket as solved']
+      end
+      let(:closed_bulk) do
+        register(statuses: %w[closed], scopes: %i[bulk])['Mark selected Zendesk tickets as closed']
+      end
+
+      it 'reads the ticket id from the host field and PUTs status=solved' do
+        allow(client).to receive(:update_ticket)
+        context = FakeCloseContext.new(records: [{ ticket_id_field => 42 }])
+
+        result = solved_single.execute.call(context, result_builder)
+
+        expect(client).to have_received(:update_ticket).with(42, 'status' => 'solved')
+        expect(result[:type]).to eq('Success')
+        expect(result[:message]).to include('Ticket #42', 'marked as solved')
+      end
+
+      it 'PUTs status=closed for every host record when run in bulk' do
+        allow(client).to receive(:update_ticket)
+        bulk_context = FakeCloseContext.new(
+          records: [7, 8, 9].map { |id| { ticket_id_field => id } }
+        )
+
+        result = closed_bulk.execute.call(bulk_context, result_builder)
+
+        [7, 8, 9].each { |id| expect(client).to have_received(:update_ticket).with(id, 'status' => 'closed') }
+        expect(result[:message]).to include('3 tickets closed')
+      end
+
+      it 'returns an error when no host record carries a ticket id' do
+        allow(client).to receive(:update_ticket)
+        context = FakeCloseContext.new(records: [{ ticket_id_field => nil }])
+
+        result = solved_single.execute.call(context, result_builder)
+
+        expect(client).not_to have_received(:update_ticket)
+        expect(result[:type]).to eq('Error')
+        expect(result[:message]).to include(ticket_id_field)
+      end
+
+      it 'works with symbol keys on the host record' do
+        allow(client).to receive(:update_ticket)
+        context = FakeCloseContext.new(records: [{ ticket_id_field.to_sym => 99 }])
+
+        solved_single.execute.call(context, result_builder)
+
+        expect(client).to have_received(:update_ticket).with(99, 'status' => 'solved')
+      end
+
+      context 'when Zendesk rejects some ids (partial success on bulk)' do
+        let(:bulk_context) do
+          FakeCloseContext.new(records: [7, 8, 9].map { |id| { ticket_id_field => id } })
+        end
+
+        it 'continues with the remaining ids and surfaces the failures in the message' do
+          allow(client).to receive(:update_ticket)
+            .with(8, anything).and_raise(StandardError, 'cannot transition open to closed')
+          allow(client).to receive(:update_ticket).with(7, anything)
+          allow(client).to receive(:update_ticket).with(9, anything)
+          allow(ForestAdminDatasourceZendesk.logger).to receive(:warn)
+
+          result = closed_bulk.execute.call(bulk_context, result_builder)
+
+          expect(result[:type]).to eq('Success')
+          expect(result[:message]).to include('2 tickets closed', '1 failed', '8')
+          expect(ForestAdminDatasourceZendesk.logger).to have_received(:warn)
+            .with(a_string_including('#8', 'cannot transition'))
+        end
+
+        it 'returns an Error when every id fails' do
+          allow(client).to receive(:update_ticket).and_raise(StandardError, 'permission denied')
+          allow(ForestAdminDatasourceZendesk.logger).to receive(:warn).exactly(3).times
+
+          result = closed_bulk.execute.call(bulk_context, result_builder)
+
+          expect(result[:type]).to eq('Error')
+          expect(result[:message]).to include('Failed to close', '3 tickets', 'permission denied')
+        end
+      end
+
+      context 'when get_records itself raises' do
+        it 'logs and returns an Error message without calling the client' do
+          context = instance_double(
+            ForestAdminDatasourceCustomizer::Decorators::Action::Context::ActionContextSingle
+          )
+          allow(context).to receive(:get_records).and_raise(StandardError, 'boom')
+          allow(ForestAdminDatasourceZendesk.logger).to receive(:warn)
+          allow(client).to receive(:update_ticket)
+
+          result = solved_single.execute.call(context, result_builder)
+
+          expect(client).not_to have_received(:update_ticket)
+          expect(result[:type]).to eq('Error')
+          expect(ForestAdminDatasourceZendesk.logger).to have_received(:warn)
+            .with(a_string_including(ticket_id_field, 'boom'))
+        end
+      end
+    end
+
+    describe "Zendesk's 'closed prevents ticket update' error" do
+      let(:already_closed_error) do
+        StandardError.new(
+          'Zendesk API call failed: update(tickets/254): ZendeskAPI::Error::RecordInvalid: ' \
+          '{"status" => [{"description" => "closed prevents ticket update"}]}'
+        )
+      end
+      let(:solved_single) do
+        register(statuses: %w[solved], scopes: %i[single])['Mark Zendesk ticket as solved']
+      end
+      let(:closed_single) do
+        register(statuses: %w[closed], scopes: %i[single])['Mark Zendesk ticket as closed']
+      end
+      let(:closed_bulk) do
+        register(statuses: %w[closed], scopes: %i[bulk])['Mark selected Zendesk tickets as closed']
+      end
+
+      it "with status='closed' on an already-closed ticket: clean Success ('was already closed')" do
+        allow(client).to receive(:update_ticket).and_raise(already_closed_error)
+        allow(ForestAdminDatasourceZendesk.logger).to receive(:warn)
+        context = FakeCloseContext.new(records: [{ ticket_id_field => 254 }])
+
+        result = closed_single.execute.call(context, result_builder)
+
+        expect(result[:type]).to eq('Success')
+        expect(result[:message]).to eq('Ticket #254 was already closed.')
+        # No warn log: an idempotent "already closed" is an expected state.
+        expect(ForestAdminDatasourceZendesk.logger).not_to have_received(:warn)
+      end
+
+      it "with status='closed' in bulk: mixes succeeded + already-closed + other failures cleanly" do
+        allow(client).to receive(:update_ticket).with(7, anything)
+        allow(client).to receive(:update_ticket).with(8, anything).and_raise(already_closed_error)
+        allow(client).to receive(:update_ticket).with(9, anything).and_raise(StandardError, 'permission denied')
+        allow(ForestAdminDatasourceZendesk.logger).to receive(:warn)
+        context = FakeCloseContext.new(records: [7, 8, 9].map { |id| { ticket_id_field => id } })
+
+        result = closed_bulk.execute.call(context, result_builder)
+
+        expect(result[:type]).to eq('Success')
+        expect(result[:message]).to include('Ticket #7 closed.', 'Ticket #8 was already closed.', '1 failed: 9')
+        # Only the genuine failure is logged; "already closed" stays quiet.
+        expect(ForestAdminDatasourceZendesk.logger).to have_received(:warn)
+          .with(a_string_including('#9', 'permission denied'))
+        expect(ForestAdminDatasourceZendesk.logger).not_to have_received(:warn)
+          .with(a_string_including('#8'))
+      end
+
+      it "with status='solved' on an already-closed ticket: Error explaining it can't be reopened" do
+        allow(client).to receive(:update_ticket).and_raise(already_closed_error)
+        context = FakeCloseContext.new(records: [{ ticket_id_field => 254 }])
+
+        result = solved_single.execute.call(context, result_builder)
+
+        expect(result[:type]).to eq('Error')
+        expect(result[:message]).to include('#254', 'already closed', 'cannot reopen')
+        # Make sure the raw API stack is gone.
+        expect(result[:message]).not_to include('RecordInvalid')
+        expect(result[:message]).not_to include('"description"')
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_zendesk/spec/forest_admin_datasource_zendesk/plugins/create_ticket_with_notification_spec.rb b/packages/forest_admin_datasource_zendesk/spec/forest_admin_datasource_zendesk/plugins/create_ticket_with_notification_spec.rb
new file mode 100644
index 000000000..3399e7624
--- /dev/null
+++ b/packages/forest_admin_datasource_zendesk/spec/forest_admin_datasource_zendesk/plugins/create_ticket_with_notification_spec.rb
@@ -0,0 +1,454 @@
+module ForestAdminDatasourceZendesk
+  # Tiny PORO standing in for an ActionContextSingle in unit tests. We can't
+  # use Struct here because `Struct` mixes in Enumerable, which already defines
+  # `#filter` — having a `:filter` member triggers Lint/StructNewOverride and
+  # is genuinely ambiguous.
+  class FakeActionContext
+    attr_reader :form_values, :collection, :filter, :record_id
+
+    def initialize(form_values: nil, collection: nil, filter: nil, record_id: nil)
+      @form_values = form_values
+      @collection = collection
+      @filter = filter
+      @record_id = record_id
+    end
+  end
+
+  RSpec.describe Plugins::CreateTicketWithNotification do
+    let(:client) { instance_double(ForestAdminDatasourceZendesk::Client) }
+    let(:datasource) do
+      instance_double(ForestAdminDatasourceZendesk::Datasource, client: client, custom_field_mapping: {})
+    end
+    let(:collection_customizer) do
+      Class.new do
+        attr_reader :registered
+
+        def initialize = @registered = {}
+        def add_action(name, action) = @registered[name] = action
+      end.new
+    end
+    let(:result_builder) { ForestAdminDatasourceCustomizer::Decorators::Action::ResultBuilder.new }
+
+    def register(opts = {})
+      described_class.new.run(nil, collection_customizer, { datasource: datasource }.merge(opts))
+      collection_customizer.registered[opts[:action_name] || described_class::NAME]
+    end
+
+    describe '#run' do
+      it 'registers a SINGLE-scoped action under the default name with the documented form fields' do
+        action = register
+
+        expect(collection_customizer.registered.keys).to contain_exactly(described_class::NAME)
+        expect(action.scope).to eq(ForestAdminDatasourceCustomizer::Decorators::Action::Types::ActionScope::SINGLE)
+        labels = action.form.map { |f| f[:label] }
+        expect(labels).to eq(['Requester email', 'Subject', 'Message', 'Priority', 'Type', 'Send as internal note'])
+      end
+
+      it 'honors :action_name as a custom label' do
+        register
+        register(action_name: 'Custom label')
+
+        expect(collection_customizer.registered.keys).to contain_exactly(described_class::NAME, 'Custom label')
+      end
+
+      it 'raises ArgumentError without :datasource' do
+        expect { described_class.new.run(nil, collection_customizer, {}) }
+          .to raise_error(ArgumentError, /datasource/)
+      end
+
+      it 'raises ArgumentError without a collection_customizer' do
+        expect { described_class.new.run(nil, nil, datasource: datasource) }
+          .to raise_error(ArgumentError, /collection/)
+      end
+    end
+
+    describe 'requester email field' do
+      let(:context_double) do
+        instance_double(ForestAdminDatasourceCustomizer::Decorators::Action::Context::ActionContextSingle,
+                        get_record: { 'id' => 42, 'email' => 'alice@x.com', 'name' => 'Alice' })
+      end
+
+      it 'uses RichText for the Message widget and marks Requester email required' do
+        action = register
+        message_field = action.form.find { |f| f[:label] == 'Message' }
+        expect(message_field[:widget]).to eq('RichText')
+        expect(action.form.first[:label]).to eq('Requester email')
+        expect(action.form.first[:is_required]).to be(true)
+      end
+
+      it 'leaves the field empty when no requester_email_default is configured' do
+        expect(register.form.first[:default_value]).to be_nil
+      end
+
+      context 'with a literal email String' do
+        it 'uses it verbatim as a static default (no record lookup)' do
+          action = register(requester_email_default: 'support@example.com')
+          expect(action.form.first[:default_value]).to eq('support@example.com')
+        end
+      end
+
+      context 'with a Proc resolver' do
+        let(:resolver) { ->(record) { record['email'] } }
+
+        it 'pre-fills the email field from the selected record via the resolver' do
+          action = register(requester_email_default: resolver)
+          expect(action.form.first[:default_value].call(context_double)).to eq('alice@x.com')
+        end
+
+        it 'returns nil (and logs) when the record lookup raises' do
+          allow(context_double).to receive(:get_record).and_raise(StandardError, 'boom')
+          allow(ForestAdminDatasourceZendesk.logger).to receive(:warn)
+
+          action = register(requester_email_default: resolver)
+          expect(action.form.first[:default_value].call(context_double)).to be_nil
+          expect(ForestAdminDatasourceZendesk.logger).to have_received(:warn)
+            .with(a_string_including('fetch record', 'StandardError', 'boom'))
+        end
+
+        it 'returns nil (and logs) when the resolver proc itself raises' do
+          allow(context_double).to receive(:get_record).and_return({ 'email' => 'a@b.com' })
+          allow(ForestAdminDatasourceZendesk.logger).to receive(:warn)
+          broken_resolver = ->(_record) { raise StandardError, 'typo in lambda' }
+
+          action = register(requester_email_default: broken_resolver)
+          expect(action.form.first[:default_value].call(context_double)).to be_nil
+          expect(ForestAdminDatasourceZendesk.logger).to have_received(:warn)
+            .with(a_string_including('requester_email_default', 'typo in lambda'))
+        end
+      end
+    end
+
+    describe 'subject / message defaults' do
+      it 'uses configured static defaults verbatim' do
+        action = register(default_subject: 'Welcome', default_message: '<p>Hi</p>')
+        expect(action.form.find { |f| f[:label] == 'Subject' }[:default_value]).to eq('Welcome')
+        expect(action.form.find { |f| f[:label] == 'Message' }[:default_value]).to eq('<p>Hi</p>')
+      end
+
+      context 'with {{record.<field>}} tokens' do
+        let(:context_double) do
+          instance_double(ForestAdminDatasourceCustomizer::Decorators::Action::Context::ActionContextSingle,
+                          get_record: { 'id' => 42, 'name' => 'Alice', 'email' => 'alice@x.com' })
+        end
+
+        it 'substitutes record fields at form-open time' do
+          action = register(default_subject: 'Follow-up for {{record.name}}',
+                            default_message: '<p>Hello {{record.name}} ({{record.email}})</p>')
+          subject_proc = action.form.find { |f| f[:label] == 'Subject' }[:default_value]
+          message_proc = action.form.find { |f| f[:label] == 'Message' }[:default_value]
+
+          expect(subject_proc.call(context_double)).to eq('Follow-up for Alice')
+          expect(message_proc.call(context_double)).to eq('<p>Hello Alice (alice@x.com)</p>')
+        end
+
+        it 'falls back to empty strings when the record lookup fails (logged)' do
+          allow(context_double).to receive(:get_record).and_raise(StandardError, 'boom')
+          allow(ForestAdminDatasourceZendesk.logger).to receive(:warn)
+
+          action = register(default_subject: 'Hi {{record.name}}')
+          expect(action.form.find { |f| f[:label] == 'Subject' }[:default_value].call(context_double)).to eq('Hi ')
+          expect(ForestAdminDatasourceZendesk.logger).to have_received(:warn)
+            .with(a_string_including('fetch record', 'boom'))
+        end
+      end
+    end
+
+    describe 'HTML escaping on the Message template' do
+      let(:context_double) do
+        instance_double(ForestAdminDatasourceCustomizer::Decorators::Action::Context::ActionContextSingle,
+                        get_record: { 'name' => "O'Brien <admin>", 'note' => '<script>alert(1)</script>' })
+      end
+
+      it 'escapes interpolated values in the Message (RichText -> html_body)' do
+        action = register(default_message: '<p>Hi {{record.name}} - {{record.note}}</p>')
+        message_proc = action.form.find { |f| f[:label] == 'Message' }[:default_value]
+
+        expect(message_proc.call(context_double)).to eq(
+          '<p>Hi O&#39;Brien &lt;admin&gt; - &lt;script&gt;alert(1)&lt;/script&gt;</p>'
+        )
+      end
+
+      it 'leaves Subject interpolation unescaped (plain-text field)' do
+        action = register(default_subject: 'Re: {{record.name}}')
+        subject_proc = action.form.find { |f| f[:label] == 'Subject' }[:default_value]
+
+        expect(subject_proc.call(context_double)).to eq("Re: O'Brien <admin>")
+      end
+    end
+
+    describe 'executor' do
+      let(:context) { Struct.new(:form_values).new(form_values) }
+
+      context 'with a public html comment (default)' do
+        let(:form_values) do
+          { 'Requester email' => 'alice@x.com', 'Subject' => 'Refund', 'Message' => 'Hi there',
+            'Priority' => 'high', 'Type' => 'question', 'Send as internal note' => false }
+        end
+
+        it 'creates a ticket targeting the requester by email and embeds the html comment' do
+          allow(client).to receive(:create_ticket) do |payload|
+            expect(payload).to eq(
+              'requester' => { 'email' => 'alice@x.com', 'name' => 'alice' },
+              'subject' => 'Refund',
+              'comment' => { 'html_body' => 'Hi there', 'public' => true },
+              'priority' => 'high',
+              'type' => 'question'
+            )
+            { 'id' => 7 }
+          end
+
+          result = register.execute.call(context, result_builder)
+          expect(client).to have_received(:create_ticket)
+          expect(result[:message]).to include('Ticket #7', 'notified')
+        end
+      end
+
+      context 'with internal note' do
+        let(:form_values) do
+          { 'Requester email' => 'b@x.com', 'Subject' => 'Internal', 'Message' => 'For agents only',
+            'Send as internal note' => true }
+        end
+
+        it 'flips the comment to private and omits the notify wording' do
+          allow(client).to receive(:create_ticket).and_return('id' => 9)
+
+          result = register.execute.call(context, result_builder)
+          expect(client).to have_received(:create_ticket).with(hash_including(
+                                                                 'comment' => { 'html_body' => 'For agents only',
+                                                                                'public' => false }
+                                                               ))
+          expect(result[:message]).to include('internal note')
+        end
+      end
+
+      context 'without a requester email' do
+        let(:form_values) { { 'Requester email' => nil, 'Subject' => 'S', 'Message' => 'M' } }
+
+        it 'returns an error and does not call the client' do
+          allow(client).to receive(:create_ticket)
+
+          result = register.execute.call(context, result_builder)
+          expect(result[:type]).to eq('Error')
+          expect(client).not_to have_received(:create_ticket)
+        end
+      end
+
+      context 'with empty optional fields' do
+        let(:form_values) do
+          { 'Requester email' => 'c@x.com', 'Subject' => 'S', 'Message' => 'M',
+            'Priority' => nil, 'Type' => '', 'Send as internal note' => nil }
+        end
+
+        it 'omits empty optional keys from the payload' do
+          allow(client).to receive(:create_ticket) do |payload|
+            expect(payload.keys).not_to include('priority', 'type')
+            expect(payload['comment']['public']).to be(true)
+            { 'id' => 1 }
+          end
+
+          register.execute.call(context, result_builder)
+          expect(client).to have_received(:create_ticket)
+        end
+      end
+
+      context 'with ticket_id_field configured (writeback to host record)' do
+        let(:form_values) do
+          { 'Requester email' => 'd@x.com', 'Subject' => 'S', 'Message' => 'M',
+            'Send as internal note' => false }
+        end
+        let(:host_collection) { instance_double('RelaxedCollection') } # rubocop:disable RSpec/VerifiedDoubleReference
+        let(:filter) { instance_double('Filter') } # rubocop:disable RSpec/VerifiedDoubleReference
+        let(:context) do
+          FakeActionContext.new(form_values: form_values, collection: host_collection, filter: filter)
+        end
+
+        before { allow(client).to receive(:create_ticket).and_return('id' => 77) }
+
+        it 'updates the host record with the new ticket id under the configured field' do
+          allow(host_collection).to receive(:update)
+          action = register(ticket_id_field: 'last_zendesk_ticket_id')
+
+          result = action.execute.call(context, result_builder)
+
+          expect(host_collection).to have_received(:update).with(filter, { 'last_zendesk_ticket_id' => 77 })
+          expect(result[:type]).to eq('Success')
+          expect(result[:message]).to include('Ticket #77')
+          expect(result[:message]).not_to include('warning')
+        end
+
+        it 'logs and surfaces a warning when the host update fails but still succeeds' do
+          allow(host_collection).to receive(:update).and_raise(StandardError, 'field is read-only')
+          allow(ForestAdminDatasourceZendesk.logger).to receive(:warn)
+          action = register(ticket_id_field: 'last_zendesk_ticket_id')
+
+          result = action.execute.call(context, result_builder)
+
+          expect(result[:type]).to eq('Success')
+          expect(result[:message]).to include('Ticket #77', 'warning', 'field is read-only')
+          expect(ForestAdminDatasourceZendesk.logger).to have_received(:warn)
+            .with(a_string_including('last_zendesk_ticket_id', 'field is read-only'))
+        end
+
+        it 'does not attempt any update when ticket_id_field is omitted' do
+          allow(host_collection).to receive(:update)
+
+          register.execute.call(context, result_builder)
+
+          expect(host_collection).not_to have_received(:update)
+        end
+      end
+    end
+
+    describe 'email_templates wizard' do
+      let(:templates) do
+        [{ title: 'Welcome', content: '<p>Welcome aboard!</p>' },
+         { title: 'Refund',  content: '<p>Refund processed.</p>' }]
+      end
+
+      it 'flips the form into a two-page wizard (Template first, body second)' do
+        action = register(email_templates: templates)
+
+        expect(action.form.size).to eq(2)
+        page_one, page_two = action.form
+        expect(page_one[:component]).to eq('Page')
+        expect(page_one[:elements].map { |f| f[:label] }).to eq(['Template'])
+        expect(page_two[:elements].map { |f| f[:label] })
+          .to eq(['Requester email', 'Subject', 'Message', 'Priority', 'Type', 'Send as internal note'])
+      end
+
+      it 'lists No template + each template title in the dropdown' do
+        template_field = register(email_templates: templates).form.first[:elements].first
+        expect(template_field[:enum_values]).to eq(['No template', 'Welcome', 'Refund'])
+        expect(template_field[:default_value]).to eq('No template')
+      end
+
+      it 'pre-fills Message via `value:` when Template was the just-changed field' do
+        message_field = register(email_templates: templates).form.last[:elements].find { |f| f[:label] == 'Message' }
+        ctx = instance_double(ForestAdminDatasourceCustomizer::Decorators::Action::Context::ActionContextSingle,
+                              field_changed?: true, get_form_value: 'Refund')
+
+        expect(message_field[:value].call(ctx)).to eq('<p>Refund processed.</p>')
+      end
+
+      it "yields an empty Message when 'No template' was just selected" do
+        message_field = register(email_templates: templates).form.last[:elements].find { |f| f[:label] == 'Message' }
+        ctx = instance_double(ForestAdminDatasourceCustomizer::Decorators::Action::Context::ActionContextSingle,
+                              field_changed?: true, get_form_value: 'No template')
+
+        expect(message_field[:value].call(ctx)).to eq('')
+      end
+
+      it 'returns nil (carry over current input) when another field triggered the re-fetch' do
+        message_field = register(email_templates: templates).form.last[:elements].find { |f| f[:label] == 'Message' }
+        ctx = instance_double(ForestAdminDatasourceCustomizer::Decorators::Action::Context::ActionContextSingle,
+                              field_changed?: false)
+
+        expect(message_field[:value].call(ctx)).to be_nil
+      end
+
+      it 'interpolates {{record.<field>}} tokens inside the selected template content' do
+        templated = [{ title: 'Welcome', content: '<p>Hi {{record.name}} ({{record.email}})</p>' }]
+        message_field = register(email_templates: templated).form.last[:elements].find { |f| f[:label] == 'Message' }
+        ctx = instance_double(ForestAdminDatasourceCustomizer::Decorators::Action::Context::ActionContextSingle,
+                              field_changed?: true, get_form_value: 'Welcome',
+                              get_record: { 'name' => 'Alice', 'email' => 'a@b.com' })
+
+        expect(message_field[:value].call(ctx)).to eq('<p>Hi Alice (a@b.com)</p>')
+      end
+
+      it 'HTML-escapes interpolated record values inside template content' do
+        templated = [{ title: 'Bug', content: '<p>{{record.note}}</p>' }]
+        message_field = register(email_templates: templated).form.last[:elements].find { |f| f[:label] == 'Message' }
+        ctx = instance_double(ForestAdminDatasourceCustomizer::Decorators::Action::Context::ActionContextSingle,
+                              field_changed?: true, get_form_value: 'Bug',
+                              get_record: { 'note' => '<script>x</script>' })
+
+        expect(message_field[:value].call(ctx)).to eq('<p>&lt;script&gt;x&lt;/script&gt;</p>')
+      end
+
+      it 'keeps the flat form when no templates are configured' do
+        expect(register(email_templates: []).form.first[:component]).to be_nil # not a Page
+      end
+    end
+
+    describe 'priority_override / type_override' do
+      it 'omits the Priority field and forces the value in the payload' do
+        allow(client).to receive(:create_ticket).and_return('id' => 1)
+
+        action = register(priority_override: 'urgent')
+        expect(action.form.map { |f| f[:label] }).not_to include('Priority')
+
+        action.execute.call(
+          FakeActionContext.new(form_values: { 'Requester email' => 'a@b.com',
+                                               'Subject' => 'S', 'Message' => 'M' }),
+          result_builder
+        )
+        expect(client).to have_received(:create_ticket).with(hash_including('priority' => 'urgent'))
+      end
+
+      it 'omits the Type field and forces the value in the payload' do
+        allow(client).to receive(:create_ticket).and_return('id' => 1)
+
+        action = register(type_override: 'incident')
+        expect(action.form.map { |f| f[:label] }).not_to include('Type')
+
+        action.execute.call(
+          FakeActionContext.new(form_values: { 'Requester email' => 'a@b.com',
+                                               'Subject' => 'S', 'Message' => 'M' }),
+          result_builder
+        )
+        expect(client).to have_received(:create_ticket).with(hash_including('type' => 'incident'))
+      end
+
+      it 'forces the override even when the form value is also present' do
+        allow(client).to receive(:create_ticket).and_return('id' => 1)
+        register(priority_override: 'urgent').execute.call(
+          FakeActionContext.new(form_values: { 'Requester email' => 'a@b.com', 'Subject' => 'S',
+                                               'Message' => 'M', 'Priority' => 'low' }),
+          result_builder
+        )
+        expect(client).to have_received(:create_ticket).with(hash_including('priority' => 'urgent'))
+      end
+    end
+
+    describe 'requester name auto-derivation' do
+      it 'sends the email local-part as requester.name in the payload' do
+        allow(client).to receive(:create_ticket).and_return('id' => 1)
+        register.execute.call(
+          FakeActionContext.new(form_values: { 'Requester email' => 'john.doe@acme.com',
+                                               'Subject' => 'S', 'Message' => 'M' }),
+          result_builder
+        )
+        expect(client).to have_received(:create_ticket).with(hash_including(
+                                                               'requester' => { 'email' => 'john.doe@acme.com',
+                                                                                'name' => 'john.doe' }
+                                                             ))
+      end
+    end
+
+    describe 'sender_email' do
+      it 'maps to Zendesk `recipient` in the payload when configured' do
+        allow(client).to receive(:create_ticket).and_return('id' => 1)
+        register(sender_email: 'support@acme.com').execute.call(
+          FakeActionContext.new(form_values: { 'Requester email' => 'a@b.com',
+                                               'Subject' => 'S', 'Message' => 'M' }),
+          result_builder
+        )
+        expect(client).to have_received(:create_ticket).with(hash_including('recipient' => 'support@acme.com'))
+      end
+
+      it 'omits recipient from the payload when sender_email is blank' do
+        allow(client).to receive(:create_ticket) do |payload|
+          expect(payload).not_to have_key('recipient')
+          { 'id' => 1 }
+        end
+        register.execute.call(
+          FakeActionContext.new(form_values: { 'Requester email' => 'a@b.com',
+                                               'Subject' => 'S', 'Message' => 'M' }),
+          result_builder
+        )
+        expect(client).to have_received(:create_ticket)
+      end
+    end
+  end
+end
diff --git a/packages/forest_admin_datasource_zendesk/spec/spec_helper.rb b/packages/forest_admin_datasource_zendesk/spec/spec_helper.rb
index 561c180a8..4630eecc7 100644
--- a/packages/forest_admin_datasource_zendesk/spec/spec_helper.rb
+++ b/packages/forest_admin_datasource_zendesk/spec/spec_helper.rb
@@ -19,6 +19,7 @@
 SimpleCov.coverage_dir 'coverage'
 
 require 'webmock/rspec'
+require 'forest_admin_datasource_customizer'
 require 'forest_admin_datasource_zendesk'
 
 WebMock.disable_net_connect!(allow_localhost: true)