diff --git a/app/models/billing_segment.rb b/app/models/billing_segment.rb
index a114c2bfb91..04e98aa99c1 100644
--- a/app/models/billing_segment.rb
+++ b/app/models/billing_segment.rb
@@ -45,6 +45,10 @@ def rate
     rate_override || rate_card_rate
   end
 
+  def target_key
+    "contract-#{contract_id}-#{contract_rate_card.product.target_key}"
+  end
+
   # The shared matcher expects a filter object for the default bucket, not nil.
   # This mirrors the empty ChargeFilter in app/services/fees/charge_service.rb:85-95.
   def empty_product_filter
diff --git a/app/services/billing_segments/process_service.rb b/app/services/billing_segments/process_service.rb
index 26bc96e0d87..02a2e128c20 100644
--- a/app/services/billing_segments/process_service.rb
+++ b/app/services/billing_segments/process_service.rb
@@ -36,8 +36,13 @@ def call
     def pending_segments
       BillingSegment.status_pending
         .where(customer_id: customer.id)
+        .joins(contract_rate_card: {rate_card: :product})
+        .where(
+          "products.product_type != :metered OR rate_cards.billing_timing != :advance",
+          metered: Product::PRODUCT_TYPES[:metered],
+          advance: RateCard::BILLING_TIMINGS[:advance]
+        )
         .includes(:pricing_unit, :rate_override, :contract, contract_rate_card: {rate_card: :product}, rate_card_rate: :rate_card)
-        .select { |segment| segment.contract_rate_card.product.fixed? }
     end
 
     def invoice_key(segment)
@@ -67,6 +72,9 @@ def payment_method_key(contract)
     def build_invoice(segments)
       contract = segments.first.contract
       invoice = nil
+      grouped = segments.group_by { |s| s.contract_rate_card.product.product_type }
+      metered_segments = grouped[Product::PRODUCT_TYPES[:metered]] || []
+      fixed_segments = grouped[Product::PRODUCT_TYPES[:fixed]] || []
 
       ActiveRecord::Base.transaction do
         invoice = Invoices::CreateGeneratingService.call!(
@@ -79,17 +87,11 @@ def build_invoice(segments)
           purchase_order_number: contract.purchase_order_number
         ).invoice
 
-        segments.flat_map do |segment|
-          if segment.contract_rate_card.product.fixed?
-            computed_fees(segment)
-          else
-            []
-          end
-        end.each do |fee|
-          fee.invoice = invoice
-          fee.billing_entity = invoice.billing_entity
-          fee.save!
-        end
+        filtered_aggregations = event_filters(metered_segments)
+
+        attach_fixed_fees(fixed_segments, invoice)
+        attach_metered_fees(metered_segments, invoice, filtered_aggregations)
+
         invoice.fees.reload
 
         Invoices::ComputeAmountsFromFees.call!(invoice:)
@@ -100,11 +102,47 @@ def build_invoice(segments)
       invoice
     end
 
-    def computed_fees(segment)
+    def attach_fixed_fees(segments, invoice)
+      segments.each do |segment|
+        compute_fixed_fees(segment).each do |fee|
+          fee.invoice = invoice
+          fee.billing_entity = invoice.billing_entity
+          fee.save!
+        end
+      end
+    end
+
+    def attach_metered_fees(segments, invoice, filtered_aggregations)
+      segments.each do |segment|
+        compute_metered_fees(segment, invoice, filtered_aggregations)
+      end
+    end
+
+    def compute_fixed_fees(segment)
       fee_result = BillingSegments::Fees::ComputeService.call!(billing_segment: segment)
       [fee_result.fee, fee_result.true_up_fee].compact
     end
 
+    # NOTE: Fees::ChargeService persists and attaches the product fees itself (within this
+    # surrounding transaction), so a failure on any segment rolls back the whole invoice group.
+    def compute_metered_fees(segment, invoice, filtered_aggregations)
+      ::Fees::ChargeService.call!(
+        invoice:,
+        metered_item: ::Fees::ChargeService::MeteredItem.from_billing_segment(segment),
+        billing_context: Billing::Context.from(contract: segment.contract),
+        options: ::Fees::ChargeService::Options.new(context: :finalize, skip_adjusted_fees: true),
+        filtered_aggregations: filtered_aggregations[segment.target_key]&.keys || []
+      )
+    end
+
+    def event_filters(metered_segments)
+      return {} if metered_segments.empty?
+
+      Events::BillingPeriodFilterService.for_billing_segments!(
+        billing_segments: metered_segments, with_last_seen_at: false
+      ).filter_targets
+    end
+
     def finalize_generating_invoices
       # Re-query done segments so a retry finalizes existing invoices without rebuilding fees.
       invoice_ids = BillingSegment.status_done
diff --git a/app/services/events/billing_period_filter_service.rb b/app/services/events/billing_period_filter_service.rb
index 6829e523890..58400990f62 100644
--- a/app/services/events/billing_period_filter_service.rb
+++ b/app/services/events/billing_period_filter_service.rb
@@ -15,10 +15,9 @@ def self.for_charges!(subscription:, boundaries:, codes: nil, with_last_seen_at:
       )
     end
 
