diff --git a/app/graphql/types/contract_applied_rate_cards/object.rb b/app/graphql/types/contract_applied_rate_cards/object.rb
index fb09c3abdc3..54f9e0c61ea 100644
--- a/app/graphql/types/contract_applied_rate_cards/object.rb
+++ b/app/graphql/types/contract_applied_rate_cards/object.rb
@@ -18,7 +18,7 @@ class Object < Types::BaseObject
       field :billing_anchor_date, GraphQL::Types::ISO8601Date, null: false
       field :effective_date, GraphQL::Types::ISO8601Date, null: false
       field :ended_date, GraphQL::Types::ISO8601Date, null: true
-      field :next_billing_at, GraphQL::Types::ISO8601DateTime, null: false
+      field :next_billing_at, GraphQL::Types::ISO8601DateTime, null: true
 
       field :rate_phases, [Types::RatePhases::Object], null: false
       field :rate_phases_count, Integer, null: false
diff --git a/app/models/contract_rate_card.rb b/app/models/contract_rate_card.rb
index 09f8d358518..e241ce5a781 100644
--- a/app/models/contract_rate_card.rb
+++ b/app/models/contract_rate_card.rb
@@ -16,7 +16,6 @@ class ContractRateCard < ApplicationRecord
   has_many :billing_segments
 
   validates :billing_anchor_date, presence: true
-  validates :next_billing_at, presence: true
   validates :effective_date, presence: true
   validates :units, numericality: {greater_than_or_equal_to: 0}, allow_nil: true
   validates :rate_card_id, uniqueness: {scope: :contract_id, conditions: -> { where(deleted_at: nil, ended_date: nil) }}
@@ -68,7 +67,7 @@ def validate_effective_before_ended
 #  deleted_at          :datetime
 #  effective_date      :date             not null
 #  ended_date          :date
-#  next_billing_at     :datetime         not null
+#  next_billing_at     :datetime
 #  units               :decimal(, )
 #  created_at          :datetime         not null
 #  updated_at          :datetime         not null
diff --git a/app/serializers/v2/contract_applied_rate_card_serializer.rb b/app/serializers/v2/contract_applied_rate_card_serializer.rb
index 57015d9cd39..77a056416e8 100644
--- a/app/serializers/v2/contract_applied_rate_card_serializer.rb
+++ b/app/serializers/v2/contract_applied_rate_card_serializer.rb
@@ -11,7 +11,7 @@ def serialize
         effective_date: model.effective_date.iso8601,
         ended_date: model.ended_date&.iso8601,
         billing_anchor_date: model.billing_anchor_date.iso8601,
-        next_billing_at: model.next_billing_at.iso8601,
+        next_billing_at: model.next_billing_at&.iso8601,
         # size counts the loaded collection when the caller preloaded it.
         rate_phases_count: model.rate_phases.size,
         created_at: model.created_at.iso8601,
