diff --git a/app/models/notification.rb b/app/models/notification.rb
index fa875e9316..f937c5e415 100644
--- a/app/models/notification.rb
+++ b/app/models/notification.rb
@@ -37,6 +37,9 @@ class Notification < ApplicationRecord
     workshop_log_submitted_fyi
 
     manual_log
+
+    bulk_email
+    bulk_email_fyi
   ].freeze
 
   # Channels for a manually logged communication. "autoemail" is the default
diff --git a/app/models/notification_composition.rb b/app/models/notification_composition.rb
new file mode 100644
index 0000000000..9b5e6a8f5d
--- /dev/null
+++ b/app/models/notification_composition.rb
@@ -0,0 +1,57 @@
+# A saved bulk-email composition: a draft (personal, one-shot) or a reusable
+# template, set by `kind`. Holds the email content plus a re-resolvable audience
+# recipe. On send it fans out into Notification rows (an FYI parent + one child
+# per recipient); the draft is deleted afterwards, so the composition itself is
+# never the record of a sent email — that lives in `notifications`.
+class NotificationComposition < ApplicationRecord
+  KINDS = %w[draft template].freeze
+  SCOPE_TYPES = %w[general event].freeze
+
+  belongs_to :user
+  belongs_to :event, optional: true
+
+  validates :kind, presence: true, inclusion: { in: KINDS }
+  validates :scope_type, presence: true, inclusion: { in: SCOPE_TYPES }
+  # Templates are picked from a list, so they need a name; a draft is a
+  # work-in-progress and may be saved before it has one.
+  validates :name, presence: true, if: :template?
+
+  scope :drafts, -> { where(kind: "draft") }
+  scope :templates, -> { where(kind: "template") }
+
+  def draft?
+    kind == "draft"
+  end
+
+  def template?
+    kind == "template"
+  end
+
+  def event_scoped?
+    scope_type == "event"
+  end
+
+  # Presence of the content is the on/off flag for the optional email blocks —
+  # there are deliberately no _enabled columns.
+  def cta_button?
+    cta_label.present?
+  end
+
+  def grey_box?
+    grey_box_text.present?
+  end
+
+  # Audience recipe + overrides, coalesced so callers always get arrays even
+  # before anything has been stored.
+  def segments
+    recipient_segments || []
+  end
+
+  def added_ids
+    (recipient_added_ids || []).map(&:to_i)
+  end
+
+  def excluded_ids
+    (recipient_excluded_ids || []).map(&:to_i)
+  end
+end
diff --git a/app/services/audience_resolver.rb b/app/services/audience_resolver.rb
new file mode 100644
index 0000000000..1669da4ab1
--- /dev/null
+++ b/app/services/audience_resolver.rb
@@ -0,0 +1,75 @@
+require "set"
+
+# Turns a composition's audience recipe into a concrete set of people.
+#
+# The recipe is an ordered list of { field, value, join } combined left→right,
+# exactly like the recipients builder: OR unions, AND intersects, AND NOT
+# subtracts. Manual add/exclude overrides are then applied on top (add wins).
+# Only emailable people are returned.
+#
+# Field predicates live in FIELDS so new filters are declarative. This starts
+# with the always-safe Person text columns; the fuller registry (county, portal
+# access, role, and the event-scoped registrant fields) is added as each maps to
+# a verified query.
+class AudienceResolver
+  # field name => ->(value) { array of matching Person ids }
+  FIELDS = {
+    "first_name" => ->(v) { AudienceResolver.text_ids(:first_name, v) },
+    "last_name"  => ->(v) { AudienceResolver.text_ids(:last_name, v) },
+    "email"      => ->(v) { AudienceResolver.text_ids(:email, v) }
+  }.freeze
+
+  def self.people_for(composition)
+    new(composition).people
+  end
+
+  # Text search over a Person column, honoring the "a--b" multi-value convention
+  # (match ANY token). Column names come from FIELDS, never user input.
+  def self.text_ids(column, value)
+    tokens = value.to_s.split("--").map(&:strip).reject(&:blank?)
+    return [] if tokens.empty?
+
+    tokens.reduce(Person.where("1 = 0")) { |rel, token|
+      rel.or(Person.where("#{column} LIKE ?", "%#{token}%"))
+    }.pluck(:id)
+  end
+
+  def initialize(composition)
+    @composition = composition
+  end
+
+  def people
+    Person.where(id: resolved_ids.to_a).select { |person| person.preferred_email.present? }
+  end
+
+  # The matched person ids (a Set), before the emailable filter. Exposed for
+  # preview counts.
+  def resolved_ids
+    ids = fold_segments
+    ids.subtract(@composition.excluded_ids)
+    ids.merge(@composition.added_ids)
+    ids
+  end
+
+  private
+
+  def fold_segments
+    active = @composition.segments.select { |seg| FIELDS.key?(seg["field"]) && seg["value"].present? }
+    return Set.new if active.empty?
+
+    result = Set.new(ids_for(active.first))
+    active.drop(1).each do |seg|
+      set = Set.new(ids_for(seg))
+      case seg["join"]
+      when "AND"     then result &= set
+      when "AND NOT" then result -= set
+      else                result |= set # OR (also the base/first segment)
+      end
+    end
+    result
+  end
+
+  def ids_for(segment)
+    FIELDS.fetch(segment["field"]).call(segment["value"])
+  end
+end
diff --git a/app/services/notification_services/send_composition.rb b/app/services/notification_services/send_composition.rb
new file mode 100644
index 0000000000..e0de287526
--- /dev/null
+++ b/app/services/notification_services/send_composition.rb
@@ -0,0 +1,65 @@
+module NotificationServices
+  # Fans a saved composition out into notifications: one FYI parent (the batch
+  # record that shows in the notifications table with a "N recipients" chevron)
+  # plus one child per recipient. Children carry person_id and point back at the
+  # FYI via batch_root_notification_id.
+  #
+  # Delivery is not wired yet: the records (and the batch/history) are created so
+  # the fan-out mechanic is in place. The mailer that renders the styled AWBW
+  # email — and enqueuing it — comes in the next slice.
+  class SendComposition
+    def self.call(composition, recipients:)
+      new(composition, recipients).call
+    end
+
+    def initialize(composition, recipients)
+      @composition = composition
+      @recipients = recipients.select { |person| person.preferred_email.present? }
+    end
+
+    def call
+      Notification.transaction do
+        fyi = create_fyi
+        @recipients.each { |person| create_delivery(person, fyi) }
+        fyi
+      end
+    end
+
+    private
+
+    attr_reader :composition, :recipients
+
+    def create_fyi
+      Notification.create!(
+        base_attributes.merge(
+          kind: "bulk_email_fyi",
+          recipient_role: "admin",
+          recipient_email: composition.user.email,
+          noticeable: composition.event
+        )
+      )
+    end
+
+    def create_delivery(person, fyi)
+      Notification.create!(
+        base_attributes.merge(
+          kind: "bulk_email",
+          recipient_role: "person",
+          recipient_email: person.preferred_email,
+          person_id: person.id,
+          noticeable: person,
+          batch_root_notification_id: fyi.id
+        )
+      )
+    end
+
+    def base_attributes
+      {
+        notification_type: 0,
+        sender_id: composition.user_id,
+        custom_subject: composition.subject,
+        custom_message: composition.body
+      }
+    end
+  end
+end
diff --git a/db/migrate/20260802011450_create_notification_compositions.rb b/db/migrate/20260802011450_create_notification_compositions.rb
new file mode 100644
index 0000000000..a5d3f6872d
--- /dev/null
+++ b/db/migrate/20260802011450_create_notification_compositions.rb
@@ -0,0 +1,33 @@
+class CreateNotificationCompositions < ActiveRecord::Migration[8.1]
+  def change
+    create_table :notification_compositions do |t|
+      # kind = draft | template (see NotificationComposition::KINDS).
+      # A string (not a boolean/STI `type`) so a future "scheduled" kind is additive.
+      t.string :kind, null: false
+      # scope_type = general | event. Independent of event_id: an event-reminder
+      # *template* is event-scoped but bound to no specific event until used.
+      t.string :scope_type, null: false, default: "general"
+      t.string :name
+      t.references :user, null: false, index: true
+      t.references :event, index: true
+
+      # EmailContent — discrete columns. Presence is the on/off flag (no _enabled
+      # columns): a present cta_label shows the button, a present grey_box_text
+      # shows the callout. cta_url null = the recipient's portal profile.
+      t.string :subject
+      t.text :body
+      t.string :cta_label
+      t.string :cta_url
+      t.text :grey_box_text
+
+      # AudienceDefinition — a re-resolvable recipe, never a frozen recipient list.
+      # recipient_segments = ordered [{ field, value, join }]; overrides split by
+      # direction so a single list never has to mean both "force in" and "force out".
+      t.json :recipient_segments
+      t.json :recipient_added_ids
+      t.json :recipient_excluded_ids
+
+      t.timestamps
+    end
+  end
+end
diff --git a/db/migrate/20260802011936_add_bulk_email_fields_to_notifications.rb b/db/migrate/20260802011936_add_bulk_email_fields_to_notifications.rb
new file mode 100644
index 0000000000..6efd366a70
--- /dev/null
+++ b/db/migrate/20260802011936_add_bulk_email_fields_to_notifications.rb
@@ -0,0 +1,13 @@
+class AddBulkEmailFieldsToNotifications < ActiveRecord::Migration[8.1]
+  def change
+    # The recipient as a record (populated when known), so a bulk delivery and a
+    # regular one share one signature — today recipients are email-string-only.
+    add_column :notifications, :person_id, :bigint
+    add_index :notifications, :person_id
+
+    # Batch membership: a bulk child points at its FYI parent. Kept separate from
+    # parent_notification_id / root_notification_id, which mean "resend chain".
+    add_column :notifications, :batch_root_notification_id, :bigint
+    add_index :notifications, :batch_root_notification_id
+  end
+end
diff --git a/db/schema.rb b/db/schema.rb
index da164c312f..bb76275bc8 100644
--- a/db/schema.rb
+++ b/db/schema.rb
@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema[8.1].define(version: 2026_07_29_020406) do
+ActiveRecord::Schema[8.1].define(version: 2026_08_02_011936) do
   create_table "action_text_mentions", charset: "utf8mb4", collation: "utf8mb4_unicode_ci", force: :cascade do |t|
     t.bigint "action_text_rich_text_id", null: false
     t.datetime "created_at", null: false