-    def self.for_billing_segments!(contract:, billing_segments:, codes: nil, with_last_seen_at: true)
+    def self.for_billing_segments!(billing_segments:, codes: nil, with_last_seen_at: true)
       call!(
         resolver: BillingPeriodFilters::BillingSegmentsResolver.new(
-          contract:,
           billing_segments:,
           codes:,
           with_last_seen_at:
diff --git a/app/services/events/billing_period_filters/base_resolver.rb b/app/services/events/billing_period_filters/base_resolver.rb
index f440c375543..da7178d4117 100644
--- a/app/services/events/billing_period_filters/base_resolver.rb
+++ b/app/services/events/billing_period_filters/base_resolver.rb
@@ -30,7 +30,7 @@ def event_values_with_history
       end
 
       def non_recurring_metric_codes
-        @non_recurring_metric_codes ||= metric_codes.to_a - recurring_metric_codes
+        @non_recurring_metric_codes ||= metric_codes(record_id: nil).to_a - recurring_metric_codes
       end
 
       def recurring_event_filter_targets
diff --git a/app/services/events/billing_period_filters/billing_segments_resolver.rb b/app/services/events/billing_period_filters/billing_segments_resolver.rb
index 6e4ac3c5a76..e3a9c5e5c63 100644
--- a/app/services/events/billing_period_filters/billing_segments_resolver.rb
+++ b/app/services/events/billing_period_filters/billing_segments_resolver.rb
@@ -3,8 +3,7 @@
 module Events
   module BillingPeriodFilters
     class BillingSegmentsResolver < BaseResolver
-      def initialize(contract:, billing_segments:, codes: nil, with_last_seen_at: true)
-        @contract = contract
+      def initialize(billing_segments:, codes: nil, with_last_seen_at: true)
         @billing_segments = billing_segments
         @codes = codes&.to_set || Set.new
         @with_last_seen_at = with_last_seen_at
@@ -12,32 +11,58 @@ def initialize(contract:, billing_segments:, codes: nil, with_last_seen_at: true
 
       def filter_targets
         return {} if target_segments.empty?
-        return {} if metric_codes.empty?
+        return {} if metric_codes_by_contract_id.empty?
+
+        # Event stores are scoped to a contract's external_id, so combinations must only
+        # be matched against targets from that contract, even when products share a code.
+        # For example, Contract A and Contract B can both use product "api_calls", while
+        # only A has a {region: "eu"} event. Matching A's combinations against B would
+        # incorrectly add A's filter bucket and last_seen_at to B's target.
+        result = recurring_event_filter_targets
+
+        contracts.each do |contract|
+          combinations = event_values_with_history_for(contract_id: contract.id) do |**options|
+            event_store_for(contract).distinct_codes_and_property_combinations(
+              filter_keys: billable_metric_filter_keys_by_contract.fetch(contract.id, []),
+              **options
+            )
+          end
+
+          event_codes = combinations.map(&:first).to_set
+          contract_targets = segments_by_contract.fetch(contract.id, []).select do |target|
+            event_codes.include?(filter_target_for(target).billable_metric.code)
+          end
+
+          filter_targets_from_combinations(
+            combinations:,
+            targets: contract_targets,
+            result:
+          )
+        end
 
-        super
+        result
       end
 
       private
 
-      attr_reader :contract, :billing_segments, :codes, :with_last_seen_at
+      attr_reader :billing_segments, :codes, :with_last_seen_at
 
-      delegate :organization, to: :contract
+      def organization
+        @organization ||= target_segments.first.organization
+      end
 
       def filter_target_for(billing_segment)
-        Events::BillingPeriodFilters::FilterTarget.from_billing_segment(billing_segment:)
+        @filter_targets ||= {}
+        @filter_targets[billing_segment.id] ||= Events::BillingPeriodFilters::FilterTarget.from_billing_segment(billing_segment:)
       end
 
       def target_segments
         @target_segments ||= billing_segments_scope.preload(
+          :contract,
           contract_rate_card: {product: [:billable_metric, {filters: {values: :billable_metric_filter}}]}
         ).to_a
       end
 
-      def targets_with_events(codes)
-        event_codes = codes.to_set
-        target_segments.select { |segment| event_codes.include?(filter_target_for(segment).billable_metric.code) }
-      end
-
       def billing_segments_scope
         scope = BillingSegment.where(id: billing_segments)
           .joins(contract_rate_card: {product: :billable_metric})
@@ -49,15 +74,38 @@ def billing_segments_scope
         end
       end
 
-      def metric_codes
-        @metric_codes ||= codes.presence || billing_segments_scope.distinct.pluck("billable_metrics.code")
+      def event_values_with_history_for(contract_id:)
+        recurring_codes = recurring_metric_codes_by_contract_id.fetch(contract_id, [])
+        non_recurring_metric_codes = metric_codes_for(contract_id:) - recurring_codes
+
+        values = yield(codes: non_recurring_metric_codes, with_last_seen_at:)
+
+        if recurring_codes.any?
+          values += yield(codes: recurring_codes, include_all_history: true, with_last_seen_at:)
+        end
+
+        values
+      end
+
+      def metric_codes_for(contract_id:)
+        @metric_codes_by_contract ||= {}
+        @metric_codes_by_contract[contract_id] ||= codes.presence || metric_codes_by_contract_id.fetch(contract_id, [])
       end
 
-      def recurring_metric_codes
-        @recurring_metric_codes ||= billing_segments_scope
+      def metric_codes_by_contract_id
+        @metric_codes_by_contract_id ||= billing_segments_scope
+          .distinct
+          .pluck("contract_rate_cards.contract_id, billable_metrics.code")
+          .group_by(&:first).transform_values! { |pairs| pairs.map(&:last) }
+      end
+
+      def recurring_metric_codes_by_contract_id
+        @recurring_metric_codes_by_contract_id ||= billing_segments_scope
           .where(billable_metrics: {recurring: true})
           .distinct
-          .pluck("billable_metrics.code")
+          .pluck("contract_rate_cards.contract_id, billable_metrics.code")
+          .group_by(&:first)
+          .transform_values! { |pairs| pairs.map(&:last) }
       end
 
       def current_recurring_targets
@@ -68,22 +116,36 @@ def period_start
         @period_start ||= target_segments.map(&:started_at).min
       end
 
-      def billable_metric_filter_keys
-        @billable_metric_filter_keys ||= billing_segments_scope
+      def billable_metric_filter_keys_by_contract
+        @billable_metric_filter_keys_by_contract ||= billing_segments_scope
           .joins(contract_rate_card: {product: {billable_metric: :filters}})
           .distinct
-          .pluck("billable_metric_filters.key")
-      end
-
-      def event_store
-        @event_store ||= Events::Stores::StoreFactory.new_instance(
-          organization:,
-          billing_context: Billing::Context.from(contract:),
-          boundaries: {
-            from_datetime: period_start,
-            to_datetime: target_segments.map(&:ended_at).max
-          }
-        )
+          .pluck("contract_rate_cards.contract_id", "billable_metric_filters.key")
+          .group_by(&:first)
+          .transform_values! { |pairs| pairs.map(&:last) }
+      end
+
+      def event_store_for(contract)
+        @event_stores ||= {}
+        @event_stores[contract.id] ||= begin
+          contract_segments = segments_by_contract[contract.id]
+          Events::Stores::StoreFactory.new_instance(
+            organization:,
+            billing_context: Billing::Context.from(contract:),
+            boundaries: {
+              from_datetime: contract_segments.map(&:started_at).min,
+              to_datetime: contract_segments.map(&:ended_at).max
+            }
+          )
+        end
+      end
+
+      def segments_by_contract
+        @segments_by_contract ||= target_segments.group_by(&:contract_id)
+      end
+
+      def contracts
+        @contracts ||= target_segments.map(&:contract).uniq
       end
     end
   end
diff --git a/app/services/events/billing_period_filters/charges_resolver.rb b/app/services/events/billing_period_filters/charges_resolver.rb
index dd88b0f9616..531f4b59036 100644
--- a/app/services/events/billing_period_filters/charges_resolver.rb
+++ b/app/services/events/billing_period_filters/charges_resolver.rb
@@ -33,12 +33,13 @@ def event_store
 
       # A code outside of the plan matches no event, so codes is used as is: dropping it would leave
       # its charge out of the result, billed as zero units instead of surfaced.
-      def metric_codes
+      def metric_codes(record_id: nil)
         @metric_codes ||= codes || plan.billable_metrics.distinct.pluck(:code)
       end
 
       def filter_target_for(charge)
-        Events::BillingPeriodFilters::FilterTarget.from_charge(charge:)
+        @filter_target_for ||= {}
+        @filter_target_for[charge.id] ||= Events::BillingPeriodFilters::FilterTarget.from_charge(charge:)
       end
 
       def targets_with_events(codes)
diff --git a/app/services/events/billing_period_filters/sources/billing_segment.rb b/app/services/events/billing_period_filters/sources/billing_segment.rb
index 2d45920a9c1..d14616f314a 100644
--- a/app/services/events/billing_period_filters/sources/billing_segment.rb
+++ b/app/services/events/billing_period_filters/sources/billing_segment.rb
@@ -32,7 +32,7 @@ def all_filter_values?(filter, key)
           filter.to_h[key] == [nil]
         end
 
-        delegate :target_key, to: :product
+        delegate :target_key, to: :billing_segment
 
         def with_filter(filter)
           self.class.new(billing_segment:, filter:)
diff --git a/spec/models/billing_segment_spec.rb b/spec/models/billing_segment_spec.rb
index e03d36b18cf..58571f68ddc 100644
--- a/spec/models/billing_segment_spec.rb
+++ b/spec/models/billing_segment_spec.rb
@@ -110,6 +110,32 @@
     end
   end
 
+  describe "#target_key" do
+    let(:organization) { create(:organization) }
+    let(:customer) { create(:customer, organization:) }
+    let(:contract) { create(:contract, organization:, customer:) }
+    let(:product) { create(:product, organization:) }
+    let(:rate_card) { create(:rate_card, organization:, product:) }
+    let(:contract_rate_card) { create(:contract_rate_card, organization:, contract:, rate_card:) }
+    let(:rate_card_rate) { create(:rate_card_rate, organization:, rate_card:) }
+    let(:billing_segment) do
+      create(
+        :billing_segment,
+        organization:,
+        customer:,
+        contract:,
+        contract_rate_card:,
+        rate_card_rate:
+      )
+    end
+
+    it "includes the contract and product identities" do
+      expect(billing_segment.target_key).to eq(
+        "contract-#{billing_segment.contract_id}-#{billing_segment.contract_rate_card.product.target_key}"
+      )
+    end
+  end
+
   describe "#duration_in_days" do
     subject(:segment) { described_class.new(customer:, started_at:, ended_at:) }
 
diff --git a/spec/services/billing_segments/process_service_spec.rb b/spec/services/billing_segments/process_service_spec.rb
index 28a291179ed..8ed72954482 100644
--- a/spec/services/billing_segments/process_service_spec.rb
+++ b/spec/services/billing_segments/process_service_spec.rb
@@ -28,8 +28,7 @@
     let(:billing_segment_pricing_unit) { nil }
     let(:billing_segment_proration_ratio) { 1 }
     let(:min_amount_cents) { 0 }
-
-    let!(:billing_segment) do
+    let(:billing_segment) do
       create(
         :billing_segment,
         organization:,
@@ -49,6 +48,10 @@
       )
     end
 
+    before do
+      billing_segment
+    end
+
     describe "#pending_segments" do
       it "loads the shared rate card once for both association paths" do
         queries = []
@@ -68,15 +71,411 @@
       end
     end
 
-    context "with only usage (metered) products" do
-      let(:product) { create(:product, organization:) }
+    context "with metered products" do
+      let(:billable_metric) { create(:billable_metric, organization:, aggregation_type:, field_name:) }
+      let(:product) { create(:product, :metered, organization:, billable_metric:) }
+      let(:rate_override) { nil }
+      let(:billing_segment_rate_properties) { rate_properties }
 
-      it "leaves segments pending without creating invoices or fees" do
-        expect { result }.to not_change(Invoice, :count).and not_change(Fee, :count)
+      before do
+        create(:event, organization:, customer:, external_subscription_id: contract.external_id,
+          code: billable_metric.code, timestamp: Time.zone.parse("2026-08-10"), properties: event_properties)
+        create(:event, organization:, customer:, external_subscription_id: contract.external_id,
+          code: billable_metric.code, timestamp: Time.zone.parse("2026-08-20"), properties: event_properties)
+      end
 
-        expect(result).to be_success
-        expect(result.invoices).to eq([])
-        expect(billing_segment.reload).to have_attributes(status: "pending", invoice_id: nil)
+      context "with an advance card" do
+        let(:aggregation_type) { :count_agg }
+        let(:field_name) { nil }
+        let(:event_properties) { {} }
+        let(:rate_card) { create(:rate_card, organization:, product:, currency: "USD", billing_timing: :advance) }
+
+        it "does not process the segment through periodic billing" do
+          expect { result }.not_to change(Invoice, :count)
+
+          expect(result).to be_success
+          expect(result.invoices).to be_empty
+          expect(billing_segment.reload).to have_attributes(status: "pending", invoice_id: nil)
+        end
+      end
+
+      context "with count aggregation" do
+        let(:aggregation_type) { :count_agg }
+        let(:field_name) { nil }
+        let(:event_properties) { {} }
+        let(:rate_properties) { {"amount" => "15.00"} }
+
+        it "creates an invoice with a metered fee based on event count" do
+          expect(result).to be_success
+
+          invoice = result.invoices.sole.reload
+          expect(invoice.status).to eq("finalized")
+
+          fee = invoice.fees.sole
+          expect(fee).to have_attributes(
+            invoiceable: product,
+            fee_type: "product",
+            units: 2,
+            events_count: 2,
+            amount_cents: 3_000,
+            precise_unit_amount: BigDecimal("15.00")
+          )
+          expect(billing_segment.reload).to have_attributes(status: "done", invoice:)
+        end
+
+        context "with a minimum amount" do
+          let(:min_amount_cents) { 10_000 }
+
+          it "creates a true-up fee when usage is below minimum" do
+            expect(result).to be_success
+
+            invoice = result.invoices.sole
+            fee, true_up_fee = invoice.fees.order(:created_at)
+
+            expect(invoice.total_amount_cents).to eq(10_000)
+            expect(fee.amount_cents).to eq(3_000)
+            expect(true_up_fee).to have_attributes(amount_cents: 7_000, true_up_parent_fee_id: fee.id)
+          end
+        end
+      end
+
+      context "with sum aggregation" do
+        let(:aggregation_type) { :sum_agg }
+        let(:field_name) { "quantity" }
+        let(:event_properties) { {"quantity" => 10} }
+        let(:rate_properties) { {"amount" => "2.50"} }
+
+        it "creates an invoice with a metered fee based on summed field values" do
+          expect(result).to be_success
+
+          invoice = result.invoices.sole.reload
+          expect(invoice.status).to eq("finalized")
+
+          fee = invoice.fees.sole
+          expect(fee).to have_attributes(
+            invoiceable: product,
+            fee_type: "product",
+            units: 20,
+            events_count: 2,
+            amount_cents: 5_000,
+            precise_unit_amount: BigDecimal("2.50")
+          )
+          expect(billing_segment.reload).to have_attributes(status: "done", invoice:)
+        end
+
+        context "with graduated pricing" do
+          let(:rate_model) { "graduated" }
+          let(:rate_properties) do
+            {"graduated_ranges" => [
+              {"from_value" => 0, "to_value" => 10, "per_unit_amount" => "5.00", "flat_amount" => "0.00"},
+              {"from_value" => 11, "to_value" => nil, "per_unit_amount" => "3.00", "flat_amount" => "0.00"}
+            ]}
+          end
+
+          it "applies graduated tiers to the summed usage" do
+            expect(result).to be_success
+
+            invoice = result.invoices.sole
+            fee = invoice.fees.sole
+
+            # First 10 units at $5, next 10 units at $3 = $50 + $30 = $80
+            expect(fee).to have_attributes(units: 20, amount_cents: 8_000)
+            expect(fee.amount_details["graduated_ranges"].size).to eq(2)
+          end
+        end
+      end
+
+      context "with mixed fixed and metered segments" do
+        let(:aggregation_type) { :count_agg }
+        let(:field_name) { nil }
+        let(:event_properties) { {} }
+        let(:rate_properties) { {"amount" => "10.00"} }
+
+        let(:fixed_product) { create(:product, :fixed, organization:) }
+        let(:fixed_rate_card) { create(:rate_card, organization:, product: fixed_product, currency: "USD") }
+        let(:fixed_contract_rate_card) do
+          create(:contract_rate_card, organization:, contract:, rate_card: fixed_rate_card, units: 3, effective_date: Date.parse("2026-07-01"))
+        end
+        let(:fixed_rate_card_rate) do
+          create(:rate_card_rate, organization:, rate_card: fixed_rate_card, rate_properties: {"amount" => "20.00"})
+        end
+        let(:fixed_segment) do
+          create(
+            :billing_segment, organization:, contract:, customer:,
+            contract_rate_card: fixed_contract_rate_card, rate_card_rate: fixed_rate_card_rate,
+            currency: "USD", rate_properties: {"amount" => "20.00"},
+            billing_at: Time.zone.parse("2026-08-31 23:59:59"), cycle_started_at: Time.zone.parse("2026-08-01"),
+            started_at: Time.zone.parse("2026-08-01"), ended_at: Time.zone.parse("2026-08-31 23:59:59")
+          )
+        end
+
+        before do
+          fixed_segment
+        end
+
+        it "creates an invoice with both fixed and metered fees" do
+          expect(result).to be_success
+
+          invoice = result.invoices.sole.reload
+          expect(invoice.status).to eq("finalized")
+          expect(invoice.fees.count).to eq(2)
+
+          metered_fee = invoice.fees.find { |f| f.invoiceable == product }
+          fixed_fee = invoice.fees.find { |f| f.invoiceable == fixed_product }
+
+          expect(metered_fee).to have_attributes(fee_type: "product", units: 2, amount_cents: 2_000)
+          expect(fixed_fee).to have_attributes(fee_type: "product", units: 3, amount_cents: 6_000)
+
+          # 2 events × $10 + 3 units × $20 = $20 + $60 = $80
+          expect(invoice.total_amount_cents).to eq(8_000)
+
+          expect(billing_segment.reload).to have_attributes(status: "done", invoice:)
+          expect(fixed_segment.reload).to have_attributes(status: "done", invoice:)
+        end
+      end
+
+      context "with two consolidated metered contracts" do
+        let(:aggregation_type) { :count_agg }
+        let(:field_name) { nil }
+        let(:event_properties) { {} }
+        let(:rate_properties) { {"amount" => "5.00"} }
+
+        let(:second_contract) do
+          create(:contract, organization:, customer:, external_id: "second_contract_ext_id", consolidate_invoice: true)
+        end
+        let(:second_billable_metric) { create(:billable_metric, organization:, aggregation_type: :count_agg) }
+        let(:second_product) { create(:product, :metered, organization:, billable_metric: second_billable_metric) }
+        let(:second_rate_card) { create(:rate_card, organization:, product: second_product, currency: "USD") }
+        let(:second_contract_rate_card) do
+          create(:contract_rate_card, organization:, contract: second_contract, rate_card: second_rate_card, effective_date: Date.parse("2026-07-01"))
+        end
+        let(:second_rate_card_rate) do
+          create(:rate_card_rate, organization:, rate_card: second_rate_card, rate_properties: {"amount" => "7.00"})
+        end
+        let(:second_segment) do
+          create(
+            :billing_segment, organization:, contract: second_contract, customer:,
+            contract_rate_card: second_contract_rate_card, rate_card_rate: second_rate_card_rate,
+            currency: "USD", rate_properties: {"amount" => "7.00"},
+            billing_at: Time.zone.parse("2026-08-31 23:59:59"), cycle_started_at: Time.zone.parse("2026-08-01"),
+            started_at: Time.zone.parse("2026-08-01"), ended_at: Time.zone.parse("2026-08-31 23:59:59")
+          )
+        end
+
+        before do
+          second_segment
+          # Events for the second contract (different external_subscription_id)
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-12"), properties: {})
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-18"), properties: {})
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-25"), properties: {})
+        end
+
+        it "creates a consolidated invoice with fees from both contracts using each contract's events" do
+          expect(result).to be_success
+
+          invoice = result.invoices.sole.reload
+          expect(invoice.status).to eq("finalized")
+          expect(invoice.fees.count).to eq(2)
+
+          first_fee = invoice.fees.find { |f| f.invoiceable == product }
+          second_fee = invoice.fees.find { |f| f.invoiceable == second_product }
+
+          # First contract: 2 events × $5 = $10
+          expect(first_fee).to have_attributes(fee_type: "product", units: 2, events_count: 2, amount_cents: 1_000)
+          # Second contract: 3 events × $7 = $21
+          expect(second_fee).to have_attributes(fee_type: "product", units: 3, events_count: 3, amount_cents: 2_100)
+
+          expect(invoice.total_amount_cents).to eq(3_100)
+
+          expect(billing_segment.reload).to have_attributes(status: "done", invoice:)
+          expect(second_segment.reload).to have_attributes(status: "done", invoice:)
+        end
+      end
+
+      context "with consolidated contracts having different billing period date ranges" do
+        let(:aggregation_type) { :count_agg }
+        let(:field_name) { nil }
+        let(:event_properties) { {} }
+        let(:rate_properties) { {"amount" => "10.00"} }
+
+        # First contract: billing period Aug 15-31 (second half of August)
+        let(:billing_segment) do
+          create(
+            :billing_segment, organization:, contract:, customer:,
+            contract_rate_card:, rate_card_rate:,
+            currency: "USD", rate_properties:,
+            billing_at: Time.zone.parse("2026-08-31 23:59:59"), cycle_started_at: Time.zone.parse("2026-08-15"),
+            started_at: Time.zone.parse("2026-08-15"), ended_at: Time.zone.parse("2026-08-31 23:59:59")
+          )
+        end
+
+        let(:second_contract) do
+          create(:contract, organization:, customer:, external_id: "second_contract_ext_id", consolidate_invoice: true)
+        end
+        let(:second_billable_metric) { create(:billable_metric, organization:, aggregation_type: :count_agg) }
+        let(:second_product) { create(:product, :metered, organization:, billable_metric: second_billable_metric) }
+        let(:second_rate_card) { create(:rate_card, organization:, product: second_product, currency: "USD") }
+        let(:second_contract_rate_card) do
+          create(:contract_rate_card, organization:, contract: second_contract, rate_card: second_rate_card, effective_date: Date.parse("2026-07-01"))
+        end
+        let(:second_rate_card_rate) do
+          create(:rate_card_rate, organization:, rate_card: second_rate_card, rate_properties: {"amount" => "20.00"})
+        end
+        # Second contract: billing period Aug 1-14 (first half of August)
+        let(:second_segment) do
+          create(
+            :billing_segment, organization:, contract: second_contract, customer:,
+            contract_rate_card: second_contract_rate_card, rate_card_rate: second_rate_card_rate,
+            currency: "USD", rate_properties: {"amount" => "20.00"},
+            billing_at: Time.zone.parse("2026-08-31 23:59:59"), cycle_started_at: Time.zone.parse("2026-08-01"),
+            started_at: Time.zone.parse("2026-08-01"), ended_at: Time.zone.parse("2026-08-14 23:59:59")
+          )
+        end
+
+        before do
+          Event.delete_all # clean previous events
+
+          billing_segment
+          second_segment
+
+          # Events for first contract in its billing period (Aug 15-31) - should be counted
+          create(:event, organization:, customer:, external_subscription_id: contract.external_id,
+            code: billable_metric.code, timestamp: Time.zone.parse("2026-08-20"), properties: {})
+          create(:event, organization:, customer:, external_subscription_id: contract.external_id,
+            code: billable_metric.code, timestamp: Time.zone.parse("2026-08-25"), properties: {})
+
+          # Events for first contract OUTSIDE its billing period (Aug 1-14) - should NOT be counted
+          create(:event, organization:, customer:, external_subscription_id: contract.external_id,
+            code: billable_metric.code, timestamp: Time.zone.parse("2026-08-05"), properties: {})
+          create(:event, organization:, customer:, external_subscription_id: contract.external_id,
+            code: billable_metric.code, timestamp: Time.zone.parse("2026-08-10"), properties: {})
+
+          # Events for second contract in its billing period (Aug 1-14) - should be counted
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-05"), properties: {})
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-10"), properties: {})
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-12"), properties: {})
+
+          # Events for second contract OUTSIDE its billing period (Aug 15-31) - should NOT be counted
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-20"), properties: {})
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-28"), properties: {})
+        end
+
+        it "only counts events within each contract's billing segment date range" do
+          expect(result).to be_success
+
+          invoice = result.invoices.sole.reload
+          expect(invoice.fees.count).to eq(2)
+
+          first_fee = invoice.fees.find { |f| f.invoiceable == product }
+          second_fee = invoice.fees.find { |f| f.invoiceable == second_product }
+
+          # First contract: only 2 events in Aug 15-31 × $10 = $20 (not 4 events)
+          expect(first_fee).to have_attributes(fee_type: "product", units: 2, events_count: 2, amount_cents: 2_000)
+          # Second contract: only 3 events in Aug 1-14 × $20 = $60 (not 5 events)
+          expect(second_fee).to have_attributes(fee_type: "product", units: 3, events_count: 3, amount_cents: 6_000)
+
+          expect(invoice.total_amount_cents).to eq(8_000)
+        end
+      end
+
+      context "with consolidated contracts having different product filters" do
+        let(:aggregation_type) { :count_agg }
+        let(:field_name) { nil }
+        let(:event_properties) { {"region" => "eu"} }
+        let(:rate_properties) { {"amount" => "10.00"} }
+
+        # First contract product has "region" filter
+        let(:region_filter) { create(:billable_metric_filter, billable_metric:, key: "region", values: %w[eu us]) }
+        let(:product_filter) { create(:product_filter, organization:, product:) }
+
+        # Second contract with different billable metric that has "country" filter
+        let(:second_contract) do
+          create(:contract, organization:, customer:, external_id: "second_contract_ext_id", consolidate_invoice: true)
+        end
+        let(:second_billable_metric) { create(:billable_metric, organization:, aggregation_type: :count_agg) }
+        let(:country_filter) { create(:billable_metric_filter, billable_metric: second_billable_metric, key: "country", values: %w[fr de]) }
+        let(:second_product) { create(:product, :metered, organization:, billable_metric: second_billable_metric) }
+        let(:second_product_filter) { create(:product_filter, organization:, product: second_product) }
+        let(:second_rate_card) { create(:rate_card, organization:, product: second_product, product_filter: second_product_filter, currency: "USD") }
+        let(:second_contract_rate_card) do
+          create(:contract_rate_card, organization:, contract: second_contract, rate_card: second_rate_card, effective_date: Date.parse("2026-07-01"))
+        end
+        let(:second_rate_card_rate) do
+          create(:rate_card_rate, organization:, rate_card: second_rate_card, rate_properties: {"amount" => "20.00"})
+        end
+        let(:second_segment) do
+          create(
+            :billing_segment, organization:, contract: second_contract, customer:,
+            contract_rate_card: second_contract_rate_card, rate_card_rate: second_rate_card_rate,
+            currency: "USD", rate_properties: {"amount" => "20.00"},
+            billing_at: Time.zone.parse("2026-08-31 23:59:59"), cycle_started_at: Time.zone.parse("2026-08-01"),
+            started_at: Time.zone.parse("2026-08-01"), ended_at: Time.zone.parse("2026-08-31 23:59:59")
+          )
+        end
+
+        # First contract rate card with product filter
+        let(:rate_card) { create(:rate_card, organization:, product:, product_filter:, currency: "USD") }
+
+        before do
+          Event.delete_all # clean before events
+          billing_segment
+          second_segment
+
+          # Setup product filter values
+          create(:product_filter_value, organization:, product_filter:, billable_metric_filter: region_filter, value: "eu")
+          create(:product_filter_value, organization:, product_filter: second_product_filter, billable_metric_filter: country_filter, value: "fr")
+
+          # Events for first contract with region=eu (matches filter)
+          create(:event, organization:, customer:, external_subscription_id: contract.external_id,
+            code: billable_metric.code, timestamp: Time.zone.parse("2026-08-10"), properties: {"region" => "eu"})
+          create(:event, organization:, customer:, external_subscription_id: contract.external_id,
+            code: billable_metric.code, timestamp: Time.zone.parse("2026-08-20"), properties: {"region" => "eu"})
+
+          # Events for second contract with country=fr (matches filter)
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-12"), properties: {"country" => "fr"})
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-18"), properties: {"country" => "fr"})
+          create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+            code: second_billable_metric.code, timestamp: Time.zone.parse("2026-08-25"), properties: {"country" => "fr"})
+        end
+
+        it "uses each contract's own filter keys when resolving event combinations" do
+          expect(result).to be_success
+
+          invoice = result.invoices.sole.reload
+          expect(invoice.fees.count).to eq(2)
+
+          first_fee = invoice.fees.find { |f| f.invoiceable == product }
+          second_fee = invoice.fees.find { |f| f.invoiceable == second_product }
+
+          # First contract: 2 events with region=eu × $10 = $20
+          expect(first_fee).to have_attributes(
+            fee_type: "product",
+            units: 2,
+            events_count: 2,
+            amount_cents: 2_000,
+            product_filter_id: product_filter.id
+          )
+          # Second contract: 3 events with country=fr × $20 = $60
+          expect(second_fee).to have_attributes(
+            fee_type: "product",
+            units: 3,
+            events_count: 3,
+            amount_cents: 6_000,
+            product_filter_id: second_product_filter.id
+          )
+
+          expect(invoice.total_amount_cents).to eq(8_000)
+        end
       end
     end
 
@@ -135,7 +534,11 @@
       end
 
       context "when the customer has a default payment method" do
-        let!(:payment_method) { create(:payment_method, organization:, customer:, is_default: true) }
+        let(:payment_method) { create(:payment_method, organization:, customer:, is_default: true) }
+
+        before do
+          payment_method
+        end
 
         it "uses the resolved default payment method" do
           expect(invoice_key[4]).to eq([payment_method.id, "provider"])
@@ -338,7 +741,7 @@
       let(:cycle_start) { Time.utc(2026, 6, 1) }
       let(:cycle_end) { Time.utc(2026, 7, 1) }
       let(:rate_change_at) { Time.utc(2026, 6, 16, 9, 30) }
-      let!(:billing_segment) do
+      let(:billing_segment) do
         build(:billing_segment, organization:, customer:, contract:, contract_rate_card:, rate_card_rate:,
           currency: "USD", rate_properties:, billing_at: cycle_end, cycle_started_at: cycle_start,
           started_at: cycle_start, ended_at: BillingSegment.inclusive_end(rate_change_at)).tap do |segment|
@@ -346,7 +749,7 @@
           segment.save!
         end
       end
-      let!(:second_segment) do
+      let(:second_segment) do
         override = create(:rate_override, organization:, rate_properties: {"amount" => "45.00"}, min_amount_cents:)
         build(:billing_segment, organization:, customer:, contract:, contract_rate_card:, rate_card_rate:,
           rate_override: override, currency: "USD", rate_properties: override.rate_properties,
@@ -357,6 +760,11 @@
         end
       end
 
+      before do
+        billing_segment
+        second_segment
+      end
+
       it "bills each segment's base fee and prorated minimum without counting June 16 twice" do
         segments = [billing_segment.reload, second_segment.reload]
         expect(segments.map(&:duration_in_days)).to eq([16, 14])
@@ -529,7 +937,7 @@
       let(:second_rate_card_rate) do
         create(:rate_card_rate, organization:, rate_card: second_rate_card, rate_properties: {"amount" => "20.00"})
       end
-      let!(:second_segment) do
+      let(:second_segment) do
         create(:billing_segment, organization:, contract: second_contract, customer:,
           contract_rate_card: second_contract_rate_card, rate_card_rate: second_rate_card_rate,
           currency: second_rate_card.currency, rate_properties: {"amount" => "20.00"},
@@ -537,6 +945,10 @@
           started_at: Time.zone.parse("2026-08-01"), ended_at: Time.zone.parse("2026-08-31 23:59:59"))
       end
 
+      before do
+        second_segment
+      end
+
       it "consolidates same-date segments into one invoice" do
         expect(result).to be_success
         invoice = result.invoices.sole.reload
diff --git a/spec/services/events/billing_period_filter_service_spec.rb b/spec/services/events/billing_period_filter_service_spec.rb
index 090c56290c5..c40e09e5acf 100644
--- a/spec/services/events/billing_period_filter_service_spec.rb
+++ b/spec/services/events/billing_period_filter_service_spec.rb
@@ -53,7 +53,7 @@
 
   describe ".for_billing_segments!" do
     subject(:filter_result) do
-      described_class.for_billing_segments!(contract:, billing_segments: [billing_segment])
+      described_class.for_billing_segments!(billing_segments: [billing_segment])
     end
 
     let(:contract) { create(:contract, organization:, customer:, external_id: "contract_external_id") }
@@ -86,7 +86,7 @@
 
     context "without billing segments" do
       subject(:filter_result) do
-        described_class.for_billing_segments!(contract:, billing_segments: [])
+        described_class.for_billing_segments!(billing_segments: [])
       end
 
       it "succeeds with no filter targets" do
@@ -104,6 +104,7 @@
           organization_id: organization.id,
           external_subscription_id: contract.external_id,
           timestamp: billing_segment.started_at + 5.days,
+          created_at: Time.zone.parse("2026-08-10 00:00:00"),
           code: billable_metric.code,
           properties: {"region" => "eu"}
         )
@@ -114,7 +115,7 @@
           result = filter_result
 
           expect(result).to be_success
-          expect(result.filter_targets.transform_values(&:keys)).to eq({product.target_key => [nil]})
+          expect(result.filter_targets.transform_values(&:keys)).to eq({billing_segment.target_key => [nil]})
         end
       end
 
@@ -131,13 +132,13 @@
           result = filter_result
 
           expect(result).to be_success
-          expect(result.filter_targets.transform_values(&:keys)).to eq({product.target_key => [product_filter.id]})
+          expect(result.filter_targets.transform_values(&:keys)).to eq({billing_segment.target_key => [product_filter.id]})
         end
 
         it "returns the last seen timestamp for the product filter" do
           result = filter_result
 
-          expect(result.filter_targets[product.target_key][product_filter.id]).to be_present
+          expect(result.filter_targets[billing_segment.target_key][product_filter.id]).to be_present
         end
 
         context "when the product filter selects the key only" do
@@ -149,7 +150,7 @@
             result = filter_result
 
             expect(result).to be_success
-            expect(result.filter_targets.transform_values(&:keys)).to eq({product.target_key => [product_filter.id]})
+            expect(result.filter_targets.transform_values(&:keys)).to eq({billing_segment.target_key => [product_filter.id]})
           end
         end
       end
@@ -170,7 +171,7 @@
 
     context "with a relation containing segments sharing a billable metric" do
       subject(:filter_result) do
-        described_class.for_billing_segments!(contract:, billing_segments: contract.billing_segments)
+        described_class.for_billing_segments!(billing_segments: contract.billing_segments)
       end
 
       before do
@@ -192,11 +193,70 @@
       end
     end
 
+    context "with segments using the same product across contracts" do
+      subject(:filter_result) do
+        described_class.for_billing_segments!(billing_segments: [billing_segment, second_segment])
+      end
+
+      let(:second_contract) { create(:contract, organization:, customer:, external_id: "second_contract_external_id") }
+      let(:first_contract_event_created_at) { Time.zone.parse("2026-08-12 00:00:00") }
+      let(:second_contract_event_created_at) { Time.zone.parse("2026-08-13 00:00:00") }
+      let(:second_contract_rate_card) { create(:contract_rate_card, organization:, contract: second_contract, rate_card:) }
+      let(:second_segment) do
+        create(
+          :billing_segment,
+          organization:,
+          customer:,
+          contract: second_contract,
+          contract_rate_card: second_contract_rate_card,
+          rate_card_rate:,
+          cycle_started_at: billing_segment.cycle_started_at,
+          started_at: billing_segment.started_at,
+          ended_at: billing_segment.ended_at
+        )
+      end
+
+      before do
+        create(:event, organization:, customer:, external_subscription_id: contract.external_id,
+          timestamp: billing_segment.started_at + 1.day, created_at: first_contract_event_created_at,
+          code: billable_metric.code, properties: {region: "eu"})
+        create(:event, organization:, customer:, external_subscription_id: second_contract.external_id,
+          timestamp: second_segment.started_at + 1.day, created_at: second_contract_event_created_at,
+          code: billable_metric.code, properties: {region: "us"})
+      end
+
+      it "keeps each contract's product bucket separate" do
+        result = filter_result
+
+        expect(result.filter_targets.keys).to match_array([billing_segment.target_key, second_segment.target_key])
+        expect(result.filter_targets[billing_segment.target_key].keys).to eq([nil])
+        expect(result.filter_targets[second_segment.target_key].keys).to eq([nil])
+      end
+
+      context "with a product filter" do
+        let(:product_filter) { create(:product_filter, organization:, product:) }
+        let(:billable_metric_filter) { create(:billable_metric_filter, billable_metric:, key: "region", values: %w[eu us]) }
+
+        before do
+          create(:product_filter_value, organization:, product_filter:, billable_metric_filter:, value: "eu")
+        end
+
+        it "matches events and timestamps only within their contract" do
+          result = filter_result
+
+          expect(result.filter_targets[billing_segment.target_key].keys).to eq([product_filter.id])
+          expect(result.filter_targets[second_segment.target_key].keys).to eq([nil])
+          expect(result.filter_targets[billing_segment.target_key][product_filter.id]).to eq(first_contract_event_created_at)
+          expect(result.filter_targets[second_segment.target_key][nil]).to eq(second_contract_event_created_at)
+        end
+      end
+    end
+
     context "with recurring product usage" do
       let(:billable_metric) { create(:sum_billable_metric, organization:, recurring: true) }
 
       it "seeds the default bucket without events" do
-        expect(filter_result.filter_targets).to eq({product.target_key => {nil => billing_segment.started_at}})
+        expect(filter_result.filter_targets).to eq({billing_segment.target_key => {nil => billing_segment.started_at}})
       end
 
       it "uses the combined period start for segments sharing a product" do
@@ -204,9 +264,9 @@
           cycle_started_at: billing_segment.cycle_started_at + 1.month,
           started_at: billing_segment.started_at + 1.month, ended_at: billing_segment.ended_at + 1.month)
 
-        result = described_class.for_billing_segments!(contract:, billing_segments: [later_segment, billing_segment])
+        result = described_class.for_billing_segments!(billing_segments: [later_segment, billing_segment])
 
-        expect(result.filter_targets).to eq({product.target_key => {nil => billing_segment.started_at}})
+        expect(result.filter_targets).to eq({billing_segment.target_key => {nil => billing_segment.started_at}})
       end
 
       context "with product filters" do
@@ -218,7 +278,7 @@
         end
 
         it "seeds current filters and the default bucket without events" do
-          expect(filter_result.filter_targets).to eq({product.target_key => {
+          expect(filter_result.filter_targets).to eq({billing_segment.target_key => {
             product_filter.id => billing_segment.started_at,
             nil => billing_segment.started_at
           }})
@@ -234,7 +294,7 @@
           end
 
           it "updates the historical usage bucket with the ingestion timestamp" do
-            expect(filter_result.filter_targets).to eq({product.target_key => {
+            expect(filter_result.filter_targets).to eq({billing_segment.target_key => {
               product_filter.id => ingested_at,
               nil => billing_segment.started_at
             }})
@@ -242,11 +302,11 @@
 
           context "when timestamp aggregation is disabled" do
             subject(:filter_result) do
-              described_class.for_billing_segments!(contract:, billing_segments: [billing_segment], with_last_seen_at: false)
+              described_class.for_billing_segments!(billing_segments: [billing_segment], with_last_seen_at: false)
             end
 
             it "retains the seeded timestamps" do
-              expect(filter_result.filter_targets).to eq({product.target_key => {
+              expect(filter_result.filter_targets).to eq({billing_segment.target_key => {
                 product_filter.id => billing_segment.started_at,
                 nil => billing_segment.started_at
               }})
@@ -259,7 +319,7 @@
         event_store = instance_double(Events::Stores::PostgresStore, distinct_codes_and_property_combinations: [])
         allow(Events::Stores::StoreFactory).to receive(:new_instance).and_return(event_store)
 
-        described_class.for_billing_segments!(contract:, billing_segments: [billing_segment],
+        described_class.for_billing_segments!(billing_segments: [billing_segment],
           codes: [billable_metric.code, "other_code"], with_last_seen_at: false)
 
         expect(event_store).to have_received(:distinct_codes_and_property_combinations)
@@ -269,7 +329,7 @@
       end
 
       it "does not seed recurring products excluded by explicit codes" do
-        result = described_class.for_billing_segments!(contract:, billing_segments: [billing_segment], codes: ["unknown_code"])
+        result = described_class.for_billing_segments!(billing_segments: [billing_segment], codes: ["unknown_code"])
 
         expect(result.filter_targets).to eq({})
       end
@@ -279,7 +339,7 @@
           code: billable_metric.code, timestamp: billing_segment.ended_at + 1.day,
           created_at: billing_segment.ended_at + 2.days)
 
-        expect(filter_result.filter_targets).to eq({product.target_key => {nil => billing_segment.started_at}})
+        expect(filter_result.filter_targets).to eq({billing_segment.target_key => {nil => billing_segment.started_at}})
       end
     end
 
@@ -296,7 +356,7 @@
 
     context "when codes restrict the lookup" do
       subject(:filter_result) do
-        described_class.for_billing_segments!(contract:, billing_segments: [billing_segment], codes: ["unknown_code"])
+        described_class.for_billing_segments!(billing_segments: [billing_segment], codes: ["unknown_code"])
       end
 
       before do