diff --git a/app/services/billing/rate_cards/schedule.rb b/app/services/billing/rate_cards/schedule.rb
index 869b2e90bcd..3852044f8c4 100644
--- a/app/services/billing/rate_cards/schedule.rb
+++ b/app/services/billing/rate_cards/schedule.rb
@@ -16,11 +16,26 @@ def initialize(rates:, terms:, phases:, starts_at:, anchor_date:, timezone:, end
       end
 
       # A rate change can make a segment due before its cycle ends.
-      def segments_due_by(timestamp)
+      def segments_due_by(timestamp, billing_from: nil)
         cycles = walker.walk_to(timestamp, from: resume_at)
 
-        billable_segments_of(cycles).select do |segment|
-          segment.billing_at <= timestamp
+        cycles.flat_map do |cycle|
+          segments = billable_segments_of([cycle])
+
+          if billing_from
+            # A saved arrears clock can point to the original cycle end even
+            # after termination shortens it. Select against the unshortened
+            # pricing windows, then persist the actual service boundaries.
+            full_cycle = cycle.with(ended_at: cycle.calendar.interval_containing(cycle.started_at).end)
+            eligible_starts = billable_segments_of([full_cycle]).filter_map do |segment|
+              if segment.billing_at >= billing_from || (segment.started_at...segment.ended_at).cover?(billing_from)
+                segment.started_at
+              end
+            end
+            segments = segments.select { eligible_starts.include?(it.started_at) }
+          end
+
+          segments.select { it.billing_at <= timestamp }
         end
       end
 
diff --git a/app/services/billing_segments/schedule_service.rb b/app/services/billing_segments/schedule_service.rb
new file mode 100644
index 00000000000..15b5f410cbc
--- /dev/null
+++ b/app/services/billing_segments/schedule_service.rb
@@ -0,0 +1,100 @@
+# frozen_string_literal: true
+
+module BillingSegments
+  # Persists the calendar's due slices. The clock is both the lower billing
+  # bound (backdated contracts can join the current period) and the next wake-up.
+  class ScheduleService < BaseService
+    Result = BaseResult[:billing_segments]
+
+    def initialize(customer:, timestamp: Time.current)
+      @customer = customer
+      @timestamp = timestamp
+      super
+    end
+
+    def call
+      result.billing_segments = []
+
+      if customer.nil?
+        return result.not_found_failure!(resource: "customer")
+      end
+
+      scheduled = []
+      ActiveRecord::Base.transaction do
+        Customers::LockService.call!(customer:, scope: :billing_schedule) do
+          due_cards.find_each do |card|
+            scheduled.concat(schedule_card(card))
+          end
+        end
+      end
+
+      result.billing_segments = scheduled
+      result
+    rescue ActiveRecord::RecordInvalid => error
+      result.record_validation_failure!(record: error.record)
+    rescue BaseService::FailedResult => error
+      result.fail_with_error!(error)
+    end
+
+    private
+
+    attr_reader :customer, :timestamp
+
+    def due_cards
+      ContractRateCard.joins(:contract)
+        .where(organization_id: customer.organization_id, next_billing_at: ..timestamp)
+        .where(contracts: {customer_id: customer.id, status: %w[active terminated], started_at: ..timestamp})
+        .includes(:rate_card, contract: :customer)
+    end
+
+    def schedule_card(card)
+      schedule = Billing::RateCards::BuildScheduleService.call!(contract_rate_card: card).schedule
+      scheduled = schedule.segments_due_by(timestamp, billing_from: card.next_billing_at).filter_map do |segment|
+        persist_segment(card, segment)
+      end
+
+      # nil means the schedule is exhausted, so this card is no longer due.
+      card.update!(next_billing_at: schedule.next_billing_at(after: timestamp))
+      scheduled
+    end
+
+    def persist_segment(card, segment)
+      # Resuming replays the last cycle, which can contain several segments.
+      # Keep existing snapshots and statuses intact, including on a stale-clock retry.
+      if card.billing_segments.exists?(started_at: segment.started_at)
+        return
+      end
+
+      card.billing_segments.create!(
+        organization: customer.organization,
+        customer:,
+        contract: card.contract,
+        cycle_started_at: segment.cycle_started_at,
+        started_at: segment.started_at,
+        ended_at: BillingSegment.inclusive_end(segment.ended_at),
+        billing_at: segment.billing_at,
+        rate_card_rate: segment.rate,
+        rate_override: segment.rate_override,
+        rate_properties: (segment.rate_override || segment.rate).properties,
+        currency: card.rate_card.currency,
+        pricing_unit: pricing_unit(card),
+        proration_ratio: segment.proration_ratio,
+        status: :pending
+      )
+    end
+
+    def pricing_unit(card)
+      code = card.rate_card.applied_pricing_unit_code
+
+      if code.present?
+        unit = customer.organization.pricing_units.find_by(code:)
+
+        if unit.nil?
+          result.not_found_failure!(resource: "pricing_unit").raise_if_error!
+        end
+
+        unit
+      end
+    end
+  end
+end
diff --git a/app/services/customers/lock_service.rb b/app/services/customers/lock_service.rb
index dabfd061802..576dc021d93 100644
--- a/app/services/customers/lock_service.rb
+++ b/app/services/customers/lock_service.rb
@@ -15,7 +15,7 @@ module Customers
   #   locking (lock_version), concurrent updates will raise StaleObjectError.
   #
   class LockService < BaseLockService
-    VALID_SCOPES = %i[prepaid_credit payment_method credit_note coupon].freeze
+    VALID_SCOPES = %i[prepaid_credit payment_method credit_note coupon billing_schedule].freeze
 
     def initialize(customer:, scope:, timeout_seconds: ACQUIRE_LOCK_TIMEOUT, transaction: true)
       @customer = customer
diff --git a/db/migrate/20260909154904_allow_exhausted_contract_rate_card_schedules.rb b/db/migrate/20260909154904_allow_exhausted_contract_rate_card_schedules.rb
new file mode 100644
index 00000000000..97e708b148e
--- /dev/null
+++ b/db/migrate/20260909154904_allow_exhausted_contract_rate_card_schedules.rb
@@ -0,0 +1,7 @@
+# frozen_string_literal: true
+
+class AllowExhaustedContractRateCardSchedules < ActiveRecord::Migration[8.0]
+  def change
+    change_column_null :contract_rate_cards, :next_billing_at, true
+  end
+end
diff --git a/db/structure.sql b/db/structure.sql
index ecf5b3b7db7..e71684817fc 100644
--- a/db/structure.sql
+++ b/db/structure.sql
@@ -2631,7 +2631,7 @@ CREATE TABLE public.contract_rate_cards (
     contract_id uuid NOT NULL,
     rate_card_id uuid NOT NULL,
     billing_anchor_date date NOT NULL,
-    next_billing_at timestamp without time zone NOT NULL,
+    next_billing_at timestamp without time zone,
     effective_date date NOT NULL,
     ended_date date,
     units numeric,
@@ -14877,6 +14877,7 @@ INSERT INTO "schema_migrations" (version) VALUES
 ('20260911144853'),
 ('20260910151708'),
 ('20260910095513'),
+('20260909154904'),
 ('20260909103355'),
 ('20260908222044'),
 ('20260908211313'),
@@ -16007,4 +16008,3 @@ INSERT INTO "schema_migrations" (version) VALUES
 ('20220530091046'),
 ('20220526101535'),
 ('20220525122759');
-
diff --git a/schema.graphql b/schema.graphql
index f7cf6ad7bb7..aad3b4a2a8f 100644
--- a/schema.graphql
+++ b/schema.graphql
@@ -1538,7 +1538,7 @@ type ContractAppliedRateCard {
   effectiveDate: ISO8601Date!
   endedDate: ISO8601Date
   id: ID!
-  nextBillingAt: ISO8601DateTime!
+  nextBillingAt: ISO8601DateTime
   product: Product!
   rateCard: RateCard!
   ratePhases: [RatePhase!]!
diff --git a/schema.json b/schema.json
index 7c7a2f8a585..2a9fa280a32 100644
--- a/schema.json
+++ b/schema.json
@@ -9240,13 +9240,9 @@
               "description": null,
               "args": [],
               "type": {
-                "kind": "NON_NULL",
-                "name": null,
-                "ofType": {
-                  "kind": "SCALAR",
-                  "name": "ISO8601DateTime",
-                  "ofType": null
-                }
+                "kind": "SCALAR",
+                "name": "ISO8601DateTime",
+                "ofType": null
               },
               "isDeprecated": false,
               "deprecationReason": null
diff --git a/spec/graphql/resolvers/contracts_resolver_spec.rb b/spec/graphql/resolvers/contracts_resolver_spec.rb
index 069811593d4..326f75f977a 100644
--- a/spec/graphql/resolvers/contracts_resolver_spec.rb
+++ b/spec/graphql/resolvers/contracts_resolver_spec.rb
@@ -99,4 +99,24 @@ def count_queries(table)
       expect(counts.sum).to eq(2)
     end
   end
+
+  context "when a rate card has exhausted its billing schedule" do
+    let(:query) do
+      <<~GQL
+        query {
+          contracts(limit: 5, status: [active]) {
+            collection { id appliedRateCards { id nextBillingAt } }
+          }
+        }
+      GQL
+    end
+
+    it "returns the card with a null next billing date" do
+      card = create(:contract_rate_card, organization:, contract: active_contract, next_billing_at: nil)
+
+      expect(execution["errors"]).to be_nil
+      expect(execution["data"]["contracts"]["collection"].sole["appliedRateCards"])
+        .to eq([{"id" => card.id, "nextBillingAt" => nil}])
+    end
+  end
 end
diff --git a/spec/models/contract_rate_card_spec.rb b/spec/models/contract_rate_card_spec.rb
index fae551a113c..336a271b3d8 100644
--- a/spec/models/contract_rate_card_spec.rb
+++ b/spec/models/contract_rate_card_spec.rb
@@ -58,7 +58,7 @@
 
   describe "validations" do
     it { is_expected.to validate_presence_of(:billing_anchor_date) }
-    it { is_expected.to validate_presence_of(:next_billing_at) }
+    it { is_expected.to allow_value(nil).for(:next_billing_at) }
     it { is_expected.to validate_presence_of(:effective_date) }
     it { is_expected.to validate_numericality_of(:units).is_greater_than_or_equal_to(0).allow_nil }
 
diff --git a/spec/requests/api/v2/contract_rate_cards_controller_spec.rb b/spec/requests/api/v2/contract_rate_cards_controller_spec.rb
index 31c58ec5604..b57a192463a 100644
--- a/spec/requests/api/v2/contract_rate_cards_controller_spec.rb
+++ b/spec/requests/api/v2/contract_rate_cards_controller_spec.rb
@@ -84,6 +84,15 @@
       expect(response).to have_http_status(:success)
       expect(json[:applied_rate_cards].map { it[:lago_id] }).to eq([contract_rate_card.id])
     end
+
+    it "serializes an exhausted billing schedule" do
+      contract_rate_card.update!(next_billing_at: nil)
+
+      subject
+
+      expect(response).to have_http_status(:success)
+      expect(json[:applied_rate_cards].sole[:next_billing_at]).to be_nil
+    end
   end
 
   describe "GET /api/v2/contracts/:external_id/applied_rate_cards/:code" do
diff --git a/spec/services/billing/rate_cards/schedule_spec.rb b/spec/services/billing/rate_cards/schedule_spec.rb
index 8c09013c96d..5ceffa7db33 100644
--- a/spec/services/billing/rate_cards/schedule_spec.rb
+++ b/spec/services/billing/rate_cards/schedule_spec.rb
@@ -143,6 +143,34 @@ def windows(cycles)
     end
   end
 
+  describe "an initial billing bound on a shortened schedule" do
+    let(:ends_at) { Time.utc(2022, 1, 25, 12) }
+
+    it "includes the final arrears segment when the clock still points to its original end" do
+      segment = schedule.segments_due_by(Time.utc(2022, 2, 1), billing_from: Time.utc(2022, 2, 1)).sole
+
+      expect(segment.started_at).to eq(starts_at)
+      expect(segment.ended_at).to eq(ends_at)
+      expect(segment.billing_at).to eq(ends_at)
+      expect(segment.proration_ratio).to eq(11.fdiv(31))
+    end
+
+    it "does not resurrect historical cycles before the initial billing bound" do
+      expect(schedule.segments_due_by(Time.utc(2022, 4, 1), billing_from: Time.utc(2022, 4, 1))).to eq([])
+    end
+
+    context "with a price change inside the final cycle" do
+      let(:rates) { [card_rate(Time.utc(2000, 1, 1)), card_rate(Time.utc(2022, 1, 20))] }
+
+      it "keeps the bound at the pricing segment rather than reopening the entire cycle" do
+        segments = schedule.segments_due_by(Time.utc(2022, 2, 1), billing_from: Time.utc(2022, 2, 1))
+
+        expect(segments.map(&:started_at)).to eq([Time.utc(2022, 1, 20)])
+        expect(segments.sole.ended_at).to eq(ends_at)
+      end
+    end
+  end
+
   describe "when a window falls due" do
     it "bills at the end of the cycle in arrears" do
       expect(schedule.segments_due_by(Time.utc(2022, 4, 1)).map(&:billing_at))
diff --git a/spec/services/billing_segments/schedule_service_spec.rb b/spec/services/billing_segments/schedule_service_spec.rb
new file mode 100644
index 00000000000..3d16562af7c
--- /dev/null
+++ b/spec/services/billing_segments/schedule_service_spec.rb
@@ -0,0 +1,370 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+RSpec.describe BillingSegments::ScheduleService do
+  subject(:result) { described_class.call(customer:, timestamp:) }
+
+  let(:organization) { create(:organization) }
+  let(:customer) { create(:customer, organization:, timezone: "UTC") }
+  let(:contract) { create(:contract, organization:, customer:, started_at: Time.utc(2026, 1, 15)) }
+  let(:product) { create(:product, :fixed, organization:) }
+  let(:rate_card) { create(:rate_card, organization:, product:, proration: true) }
+  let(:rate) { create(:rate_card_rate, organization:, rate_card:, effective_from: Time.utc(2025, 1, 1)) }
+  let(:timestamp) { Time.utc(2026, 2, 1) }
+  let(:card) do
+    create(:contract_rate_card, organization:, contract:, rate_card:,
+      effective_date: Date.new(2026, 1, 15), billing_anchor_date: Date.new(2026, 1, 1),
+      next_billing_at: Time.utc(2026, 1, 15))
+  end
+
+  before do
+    rate
+    card
+  end
+
+  it "persists the calendar's priced segment and advances the clock" do
+    expect(result).to be_success
+    segment = result.billing_segments.sole.reload
+
+    expect(segment).to have_attributes(
+      organization_id: organization.id, customer_id: customer.id, contract_id: contract.id,
+      contract_rate_card_id: card.id, rate_card_rate_id: rate.id, rate_override_id: nil,
+      rate_properties: {"amount" => "10"}, currency: "EUR", pricing_unit_id: nil,
+      status: "pending", invoice_id: nil,
+      cycle_started_at: Time.utc(2026, 1, 15), started_at: Time.utc(2026, 1, 15),
+      ended_at: BillingSegment.inclusive_end(Time.utc(2026, 2, 1)), billing_at: Time.utc(2026, 2, 1)
+    )
+    expect(segment.proration_ratio).to be_within(1e-10).of(17.fdiv(31))
+    expect(card.reload.next_billing_at).to eq(Time.utc(2026, 3, 1))
+  end
+
+  it "returns no new segments when the same timestamp is retried" do
+    first = result.billing_segments.sole
+    retry_result = described_class.call(customer:, timestamp:)
+
+    expect(retry_result).to be_success
+    expect(retry_result.billing_segments).to eq([])
+    expect(card.billing_segments.reload).to eq([first])
+  end
+
+  it "does not rewrite an existing snapshot or status when the clock is stale" do
+    first = result.billing_segments.sole
+    first.update!(status: :done)
+    card.update!(next_billing_at: Time.utc(2026, 1, 15))
+    rate.update!(rate_properties: {"amount" => "99"})
+
+    retry_result = described_class.call(customer:, timestamp:)
+
+    expect(retry_result).to be_success
+    expect(retry_result.billing_segments).to eq([])
+    expect(first.reload).to have_attributes(status: "done", rate_properties: {"amount" => "10"})
+    expect(card.billing_segments.count).to eq(1)
+    expect(card.reload.next_billing_at).to eq(Time.utc(2026, 3, 1))
+  end
+
+  it "recovers all due periods after a delayed run" do
+    catch_up = described_class.call(customer:, timestamp: Time.utc(2026, 4, 15))
+
+    expect(catch_up).to be_success
+    expect(catch_up.billing_segments.map(&:billing_at))
+      .to eq([Time.utc(2026, 2, 1), Time.utc(2026, 3, 1), Time.utc(2026, 4, 1)])
+    expect(card.reload.next_billing_at).to eq(Time.utc(2026, 5, 1))
+  end
+
+  it "respects the initial clock when a backdated contract joins a later period" do
+    card.update!(next_billing_at: Time.utc(2026, 4, 1))
+
+    catch_up = described_class.call(customer:, timestamp: Time.utc(2026, 5, 1))
+
+    expect(catch_up).to be_success
+    expect(catch_up.billing_segments.map(&:started_at)).to eq([Time.utc(2026, 3, 1), Time.utc(2026, 4, 1)])
+  end
+
+  it "does not persist an arrears segment before it is due" do
+    early = described_class.call(customer:, timestamp: Time.utc(2026, 1, 20))
+
+    expect(early).to be_success
+    expect(early.billing_segments).to eq([])
+    expect(card.reload.next_billing_at).to eq(Time.utc(2026, 2, 1))
+  end
+
+  context "without proration" do
+    let(:rate_card) { create(:rate_card, organization:, product:, proration: false) }
+
+    it "stores a full-price ratio even for a partial first period" do
+      segment = result.billing_segments.sole
+
+      expect(segment.duration_in_days).to eq(17)
+      expect(segment.proration_ratio).to eq(1)
+    end
+  end
+
+  context "with concurrent runs for the same customer", transaction: false do
+    it "persists each due segment once and advances the clock once" do
+      customer_id = customer.id
+      run_at = timestamp
+      ready = Queue.new
+      start = Queue.new
+
+      threads = Array.new(2) do
+        Thread.new do
+          ActiveRecord::Base.connection_pool.with_connection do
+            ready << true
+            start.pop
+            described_class.call!(customer: Customer.find(customer_id), timestamp: run_at)
+          end
+        end
+      end
+
+      results = Timeout.timeout(10) do
+        2.times { ready.pop }
+        2.times { start << true }
+        threads.map(&:value)
+      end
+
+      expect(results.map { it.billing_segments.size }.sort).to eq([0, 1])
+      expect(card.billing_segments.count).to eq(1)
+      expect(card.reload.next_billing_at).to eq(Time.utc(2026, 3, 1))
+    ensure
+      threads&.each { it.kill if it.alive? }
+      threads&.each(&:join)
+    end
+  end
+
+  context "with a bounded introductory phase" do
+    let(:override) do
+      create(:rate_override, organization:, billing_interval_count: 1, billing_interval_unit: "week",
+        rate_properties: {"amount" => "3"})
+    end
+
+    before do
+      card.update!(billing_anchor_date: card.effective_date)
+      create(:rate_phase, organization:, plan_rate_card: nil, contract_rate_card: card,
+        position: 1, code: "intro", billing_interval_cycle_count: 2, rate_override: override)
+    end
+
+    it "resumes through a phase transition with the correct cadence and price" do
+      first = described_class.call!(customer:, timestamp: Time.utc(2026, 1, 22)).billing_segments.sole
+      later = described_class.call!(customer:, timestamp: Time.utc(2026, 2, 28)).billing_segments
+
+      expect([first, *later].map { [it.started_at, it.billing_at, it.rate_override_id] }).to eq([
+        [Time.utc(2026, 1, 15), Time.utc(2026, 1, 22), override.id],
+        [Time.utc(2026, 1, 22), Time.utc(2026, 1, 29), override.id],
+        [Time.utc(2026, 1, 29), Time.utc(2026, 2, 28), nil]
+      ])
+      expect(later.map(&:rate_properties)).to eq([{"amount" => "3"}, {"amount" => "10"}])
+      expect(card.reload.next_billing_at).to eq(Time.utc(2026, 3, 29))
+    end
+  end
+
+  it "excludes other customers, unsigned contracts, discarded cards and future clocks" do
+    card.update!(next_billing_at: Time.utc(2026, 3, 1))
+    other_customer = create(:customer, organization:)
+    other_contract = create(:contract, organization:, customer: other_customer, started_at: contract.started_at)
+    create(:contract_rate_card, organization:, contract: other_contract, rate_card:, **card.attributes.slice(
+      "effective_date", "billing_anchor_date"
+    ), next_billing_at: Time.utc(2026, 1, 15))
+
+    %i[pending canceled].each do |status|
+      unsigned = create(:contract, organization:, customer:, status:, started_at: contract.started_at)
+      create(:contract_rate_card, organization:, contract: unsigned, rate_card:,
+        effective_date: card.effective_date, next_billing_at: Time.utc(2026, 1, 15))
+    end
+    discarded = create(:contract_rate_card, organization:, contract:, next_billing_at: Time.utc(2026, 1, 15))
+    discarded.discard!
+
+    expect(result).to be_success
+    expect(result.billing_segments).to eq([])
+    expect(BillingSegment.count).to eq(0)
+  end
+
+  context "with a rate change inside the cycle" do
+    let!(:next_rate) do
+      create(:rate_card_rate, organization:, rate_card:,
+        effective_from: Time.utc(2026, 1, 20), rate_properties: {"amount" => "20"})
+    end
+
+    it "resumes a partly persisted cycle without duplicating its first segment" do
+      first_run = described_class.call(customer:, timestamp: Time.utc(2026, 1, 20))
+      first = first_run.billing_segments.sole
+      expect(first.rate_card_rate).to eq(rate)
+      expect(card.reload.next_billing_at).to eq(Time.utc(2026, 2, 1))
+
+      second = result.billing_segments.sole
+      expect(second.rate_card_rate).to eq(next_rate)
+      expect(second.started_at).to eq(BillingSegment.exclusive_end(first.ended_at))
+      expect([first.cycle_started_at, second.cycle_started_at]).to eq([Time.utc(2026, 1, 15)] * 2)
+      expect([first.duration_in_days, second.duration_in_days]).to eq([5, 12])
+      expect(card.billing_segments.count).to eq(2)
+    end
+  end
+
+  context "with advance billing" do
+    let(:rate_card) { create(:rate_card, :advance, organization:, product:, proration: true) }
+    let(:timestamp) { Time.utc(2026, 1, 15) }
+
+    it "persists the segment when it opens, with its future service end" do
+      segment = result.billing_segments.sole
+
+      expect(segment.billing_at).to eq(timestamp)
+      expect(segment.ended_at).to eq(BillingSegment.inclusive_end(Time.utc(2026, 2, 1)))
+      expect(segment.proration_ratio).to be_within(1e-10).of(17.fdiv(31))
+      expect(card.reload.next_billing_at).to eq(Time.utc(2026, 2, 1))
+    end
+
+    context "when the initial clock is the signing instant inside the segment" do
+      let(:timestamp) { Time.utc(2026, 1, 15, 12) }
+
+      before do
+        contract.update!(started_at: timestamp)
+        card.update!(next_billing_at: timestamp)
+      end
+
+      it "includes the advance segment serving that instant" do
+        segment = result.billing_segments.sole
+
+        expect(segment.started_at).to eq(Time.utc(2026, 1, 15))
+        expect(segment.billing_at).to eq(Time.utc(2026, 1, 15))
+        expect(card.reload.next_billing_at).to eq(Time.utc(2026, 2, 1))
+      end
+    end
+
+    it "does not recover advance periods before a backdated contract's initial clock" do
+      card.update!(next_billing_at: Time.utc(2026, 3, 15, 12))
+
+      catch_up = described_class.call(customer:, timestamp: Time.utc(2026, 4, 1))
+
+      expect(catch_up.billing_segments.map(&:billing_at)).to eq([Time.utc(2026, 3, 1), Time.utc(2026, 4, 1)])
+    end
+  end
+
+  context "with an override and a pricing unit" do
+    let(:pricing_unit) { create(:pricing_unit, organization:) }
+    let(:rate_card) do
+      create(:rate_card, organization:, product:, proration: true, applied_pricing_unit_code: pricing_unit.code)
+    end
+    let(:rate) do
+      create(:rate_card_rate, organization:, rate_card:, effective_from: Time.utc(2025, 1, 1),
+        applied_pricing_unit_conversion_rate: 2)
+    end
+    let(:override) do
+      create(:rate_override, organization:, rate_properties: {"amount" => "3"}, pricing_unit_conversion_rate: 4)
+    end
+
+    before do
+      create(:rate_phase, organization:, plan_rate_card: nil, contract_rate_card: card,
+        position: 1, code: "intro", rate_override: override)
+    end
+
+    it "stores the effective price and keeps both pricing references" do
+      segment = result.billing_segments.sole
+
+      expect(segment).to have_attributes(rate_card_rate_id: rate.id, rate_override_id: override.id,
+        pricing_unit_id: pricing_unit.id, currency: "EUR", rate_properties: {"amount" => "3"})
+      expect(segment.pricing_unit_conversion_rate).to eq(4)
+    end
+
+    it "fails without advancing the clock if the configured pricing unit no longer exists" do
+      pricing_unit.update!(code: "renamed_unit")
+      original_clock = card.next_billing_at
+
+      expect(result).not_to be_success
+      expect(result.error.error_code).to eq("pricing_unit_not_found")
+      expect(result.billing_segments).to eq([])
+      expect(card.billing_segments.count).to eq(0)
+      expect(card.reload.next_billing_at).to eq(original_clock)
+    end
+  end
+
+  context "when the card has ended" do
+    before { card.update!(ended_date: Date.new(2026, 1, 20)) }
+
+    it "recovers the last arrears segment and clears the exhausted clock" do
+      segment = result.billing_segments.sole
+
+      expect(segment.ended_at).to eq(BillingSegment.inclusive_end(Time.utc(2026, 1, 21)))
+      expect(segment.billing_at).to eq(Time.utc(2026, 1, 21))
+      expect(segment.proration_ratio).to be_within(1e-10).of(6.fdiv(31))
+      expect(card.reload.next_billing_at).to be_nil
+      expect(described_class.call(customer:, timestamp: Time.utc(2026, 3, 1)).billing_segments).to eq([])
+    end
+  end
+
+  context "when the contract has terminated" do
+    before { contract.update!(status: :terminated, ended_at: Time.utc(2026, 1, 20, 12)) }
+
+    it "keeps the exact service end and the calendar's consumed days" do
+      segment = result.billing_segments.sole
+
+      expect(segment.ended_at).to eq(BillingSegment.inclusive_end(contract.ended_at))
+      expect(segment.proration_ratio).to be_within(1e-10).of(6.fdiv(31))
+      expect(card.reload.next_billing_at).to be_nil
+    end
+  end
+
+  it "recovers the final arrears segment when termination precedes the saved clock" do
+    described_class.call!(customer:, timestamp: Time.utc(2026, 1, 20))
+    expect(card.reload.next_billing_at).to eq(Time.utc(2026, 2, 1))
+    contract.update!(status: :terminated, ended_at: Time.utc(2026, 1, 25, 12))
+
+    segment = result.billing_segments.sole
+
+    expect(segment.billing_at).to eq(contract.ended_at)
+    expect(segment.ended_at).to eq(BillingSegment.inclusive_end(contract.ended_at))
+    expect(card.reload.next_billing_at).to be_nil
+  end
+
+  it "recovers an inclusive card end that precedes the saved clock" do
+    described_class.call!(customer:, timestamp: Time.utc(2026, 1, 20))
+    card.update!(ended_date: Date.new(2026, 1, 25))
+
+    segment = result.billing_segments.sole
+
+    expect(segment.billing_at).to eq(Time.utc(2026, 1, 26))
+    expect(segment.proration_ratio).to be_within(1e-10).of(11.fdiv(31))
+    expect(card.reload.next_billing_at).to be_nil
+  end
+
+  context "when the period crosses daylight saving time" do
+    let(:customer) { create(:customer, organization:, timezone: "Europe/Paris") }
+    let(:timestamp) { Time.utc(2026, 3, 31, 22) }
+
+    before do
+      card.update!(effective_date: Date.new(2026, 3, 1), billing_anchor_date: Date.new(2026, 3, 1),
+        next_billing_at: Time.utc(2026, 2, 28, 23))
+    end
+
+    it "stores local-month boundaries without changing the number of billable days" do
+      segment = result.billing_segments.sole
+
+      expect(segment.started_at).to eq(Time.utc(2026, 2, 28, 23))
+      expect(segment.ended_at).to eq(BillingSegment.inclusive_end(timestamp))
+      expect(segment.duration_in_days).to eq(31)
+      expect(segment.proration_ratio).to eq(1)
+      expect(card.reload.next_billing_at).to eq(Time.utc(2026, 4, 30, 22))
+    end
+  end
+
+  it "advances to the first billing date when pricing has not started yet" do
+    rate.update!(effective_from: Time.utc(2026, 3, 1))
+
+    expect(result).to be_success
+    expect(result.billing_segments).to eq([])
+    expect(card.reload.next_billing_at).to eq(Time.utc(2026, 4, 1))
+  end
+
+  it "rolls back earlier segments and clocks when another card cannot be scheduled" do
+    create(:contract_rate_card, id: "ffffffff-ffff-ffff-ffff-ffffffffffff", organization:, contract:,
+      effective_date: card.effective_date, billing_anchor_date: card.billing_anchor_date,
+      next_billing_at: card.next_billing_at)
+    original_clock = card.next_billing_at
+
+    expect(result).not_to be_success
+    expect(result.error).to be_a(BaseService::NotFoundFailure)
+    expect(result.error.error_code).to eq("rate_not_found")
+    expect(result.billing_segments).to eq([])
+    expect(card.billing_segments.count).to eq(0)
+    expect(card.reload.next_billing_at).to eq(original_clock)
+  end
+end
diff --git a/spec/services/customers/lock_service_spec.rb b/spec/services/customers/lock_service_spec.rb
index 065d46852e0..ab6cf02a10b 100644
--- a/spec/services/customers/lock_service_spec.rb
+++ b/spec/services/customers/lock_service_spec.rb
@@ -18,7 +18,7 @@
   end
 
   describe "lock scoping" do
-    %i[prepaid_credit payment_method credit_note coupon].each do |scope|
+    %i[prepaid_credit payment_method credit_note coupon billing_schedule].each do |scope|
       context "with the #{scope} scope", transaction: false do
         subject(:lock_service) { described_class.new(customer:, scope:, timeout_seconds: 0.seconds) }