@@ -754,7 +754,28 @@
     t.index ["organization_user_id"], name: "index_monthly_reports_on_organization_user_id"
   end
 
+  create_table "notification_compositions", charset: "utf8mb4", collation: "utf8mb4_unicode_ci", force: :cascade do |t|
+    t.text "body"
+    t.datetime "created_at", null: false
+    t.string "cta_label"
+    t.string "cta_url"
+    t.bigint "event_id"
+    t.text "grey_box_text"
+    t.string "kind", null: false
+    t.string "name"
+    t.json "recipient_added_ids"
+    t.json "recipient_excluded_ids"
+    t.json "recipient_segments"
+    t.string "scope_type", default: "general", null: false
+    t.string "subject"
+    t.datetime "updated_at", null: false
+    t.bigint "user_id", null: false
+    t.index ["event_id"], name: "index_notification_compositions_on_event_id"
+    t.index ["user_id"], name: "index_notification_compositions_on_user_id"
+  end
+
   create_table "notifications", id: :integer, charset: "utf8mb4", collation: "utf8mb4_unicode_ci", force: :cascade do |t|
+    t.bigint "batch_root_notification_id"
     t.string "channel", default: "autoemail", null: false
     t.datetime "created_at", precision: nil, null: false
     t.text "custom_message"
@@ -771,15 +792,18 @@
     t.string "noticeable_type"
     t.integer "notification_type"
     t.integer "parent_notification_id"
+    t.bigint "person_id"
     t.string "recipient_email", null: false
     t.string "recipient_role", null: false
     t.boolean "responded", default: false, null: false
     t.integer "root_notification_id"
     t.integer "sender_id"
     t.datetime "updated_at", precision: nil, null: false
+    t.index ["batch_root_notification_id"], name: "index_notifications_on_batch_root_notification_id"
     t.index ["kind"], name: "index_notifications_on_kind"
     t.index ["noticeable_type", "noticeable_id"], name: "index_notifications_on_noticeable_type_and_noticeable_id"
     t.index ["parent_notification_id"], name: "index_notifications_on_parent_notification_id"
+    t.index ["person_id"], name: "index_notifications_on_person_id"
     t.index ["root_notification_id"], name: "index_notifications_on_root_notification_id"
     t.index ["sender_id"], name: "index_notifications_on_sender_id"
   end
diff --git a/spec/factories/notification_compositions.rb b/spec/factories/notification_compositions.rb
new file mode 100644
index 0000000000..fe799b9e55
--- /dev/null
+++ b/spec/factories/notification_compositions.rb
@@ -0,0 +1,21 @@
+FactoryBot.define do
+  factory :notification_composition do
+    association :user
+    kind { "draft" }
+    scope_type { "general" }
+    subject { "A note from Art With A Woman" }
+    body { "Hi {{first_name}}," }
+    cta_label { "View your portal profile" }
+    recipient_segments { [ { "field" => "county", "value" => "LA", "join" => "AND" } ] }
+
+    trait :template do
+      kind { "template" }
+      name { "Monthly newsletter" }
+    end
+
+    trait :event_scoped do
+      scope_type { "event" }
+      association :event
+    end
+  end
+end
diff --git a/spec/models/notification_composition_spec.rb b/spec/models/notification_composition_spec.rb
new file mode 100644
index 0000000000..481f41c5bb
--- /dev/null
+++ b/spec/models/notification_composition_spec.rb
@@ -0,0 +1,60 @@
+require "rails_helper"
+
+RSpec.describe NotificationComposition, type: :model do
+  it "has a valid factory" do
+    expect(build(:notification_composition)).to be_valid
+  end
+
+  describe "validations" do
+    it "requires a known kind" do
+      expect(build(:notification_composition, kind: "bogus")).not_to be_valid
+      expect(build(:notification_composition, kind: nil)).not_to be_valid
+    end
+
+    it "requires a known scope_type" do
+      expect(build(:notification_composition, scope_type: "bogus")).not_to be_valid
+    end
+
+    it "requires a name for templates but not drafts" do
+      expect(build(:notification_composition, :template, name: nil)).not_to be_valid
+      expect(build(:notification_composition, kind: "draft", name: nil)).to be_valid
+    end
+  end
+
+  describe "scopes" do
+    it "separates drafts from templates" do
+      draft = create(:notification_composition, kind: "draft")
+      template = create(:notification_composition, :template)
+
+      expect(described_class.drafts).to include(draft)
+      expect(described_class.drafts).not_to include(template)
+      expect(described_class.templates).to include(template)
+      expect(described_class.templates).not_to include(draft)
+    end
+  end
+
+  describe "content block flags (presence is the flag)" do
+    it "shows the CTA button only when a label is present" do
+      expect(build(:notification_composition, cta_label: "Go").cta_button?).to be(true)
+      expect(build(:notification_composition, cta_label: nil).cta_button?).to be(false)
+    end
+
+    it "shows the grey callout only when text is present" do
+      expect(build(:notification_composition, grey_box_text: "Note").grey_box?).to be(true)
+      expect(build(:notification_composition, grey_box_text: nil).grey_box?).to be(false)
+    end
+  end
+
+  describe "audience recipe accessors" do
+    it "coalesce nil to arrays and cast override ids to integers" do
+      comp = build(:notification_composition,
+                   recipient_segments: nil,
+                   recipient_added_ids: [ "1", "2" ],
+                   recipient_excluded_ids: nil)
+
+      expect(comp.segments).to eq([])
+      expect(comp.added_ids).to eq([ 1, 2 ])
+      expect(comp.excluded_ids).to eq([])
+    end
+  end
+end
diff --git a/spec/services/audience_resolver_spec.rb b/spec/services/audience_resolver_spec.rb
new file mode 100644
index 0000000000..0b0645b65f
--- /dev/null
+++ b/spec/services/audience_resolver_spec.rb
@@ -0,0 +1,61 @@
+require "rails_helper"
+
+RSpec.describe AudienceResolver do
+  let!(:amy)   { create(:person, first_name: "Amy",   last_name: "User") }
+  let!(:aisha) { create(:person, first_name: "Aisha", last_name: "Sharma") }
+  let!(:bob)   { create(:person, first_name: "Bob",   last_name: "Jones") }
+
+  def resolve(segments, added: [], excluded: [])
+    comp = build(:notification_composition,
+                 recipient_segments: segments,
+                 recipient_added_ids: added,
+                 recipient_excluded_ids: excluded)
+    described_class.new(comp).people
+  end
+
+  it "unions matches with OR" do
+    people = resolve([
+      { "field" => "first_name", "value" => "Amy",   "join" => "AND" },
+      { "field" => "first_name", "value" => "Aisha", "join" => "OR" }
+    ])
+    expect(people).to match_array([ amy, aisha ])
+  end
+
+  it "intersects matches with AND" do
+    people = resolve([
+      { "field" => "last_name",  "value" => "Sharma", "join" => "AND" },
+      { "field" => "first_name", "value" => "Aisha",  "join" => "AND" }
+    ])
+    expect(people).to eq([ aisha ])
+  end
+
+  it "subtracts matches with AND NOT" do
+    people = resolve([
+      { "field" => "first_name", "value" => "a",     "join" => "AND" },     # Amy + Aisha
+      { "field" => "first_name", "value" => "Aisha", "join" => "AND NOT" }
+    ])
+    expect(people).to eq([ amy ])
+  end
+
+  it "honors the a--b multi-value convention on text fields" do
+    people = resolve([ { "field" => "first_name", "value" => "Amy--Bob", "join" => "AND" } ])
+    expect(people).to match_array([ amy, bob ])
+  end
+
+  it "applies manual add and exclude overrides (add wins)" do
+    added = resolve([ { "field" => "first_name", "value" => "Amy", "join" => "AND" } ], added: [ bob.id ])
+    expect(added).to match_array([ amy, bob ])
+
+    excluded = resolve([ { "field" => "first_name", "value" => "Amy--Bob", "join" => "AND" } ], excluded: [ bob.id ])
+    expect(excluded).to eq([ amy ])
+  end
+
+  it "drops matched people who have no email" do
+    create(:person, first_name: "Zed", user: nil, email: nil, email_2: nil)
+    expect(resolve([ { "field" => "first_name", "value" => "Zed", "join" => "AND" } ])).to be_empty
+  end
+
+  it "returns nothing when no segment matches a known field" do
+    expect(resolve([ { "field" => "unknown_field", "value" => "x", "join" => "AND" } ])).to be_empty
+  end
+end
diff --git a/spec/services/notification_services/send_composition_spec.rb b/spec/services/notification_services/send_composition_spec.rb
new file mode 100644
index 0000000000..7ebc075f52
--- /dev/null
+++ b/spec/services/notification_services/send_composition_spec.rb
@@ -0,0 +1,38 @@
+require "rails_helper"
+
+RSpec.describe NotificationServices::SendComposition do
+  let(:owner) { create(:user, email: "admin@example.org") }
+  let(:composition) do
+    create(:notification_composition, user: owner, subject: "Spring news", body: "Hello there")
+  end
+  let(:recipients) { create_list(:person, 3) }
+
+  it "creates one FYI parent plus one child per recipient" do
+    fyi = described_class.call(composition, recipients: recipients)
+
+    expect(fyi.kind).to eq("bulk_email_fyi")
+    expect(fyi.recipient_role).to eq("admin")
+    expect(fyi.recipient_email).to eq("admin@example.org")
+
+    children = Notification.where(batch_root_notification_id: fyi.id)
+    expect(children.count).to eq(3)
+    expect(children.pluck(:kind).uniq).to eq([ "bulk_email" ])
+    expect(children.pluck(:person_id)).to match_array(recipients.map(&:id))
+  end
+
+  it "copies the composed subject and body onto the FYI and every child" do
+    fyi = described_class.call(composition, recipients: recipients)
+    batch = Notification.where(batch_root_notification_id: fyi.id).to_a + [ fyi ]
+
+    expect(batch.map(&:custom_subject).uniq).to eq([ "Spring news" ])
+    expect(batch.map(&:custom_message).uniq).to eq([ "Hello there" ])
+  end
+
+  it "skips recipients without an email" do
+    no_email = create(:person)
+    allow(no_email).to receive(:preferred_email).and_return(nil)
+
+    fyi = described_class.call(composition, recipients: recipients + [ no_email ])
+    expect(Notification.where(batch_root_notification_id: fyi.id).count).to eq(3)
+  end
+end