diff --git a/app/controllers/api/v1/subscriptions_controller.rb b/app/controllers/api/v1/subscriptions_controller.rb
index 6a4d06239d1d..f1b6619d9db0 100644
--- a/app/controllers/api/v1/subscriptions_controller.rb
+++ b/app/controllers/api/v1/subscriptions_controller.rb
@@ -180,6 +180,12 @@ def create_params
               :payment_method_type,
               :payment_method_id
             ],
+            connections: [
+              payment: [:behavior, :code],
+              tax: [:behavior, :code],
+              accounting: [:behavior, :code],
+              crm: [:behavior, :code]
+            ],
             usage_thresholds: usage_thresholds_params,
             plan_overrides:
           )
@@ -206,6 +212,12 @@ def update_params
             :payment_method_type,
             :payment_method_id
           ],
+          connections: [
+            payment: [:behavior, :code],
+            tax: [:behavior, :code],
+            accounting: [:behavior, :code],
+            crm: [:behavior, :code]
+          ],
           usage_thresholds: usage_thresholds_params,
           plan_overrides:
         )
diff --git a/app/controllers/concerns/wallet_actions.rb b/app/controllers/concerns/wallet_actions.rb
index a92f76b1db8c..b502e55ba228 100644
--- a/app/controllers/concerns/wallet_actions.rb
+++ b/app/controllers/concerns/wallet_actions.rb
@@ -140,6 +140,12 @@ def input_params
         payment_method: [
           :payment_method_type,
           :payment_method_id
+        ],
+        connections: [
+          payment: [:behavior, :code],
+          tax: [:behavior, :code],
+          accounting: [:behavior, :code],
+          crm: [:behavior, :code]
         ]
       ],
       applies_to: [
@@ -153,6 +159,12 @@ def input_params
       payment_method: [
         :payment_method_type,
         :payment_method_id
+      ],
+      connections: [
+        payment: [:behavior, :code],
+        tax: [:behavior, :code],
+        accounting: [:behavior, :code],
+        crm: [:behavior, :code]
       ]
     )
   end
@@ -196,6 +208,12 @@ def update_params
         payment_method: [
           :payment_method_type,
           :payment_method_id
+        ],
+        connections: [
+          payment: [:behavior, :code],
+          tax: [:behavior, :code],
+          accounting: [:behavior, :code],
+          crm: [:behavior, :code]
         ]
       ],
       applies_to: [
@@ -209,6 +227,12 @@ def update_params
       payment_method: [
         :payment_method_type,
         :payment_method_id
+      ],
+      connections: [
+        payment: [:behavior, :code],
+        tax: [:behavior, :code],
+        accounting: [:behavior, :code],
+        crm: [:behavior, :code]
       ]
     )
   end
diff --git a/app/services/billing_object_connections/attach_to_resource_service.rb b/app/services/billing_object_connections/attach_to_resource_service.rb
new file mode 100644
index 000000000000..b7b5604e0dca
--- /dev/null
+++ b/app/services/billing_object_connections/attach_to_resource_service.rb
@@ -0,0 +1,105 @@
+# frozen_string_literal: true
+
+module BillingObjectConnections
+  class AttachToResourceService < BaseService
+    Result = BaseResult[:billing_object_connections]
+
+    INHERIT_BEHAVIOR = "inherit"
+
+    def initialize(resource:, params:)
+      @resource = resource
+      @params = params
+      super
+    end
+
+    def call
+      return result unless params.key?(:connections)
+      return result if connections.blank?
+
+      ActiveRecord::Base.transaction do
+        connections.each do |category, choice|
+          next if choice.blank?
+
+          apply_choice(category.to_s, choice)
+        end
+      end
+
+      result.billing_object_connections = resource.billing_object_connections.reload
+      result
+    rescue ActiveRecord::RecordInvalid => e
+      result.record_validation_failure!(record: e.record)
+    rescue BaseService::FailedResult => e
+      # raise_if_error! unwinds the transaction on an unresolvable code; the failure is returned
+      # rather than propagated so `.call` keeps the BaseService contract.
+      e.result
+    end
+
+    private
+
+    attr_reader :resource, :params
+
+    def connections
+      params[:connections]
+    end
+
+    def customer
+      resource.customer
+    end
+
+    def apply_choice(category, choice)
+      behavior = choice[:behavior].to_s
+
+      if behavior == INHERIT_BEHAVIOR
+        destroy_override(category)
+      elsif behavior == BillingObjectConnection::BEHAVIORS[:skip]
+        upsert_override(category, behavior: :skip, connection: nil)
+      else
+        connection = resolve_connection(category, choice[:code])
+
+        if connection.nil?
+          result.single_validation_failure!(field: :connections, error_code: "connection_not_found")
+          result.raise_if_error!
+        end
+
+        upsert_override(category, behavior: :specific, connection:)
+      end
+    end
+
+    def destroy_override(category)
+      resource.billing_object_connections.find_by(category:)&.destroy!
+    end
+
+    def upsert_override(category, behavior:, connection:)
+      override = resource.billing_object_connections.find_or_initialize_by(category:)
+
+      override.organization_id = resource.organization_id
+      override.behavior = behavior
+      override.payment_provider_customer = nil
+      override.integration_customer = nil
+
+      if payment?(category)
+        override.payment_provider_customer = connection
+      else
+        override.integration_customer = connection
+      end
+
+      override.save!
+    end
+
+    # Both foreign keys are optional on the model and the category/column pairing lives only in
+    # ConnectionResolvable, so the mapping is mirrored here.
+    def resolve_connection(category, code)
+      return nil if code.blank? || customer.nil?
+
+      if payment?(category)
+        customer.payment_connection(code)
+      else
+        customer.integration_customers.find_by(category:, code:)
+      end
+    end
+
+    def payment?(category)
+      category == BillingObjectConnection::CATEGORIES[:payment]
+    end
+  end
+end
diff --git a/app/services/billing_object_connections/validate_service.rb b/app/services/billing_object_connections/validate_service.rb
new file mode 100644
index 000000000000..1048bb50ee5b
--- /dev/null
+++ b/app/services/billing_object_connections/validate_service.rb
@@ -0,0 +1,70 @@
+# frozen_string_literal: true
+
+module BillingObjectConnections
+  class ValidateService < BaseValidator
+    CATEGORIES = BillingObjectConnection::CATEGORIES.values.freeze
+    # "specific" is never sent: it is implied by supplying a code. "inherit" is params-only
+    # and means "destroy the override row", since row absence is what ConnectionResolvable
+    # reads as inheritance.
+    BEHAVIORS = %w[inherit skip].freeze
+
+    # Pure validator: it accumulates error codes and leaves surfacing them to the caller, because
+    # the wallet create path merges them into an accumulating validator while the update and
+    # recurring-rule paths fail the result directly.
+    def valid?
+      validate_connections if connections.present?
+
+      !errors?
+    end
+
+    def error_codes
+      errors[:connections] || []
+    end
+
+    private
+
+    def connections
+      args[:connections]
+    end
+
+    def validate_connections
+      unless connections.is_a?(Hash)
+        add_error(field: :connections, error_code: "invalid_connections")
+        return
+      end
+
+      connections.each do |category, choice|
+        validate_category(category)
+        validate_choice(choice)
+      end
+    end
+
+    def validate_category(category)
+      return true if CATEGORIES.include?(category.to_s)
+
+      add_error(field: :connections, error_code: "invalid_connection_category")
+    end
+
+    def validate_choice(choice)
+      unless choice.is_a?(Hash)
+        add_error(field: :connections, error_code: "invalid_connection_choice")
+        return
+      end
+
+      code = choice[:code]
+      behavior = choice[:behavior]
+
+      if code.present? && behavior.present?
+        return add_error(field: :connections, error_code: "invalid_connection_choice")
+      end
+
+      if code.blank? && behavior.blank?
+        return add_error(field: :connections, error_code: "invalid_connection_choice")
+      end
+
+      return true if behavior.blank? || BEHAVIORS.include?(behavior.to_s)
+
+      add_error(field: :connections, error_code: "invalid_connection_behavior")
+    end
+  end
+end
diff --git a/app/services/subscriptions/create_service.rb b/app/services/subscriptions/create_service.rb
index 25e58ea1353e..3752e3272544 100644
--- a/app/services/subscriptions/create_service.rb
+++ b/app/services/subscriptions/create_service.rb
@@ -28,12 +28,14 @@ def call
         subscription_at:,
         ending_at: params[:ending_at],
         payment_method: params[:payment_method],
+        connections: params[:connections],
         activation_rules: params[:activation_rules],
         subscription_type:,
         consolidate_invoice: params[:consolidate_invoice],
         consolidate_invoice_provided: params.key?(:consolidate_invoice)
       )
       return result.forbidden_failure! if !License.premium? && params.key?(:plan_overrides)
+      return result.forbidden_failure! if connections_requested? && organization_flag_disabled?(:multi_connection)
 
       if params.key?(:plan_overrides) && plan.organization.product_catalog_enabled?
         return result.single_validation_failure!(field: :plan_overrides, error_code: "legacy_billing_disabled")
@@ -83,6 +85,10 @@ def call
           end
           InvoiceCustomSections::AttachToResourceService.call(resource: subscription, params:) unless downgrade?
 
+          if connections_requested?
+            BillingObjectConnections::AttachToResourceService.call!(resource: subscription, params:)
+          end
+
           result.subscription = subscription
         end
       end
@@ -114,6 +120,14 @@ def valid?(args)
       Subscriptions::ValidateService.new(result, **args).valid?
     end
 
+    def connections_requested?
+      params[:connections].present?
+    end
+
+    def organization_flag_disabled?(flag)
+      !customer.organization.feature_flag_enabled?(flag)
+    end
+
     def handle_subscription
       return upgrade_subscription if upgrade?
       return downgrade_subscription if downgrade?
diff --git a/app/services/subscriptions/update_service.rb b/app/services/subscriptions/update_service.rb
index 2a38fdb605f3..3fb180116021 100644
--- a/app/services/subscriptions/update_service.rb
+++ b/app/services/subscriptions/update_service.rb
@@ -37,6 +37,7 @@ def call
         on_termination_credit_note: params[:on_termination_credit_note],
         on_termination_invoice: params[:on_termination_invoice],
         payment_method: params[:payment_method],
+        connections: params[:connections],
         activation_rules: params[:activation_rules],
         subscription_type: "update",
         subscription:,
@@ -55,6 +56,7 @@ def call
       end
 
       return result.forbidden_failure! if !License.premium? && params.key?(:plan_overrides)
+      return result.forbidden_failure! if connections_requested? && organization_flag_disabled?(:multi_connection)
 
       if params.key?(:plan_overrides) && subscription.plan.organization.product_catalog_enabled?
         return result.single_validation_failure!(field: :plan_overrides, error_code: "legacy_billing_disabled")
@@ -123,6 +125,10 @@ def call
         end
 
         InvoiceCustomSections::AttachToResourceService.call(resource: subscription, params:)
+
+        if connections_requested?
+          BillingObjectConnections::AttachToResourceService.call!(resource: subscription, params:)
+        end
       end
 
       result.subscription = subscription
@@ -287,6 +293,14 @@ def valid?(args)
       Subscriptions::ValidateService.new(result, **args).valid?
     end
 
+    def connections_requested?
+      params[:connections].present?
+    end
+
+    def organization_flag_disabled?(flag)
+      !subscription.organization.feature_flag_enabled?(flag)
+    end
+
     def payment_method
       return @payment_method if defined? @payment_method
       return nil if params[:payment_method].blank? || params[:payment_method][:payment_method_id].blank?
diff --git a/app/services/subscriptions/validate_service.rb b/app/services/subscriptions/validate_service.rb
index 7e707251bba0..e6032d11037e 100644
--- a/app/services/subscriptions/validate_service.rb
+++ b/app/services/subscriptions/validate_service.rb
@@ -11,6 +11,7 @@ def valid?
       valid_on_termination_credit_note?
       valid_on_termination_invoice?
       valid_payment_method?
+      valid_connections?
       valid_activation_rules?
       valid_consolidate_invoice?
 
@@ -101,6 +102,17 @@ def valid_payment_method?
       false
     end
 
+    def valid_connections?
+      return true if args[:connections].blank?
+
+      validator = BillingObjectConnections::ValidateService.new(result, connections: args[:connections])
+      return true if validator.valid?
+
+      validator.error_codes.each { |error_code| add_error(field: :connections, error_code:) }
+
+      false
+    end
+
     def valid_activation_rules?
       return true unless args[:activation_rules]
 
diff --git a/app/services/wallets/create_service.rb b/app/services/wallets/create_service.rb
index 39e2054b0253..1bbefd45dc10 100644
--- a/app/services/wallets/create_service.rb
+++ b/app/services/wallets/create_service.rb
@@ -20,6 +20,7 @@ def call
       result.payment_method = payment_method
 
       return result unless valid?
+      return result.forbidden_failure! if connections_requested? && organization_flag_disabled?(:multi_connection)
 
       code = params[:code]
 
@@ -88,6 +89,10 @@ def call
           InvoiceCustomSections::AttachToResourceService.call(resource: wallet, params:)
         end
 
+        if connections_requested?
+          BillingObjectConnections::AttachToResourceService.call!(resource: wallet, params:)
+        end
+
         billable_metrics.each do |bm|
           WalletTarget.create!(wallet:, billable_metric: bm, organization_id:)
         end
@@ -163,6 +168,16 @@ def organization_flag_enabled?(flag)
       customer.organization.feature_flag_enabled?(flag)
     end
 
+    def organization_flag_disabled?(flag)
+      !organization_flag_enabled?(flag)
+    end
+
+    def connections_requested?
+      return true if params[:connections].present?
+
+      Array(params[:recurring_transaction_rules]).any? { |rule| rule[:connections].present? }
+    end
+
     def valid?
       Wallets::ValidateService.new(result, **params).valid?
     end
diff --git a/app/services/wallets/recurring_transaction_rules/create_service.rb b/app/services/wallets/recurring_transaction_rules/create_service.rb
index 2ea5353e6e29..2d45b31b5e03 100644
--- a/app/services/wallets/recurring_transaction_rules/create_service.rb
+++ b/app/services/wallets/recurring_transaction_rules/create_service.rb
@@ -15,6 +15,7 @@ def initialize(wallet:, wallet_params:)
       def call
         return unless License.premium?
         return result unless valid_payment_method?
+        return result unless valid_connections?
 
         if method == "fixed" && rule_params[:paid_credits].nil? && rule_params[:granted_credits].nil?
           paid_credits = wallet_params[:paid_credits]
@@ -67,6 +68,10 @@ def call
           InvoiceCustomSections::AttachToResourceService.call(resource: rule, params: rule_params)
         end
 
+        if rule_params[:connections].present?
+          BillingObjectConnections::AttachToResourceService.call!(resource: rule, params: rule_params)
+        end
+
         result.recurring_transaction_rule = rule
         result
       rescue ActiveRecord::RecordInvalid => e
@@ -109,6 +114,15 @@ def valid_payment_method?
         PaymentMethods::ValidateService.new(result, **rule_params).valid?
       end
 
+      def valid_connections?
+        validator = BillingObjectConnections::ValidateService.new(result, **rule_params)
+        return true if validator.valid?
+
+        result.single_validation_failure!(field: :connections, error_code: validator.error_codes.first)
+
+        false
+      end
+
       def payment_method
         return @payment_method if defined? @payment_method
         return nil if rule_params[:payment_method].blank? || rule_params[:payment_method][:payment_method_id].blank?
diff --git a/app/services/wallets/recurring_transaction_rules/update_service.rb b/app/services/wallets/recurring_transaction_rules/update_service.rb
index 67305fc2911c..75a084f37861 100644
--- a/app/services/wallets/recurring_transaction_rules/update_service.rb
+++ b/app/services/wallets/recurring_transaction_rules/update_service.rb
@@ -14,6 +14,7 @@ def initialize(wallet:, params:)
 
       def call
         return result unless valid_payment_methods?
+        return result unless valid_connections?
 
         created_recurring_rules_ids = []
 
@@ -33,6 +34,8 @@ def call
             rule_attributes.delete(:payment_method)
           end
 
+          connections = rule_attributes.delete(:connections)
+
           recurring_rule = wallet.recurring_transaction_rules.active.find_by(id: lago_id)
 
           normalize_grants_target_top_up!(rule_attributes, recurring_rule)
@@ -52,6 +55,8 @@ def call
             end
 
             recurring_rule.update!(rule_attributes)
+
+            attach_connections(recurring_rule, connections)
           else
             unless rule_attributes.key?(:invoice_requires_successful_payment)
               rule_attributes[:invoice_requires_successful_payment] = wallet.invoice_requires_successful_payment
@@ -68,6 +73,8 @@ def call
               )
             end
 
+            attach_connections(created_recurring_rule, connections)
+
             created_recurring_rules_ids.push(created_recurring_rule.id)
           end
         end
@@ -130,6 +137,28 @@ def valid_payment_methods?
         true
       end
 
+      def attach_connections(recurring_rule, connections)
+        return if connections.blank?
+
+        BillingObjectConnections::AttachToResourceService.call!(
+          resource: recurring_rule,
+          params: {connections:}
+        )
+      end
+
+      def valid_connections?
+        hash_recurring_rules.each do |payload_rule|
+          validator = BillingObjectConnections::ValidateService.new(result, **payload_rule)
+          next if validator.valid?
+
+          result.single_validation_failure!(field: :connections, error_code: validator.error_codes.first)
+
+          return false
+        end
+
+        true
+      end
+
       def payment_method(rule_params)
         return nil if rule_params[:payment_method].blank? || rule_params[:payment_method][:payment_method_id].blank?
 
diff --git a/app/services/wallets/update_service.rb b/app/services/wallets/update_service.rb
index 5d888a3747cf..c6cbfa8a4136 100644
--- a/app/services/wallets/update_service.rb
+++ b/app/services/wallets/update_service.rb
@@ -24,6 +24,8 @@ def call
       return result unless valid_recurring_transaction_rules?
       return result unless valid_limitations?
       return result unless valid_payment_method?
+      return result unless valid_connections?
+      return result.forbidden_failure! if connections_requested? && organization_flag_disabled?(:multi_connection)
 
       if billing_entity_param_sent?
         if billing_entity_value_provided? && billing_entity.nil?
@@ -78,6 +80,8 @@ def call
         end
 
         InvoiceCustomSections::AttachToResourceService.call!(resource: wallet, params:)
+
+        BillingObjectConnections::AttachToResourceService.call!(resource: wallet, params:) if connections_requested?
         SendWebhookJob.perform_after_commit("wallet.updated", wallet)
       end
 
@@ -134,6 +138,15 @@ def valid_payment_method?
       PaymentMethods::ValidateService.new(result, **params).valid?
     end
 
+    def valid_connections?
+      validator = BillingObjectConnections::ValidateService.new(result, **params)
+      return true if validator.valid?
+
+      result.validation_failure!(errors: {connections: validator.error_codes})
+
+      false
+    end
+
     def process_billable_metrics
       # In case of adding new type of limitation in wallet_targets, query from below should use compact to avoid nil values in the array
       existing_wallet_billable_metric_ids = wallet.wallet_targets.pluck(:billable_metric_id)
@@ -203,6 +216,16 @@ def organization_flag_enabled?(flag)
       wallet.customer.organization.feature_flag_enabled?(flag)
     end
 
+    def organization_flag_disabled?(flag)
+      !organization_flag_enabled?(flag)
+    end
+
+    def connections_requested?
+      return true if params[:connections].present?
+
+      Array(params[:recurring_transaction_rules]).any? { |rule| rule[:connections].present? }
+    end
+
     def billing_entity_param_sent?
       params.key?(:billing_entity_id) || params.key?(:billing_entity_code)
     end
diff --git a/app/services/wallets/validate_service.rb b/app/services/wallets/validate_service.rb
index da9db35fcd1b..66b3ca964689 100644
--- a/app/services/wallets/validate_service.rb
+++ b/app/services/wallets/validate_service.rb
@@ -15,6 +15,7 @@ def valid?
       valid_limitations? if args[:applies_to]
       valid_wallet_limit?
       valid_payment_method? if args[:payment_method]
+      valid_connections? if args[:connections].present?
 
       if errors?
         result.validation_failure!(errors:)
@@ -117,5 +118,14 @@ def valid_payment_method?
 
       add_error(field: :payment_method, error_code: "invalid_payment_method")
     end
+
+    def valid_connections?
+      validator = BillingObjectConnections::ValidateService.new(result, **args)
+      return true if validator.valid?
+
+      validator.error_codes.each { |error_code| add_error(field: :connections, error_code:) }
+
+      false
+    end
   end
 end
diff --git a/spec/requests/api/v1/subscriptions_controller_spec.rb b/spec/requests/api/v1/subscriptions_controller_spec.rb
index 0b4576ef4d2c..c8e8d3236c7f 100644
--- a/spec/requests/api/v1/subscriptions_controller_spec.rb
+++ b/spec/requests/api/v1/subscriptions_controller_spec.rb
@@ -866,6 +866,88 @@
         end
       end
     end
+
+    context "with connections" do
+      let(:stripe_connection) { create(:stripe_customer, customer:, code: "stripe_us") }
+      let(:netsuite_connection) { create(:netsuite_customer, customer:, code: "netsuite_main") }
+      let(:params) do
+        {
+          external_customer_id: customer.external_id,
+          plan_code: plan.code,
+          external_id: SecureRandom.uuid,
+          connections: {
+            payment: {code: "stripe_us"},
+            tax: {behavior: "skip"},
+            accounting: {code: "netsuite_main"},
+            crm: {behavior: "skip"}
+          }
+        }
+      end
+
+      before do
+        organization.enable_feature_flag!(:multi_connection)
+        stripe_connection
+        netsuite_connection
+      end
+
+      it "persists one connection per category" do
+        expect { subject }.to change(BillingObjectConnection, :count).by(4)
+
+        expect(response).to have_http_status(:success)
+
+        subscription = Subscription.find(json[:subscription][:lago_id])
+        expect(subscription.billing_object_connections.pluck(:category)).to match_array(%w[payment tax accounting crm])
+        expect(subscription.effective_payment_connection).to eq(stripe_connection)
+        expect(subscription.effective_accounting_connection).to eq(netsuite_connection)
+        expect(subscription.effective_tax_connection).to be_nil
+      end
+
+      context "when a code does not resolve" do
+        let(:params) do
+          {
+            external_customer_id: customer.external_id,
+            plan_code: plan.code,
+            external_id: SecureRandom.uuid,
+            connections: {payment: {code: "unknown_connection"}}
+          }
+        end
+
+        it "returns a validation error" do
+          subject
+
+          expect(response).to have_http_status(:unprocessable_content)
+          expect(json[:error_details][:connections]).to include("connection_not_found")
+        end
+      end
+
+      context "when the behavior is invalid" do
+        let(:params) do
+          {
+            external_customer_id: customer.external_id,
+            plan_code: plan.code,
+            external_id: SecureRandom.uuid,
+            connections: {payment: {behavior: "nonsense"}}
+          }
+        end
+
+        it "returns a validation error" do
+          subject
+
+          expect(response).to have_http_status(:unprocessable_content)
+          expect(json[:error_details][:connections]).to include("invalid_connection_behavior")
+        end
+      end
+
+      context "when the multi_connection flag is disabled" do
+        before { organization.disable_feature_flag!(:multi_connection) }
+
+        it "returns a forbidden error" do
+          subject
+
+          expect(response).to have_http_status(:forbidden)
+        end
+      end
+    end
   end
 
   describe "DELETE /api/v1/subscriptions/:external_id" do
@@ -1913,6 +1995,48 @@ def test_termination(expected_on_termination_credit_note: nil, expected_on_termi
         end
       end
     end
+
+    context "with connections" do
+      let(:stripe_connection) { create(:stripe_customer, customer:, code: "stripe_us") }
+      let(:update_params) { {connections: {payment: {code: "stripe_us"}}} }
+
+      before do
+        organization.enable_feature_flag!(:multi_connection)
+        stripe_connection
+      end
+
+      it "pins the connection on the subscription" do
+        expect { subject }.to change(BillingObjectConnection, :count).by(1)
+
+        expect(response).to have_http_status(:success)
+        expect(subscription.reload.effective_payment_connection).to eq(stripe_connection)
+      end
+
+      context "when inherit is sent for a category that has an override" do
+        let(:update_params) { {connections: {payment: {behavior: "inherit"}}} }
+
+        before do
+          create(:billing_object_connection, owner: subscription, organization:, category: "payment", behavior: "skip")
+        end
+
+        it "clears the override so resolution falls back to the customer" do
+          expect { subject }.to change(BillingObjectConnection, :count).by(-1)
+
+          expect(response).to have_http_status(:success)
+          expect(subscription.reload.billing_object_connections).to be_empty
+        end
+      end
+
+      context "when the multi_connection flag is disabled" do
+        before { organization.disable_feature_flag!(:multi_connection) }
+
+        it "returns a forbidden error" do
+          subject
+
+          expect(response).to have_http_status(:forbidden)
+        end
+      end
+    end
   end
 
   describe "GET /api/v1/subscriptions/:external_id" do
diff --git a/spec/services/billing_object_connections/attach_to_resource_service_spec.rb b/spec/services/billing_object_connections/attach_to_resource_service_spec.rb
new file mode 100644
index 000000000000..929bd35ba17e
--- /dev/null
+++ b/spec/services/billing_object_connections/attach_to_resource_service_spec.rb
@@ -0,0 +1,227 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+RSpec.describe BillingObjectConnections::AttachToResourceService do
+  subject(:result) { described_class.call(resource:, params:) }
+
+  let(:organization) { create(:organization) }
+  let(:customer) { create(:customer, organization:) }
+  let(:resource) { create(:wallet, customer:, organization:) }
+
+  let(:stripe_connection) { create(:stripe_customer, customer:, code: "stripe_us") }
+  let(:netsuite_connection) { create(:netsuite_customer, customer:, code: "netsuite_main") }
+
+  describe "#call" do
+    context "when the connections key is absent" do
+      let(:params) { {name: "whatever"} }
+
+      it "does not touch any row" do
+        expect { result }.not_to change(BillingObjectConnection, :count)
+        expect(result).to be_success
+      end
+    end
+
+    context "when connections is empty" do
+      let(:params) { {connections: {}} }
+
+      it "does not touch any row" do
+        expect { result }.not_to change(BillingObjectConnection, :count)
+        expect(result).to be_success
+      end
+    end
+
+    context "with a specific payment connection" do
+      let(:params) { {connections: {payment: {code: "stripe_us"}}} }
+
+      before { stripe_connection }
+
+      it "pins the payment provider customer" do
+        expect { result }.to change(BillingObjectConnection, :count).by(1)
+
+        override = resource.billing_object_connections.sole
+        expect(override).to have_attributes(
+          category: "payment",
+          behavior: "specific",
+          organization_id: organization.id,
+          payment_provider_customer_id: stripe_connection.id,
+          integration_customer_id: nil
+        )
+      end
+    end
+
+    context "with a specific accounting connection" do
+      let(:params) { {connections: {accounting: {code: "netsuite_main"}}} }
+
+      before { netsuite_connection }
+
+      it "pins the integration customer" do
+        result
+
+        override = resource.billing_object_connections.sole
+        expect(override).to have_attributes(
+          category: "accounting",
+          behavior: "specific",
+          integration_customer_id: netsuite_connection.id,
+          payment_provider_customer_id: nil
+        )
+      end
+    end
+
+    context "with skip" do
+      let(:params) { {connections: {tax: {behavior: "skip"}}} }
+
+      it "stores a skip row with no connection attached" do
+        result
+
+        override = resource.billing_object_connections.sole
+        expect(override).to have_attributes(
+          category: "tax",
+          behavior: "skip",
+          payment_provider_customer_id: nil,
+          integration_customer_id: nil
+        )
+      end
+    end
+
+    context "with inherit" do
+      let(:params) { {connections: {tax: {behavior: "inherit"}}} }
+
+      context "when an override exists" do
+        before { create(:billing_object_connection, owner: resource, organization:, category: "tax", behavior: "skip") }
+
+        it "destroys the override so resolution falls back to the customer" do
+          expect { result }.to change(BillingObjectConnection, :count).by(-1)
+          expect(resource.billing_object_connections.reload).to be_empty
+        end
+      end
+
+      context "when no override exists" do
+        it "is a no-op" do
+          expect { result }.not_to change(BillingObjectConnection, :count)
+          expect(result).to be_success
+        end
+      end
+    end
+
+    context "when a category is omitted" do
+      let(:params) { {connections: {payment: {behavior: "skip"}}} }
+
+      before { create(:billing_object_connection, owner: resource, organization:, category: "tax", behavior: "skip") }
+
+      it "leaves the existing row for that category untouched" do
+        expect { result }.to change(BillingObjectConnection, :count).by(1)
+        expect(resource.billing_object_connections.reload.pluck(:category)).to match_array(%w[payment tax])
+      end
+    end
+
+    context "when the same category is written twice" do
+      let(:params) { {connections: {payment: {code: "stripe_us"}}} }
+
+      before do
+        stripe_connection
+        create(:billing_object_connection, owner: resource, organization:, category: "payment", behavior: "skip")
+      end
+
+      it "updates the existing row rather than duplicating it" do
+        expect { result }.not_to change(BillingObjectConnection, :count)
+
+        expect(resource.billing_object_connections.sole).to have_attributes(
+          behavior: "specific",
+          payment_provider_customer_id: stripe_connection.id
+        )
+      end
+    end
+
+    context "when the code does not resolve" do
+      let(:params) { {connections: {payment: {code: "does_not_exist"}}} }
+
+      it "fails with connection_not_found" do
+        expect(result).not_to be_success
+        expect(result.error.messages[:connections]).to include("connection_not_found")
+      end
+
+      it "does not persist anything" do
+        expect { result }.not_to change(BillingObjectConnection, :count)
+      end
+    end
+
+    context "when the code belongs to another customer" do
+      let(:other_customer) { create(:customer, organization:) }
+      let(:params) { {connections: {payment: {code: "stripe_other"}}} }
+
+      before { create(:stripe_customer, customer: other_customer, code: "stripe_other") }
+
+      it "fails with connection_not_found" do
+        expect(result).not_to be_success
+        expect(result.error.messages[:connections]).to include("connection_not_found")
+      end
+    end
+
+    context "when the code exists but in another category" do
+      let(:params) { {connections: {crm: {code: "netsuite_main"}}} }
+
+      before { netsuite_connection }
+
+      it "fails with connection_not_found" do
+        expect(result).not_to be_success
+        expect(result.error.messages[:connections]).to include("connection_not_found")
+      end
+    end
+
+    context "when one category resolves and a later one does not" do
+      let(:params) do
+        {connections: {payment: {code: "stripe_us"}, accounting: {code: "nope"}}}
+      end
+
+      before { stripe_connection }
+
+      it "rolls back the rows written before the failure" do
+        expect { result }.not_to change(BillingObjectConnection, :count)
+        expect(result).not_to be_success
+      end
+    end
+
+    context "when the resource is a recurring transaction rule" do
+      let(:wallet) { create(:wallet, customer:, organization:) }
+      let(:resource) { create(:recurring_transaction_rule, wallet:, organization:) }
+      let(:params) { {connections: {payment: {code: "stripe_us"}}} }
+
+      before { stripe_connection }
+
+      it "owns the connection by the rule" do
+        result
+
+        override = resource.billing_object_connections.sole
+        expect(override).to have_attributes(
+          owner_id: resource.id,
+          owner_type: "RecurringTransactionRule",
+          payment_provider_customer_id: stripe_connection.id
+        )
+      end
+    end
+
+    context "with every category at once" do
+      let(:params) do
+        {
+          connections: {
+            payment: {code: "stripe_us"},
+            tax: {behavior: "skip"},
+            accounting: {code: "netsuite_main"},
+            crm: {behavior: "skip"}
+          }
+        }
+      end
+
+      before do
+        stripe_connection
+        netsuite_connection
+      end
+
+      it "writes one row per category" do
+        expect { result }.to change(BillingObjectConnection, :count).by(4)
+        expect(result.billing_object_connections.pluck(:category)).to match_array(%w[payment tax accounting crm])
+      end
+    end
+  end
+end
diff --git a/spec/services/billing_object_connections/validate_service_spec.rb b/spec/services/billing_object_connections/validate_service_spec.rb
new file mode 100644
index 000000000000..f2f866b53381
--- /dev/null
+++ b/spec/services/billing_object_connections/validate_service_spec.rb
@@ -0,0 +1,125 @@
+# frozen_string_literal: true
+
+require "rails_helper"
+
+RSpec.describe BillingObjectConnections::ValidateService do
+  subject(:validator) { described_class.new(result, connections:) }
+
+  let(:result) { BaseResult.new }
+
+  describe "#valid?" do
+    context "when connections is absent" do
+      let(:connections) { nil }
+
+      it "is valid" do
+        expect(validator).to be_valid
+      end
+    end
+
+    context "when connections is empty" do
+      let(:connections) { {} }
+
+      it "is valid" do
+        expect(validator).to be_valid
+      end
+    end
+
+    context "with a code" do
+      let(:connections) { {payment: {code: "stripe_us"}} }
+
+      it "is valid" do
+        expect(validator).to be_valid
+      end
+    end
+
+    context "with each supported behavior" do
+      %w[inherit skip].each do |behavior|
+        context "when behavior is #{behavior}" do
+          let(:connections) { {tax: {behavior:}} }
+
+          it "is valid" do
+            expect(validator).to be_valid
+          end
+        end
+      end
+    end
+
+    context "with every category" do
+      let(:connections) do
+        {
+          payment: {behavior: "skip"},
+          tax: {behavior: "skip"},
+          accounting: {code: "netsuite_main"},
+          crm: {behavior: "inherit"}
+        }
+      end
+
+      it "is valid" do
+        expect(validator).to be_valid
+      end
+    end
+
+    context "when the category is unknown" do
+      let(:connections) { {shipping: {behavior: "skip"}} }
+
+      it "returns an invalid_connection_category error" do
+        expect(validator).not_to be_valid
+        expect(validator.error_codes).to include("invalid_connection_category")
+      end
+    end
+
+    context "when the behavior is unknown" do
+      let(:connections) { {payment: {behavior: "whatever"}} }
+
+      it "returns an invalid_connection_behavior error" do
+        expect(validator).not_to be_valid
+        expect(validator.error_codes).to include("invalid_connection_behavior")
+      end
+    end
+
+    context "when behavior is specific" do
+      let(:connections) { {payment: {behavior: "specific"}} }
+
+      it "is rejected, since specific is implied by supplying a code" do
+        expect(validator).not_to be_valid
+        expect(validator.error_codes).to include("invalid_connection_behavior")
+      end
+    end
+
+    context "when both a code and a behavior are given" do
+      let(:connections) { {payment: {code: "stripe_us", behavior: "skip"}} }
+
+      it "returns an invalid_connection_choice error" do
+        expect(validator).not_to be_valid
+        expect(validator.error_codes).to include("invalid_connection_choice")
+      end
+    end
+
+    context "when neither a code nor a behavior is given" do
+      let(:connections) { {payment: {}} }
+
+      it "returns an invalid_connection_choice error" do
+        expect(validator).not_to be_valid
+        expect(validator.error_codes).to include("invalid_connection_choice")
+      end
+    end
+
+    context "when the choice is not an object" do
+      let(:connections) { {payment: "stripe_us"} }
+
+      it "returns an invalid_connection_choice error" do
+        expect(validator).not_to be_valid
+        expect(validator.error_codes).to include("invalid_connection_choice")
+      end
+    end
+
+    context "when connections is not an object" do
+      let(:connections) { ["payment"] }
+
+      it "returns an invalid_connections error" do
+        expect(validator).not_to be_valid
+        expect(validator.error_codes).to eq(["invalid_connections"])
+      end
+    end
+  end
+end
diff --git a/spec/services/subscriptions/create_service_spec.rb b/spec/services/subscriptions/create_service_spec.rb
index 6ad492fa6bf0..2589bf30b3fb 100644
--- a/spec/services/subscriptions/create_service_spec.rb
+++ b/spec/services/subscriptions/create_service_spec.rb
@@ -2219,5 +2219,90 @@
         end
       end
     end
+
+    context "with connections" do
+      let(:stripe_connection) { create(:stripe_customer, customer:, code: "stripe_us") }
+      let(:params) do
+        {
+          external_customer_id:,
+          plan_code:,
+          name:,
+          external_id:,
+          billing_time:,
+          connections: {payment: {code: "stripe_us"}}
+        }
+      end
+
+      before do
+        organization.enable_feature_flag!(:multi_connection)
+        stripe_connection
+      end
+
+      it "pins the connection on the created subscription" do
+        result = create_service.call
+
+        expect(result).to be_success
+        expect(result.subscription.billing_object_connections.sole).to have_attributes(
+          category: "payment",
+          behavior: "specific",
+          payment_provider_customer_id: stripe_connection.id
+        )
+      end
+
+      context "when the code does not resolve" do
+        let(:params) { super().merge(connections: {payment: {code: "nope"}}) }
+
+        it "fails with connection_not_found and creates no subscription" do
+          expect { create_service.call }.not_to change(Subscription, :count)
+          expect(create_service.call.error.messages[:connections]).to include("connection_not_found")
+        end
+      end
+
+      context "when the multi_connection flag is disabled" do
+        before { organization.disable_feature_flag!(:multi_connection) }
+
+        it "returns a forbidden failure rather than dropping the routing choice" do
+          result = create_service.call
+
+          expect(result).not_to be_success
+          expect(result.error).to be_a(BaseService::ForbiddenFailure)
+        end
+      end
+
+      context "when upgrading an existing subscription" do
+        let(:old_plan) { create(:plan, amount_cents: 50, organization:, amount_currency: "EUR") }
+        let!(:current_subscription) do
+          create(:subscription, customer:, organization:, plan: old_plan, external_id:)
+        end
+
+        it "pins the connection on the new subscription, not the terminated one" do
+          result = create_service.call
+
+          expect(result).to be_success
+          expect(result.subscription.id).not_to eq(current_subscription.id)
+          expect(result.subscription.billing_object_connections.sole.payment_provider_customer_id)
+            .to eq(stripe_connection.id)
+          expect(current_subscription.reload.billing_object_connections).to be_empty
+        end
+      end
+
+      context "when downgrading an existing subscription" do
+        let(:plan) { create(:plan, amount_cents: 50, organization:, amount_currency: "EUR") }
+        let(:old_plan) { create(:plan, amount_cents: 500, organization:, amount_currency: "EUR") }
+
+        before { create(:subscription, customer:, organization:, plan: old_plan, external_id:) }
+
+        # Deliberately not skipped the way invoice_custom_section is on a downgrade: connections are
+        # routing config in the same family as payment_method, which the downgrade path already
+        # carries onto the pending subscription.
+        it "pins the connection on the pending subscription" do
+          result = create_service.call
+
+          expect(result).to be_success
+          expect(result.subscription.billing_object_connections.sole.payment_provider_customer_id)
+            .to eq(stripe_connection.id)
+        end
+      end
+    end
   end
 end
diff --git a/spec/services/wallets/recurring_transaction_rules/update_service_spec.rb b/spec/services/wallets/recurring_transaction_rules/update_service_spec.rb
index 5c325a03652d..0aa52b39f09b 100644
--- a/spec/services/wallets/recurring_transaction_rules/update_service_spec.rb
+++ b/spec/services/wallets/recurring_transaction_rules/update_service_spec.rb
@@ -483,5 +483,140 @@
         end
       end
     end
+
+    context "with connections" do
+      let(:customer) { create(:customer) }
+      let(:wallet) { create(:wallet, customer:, organization: customer.organization) }
+      let(:stripe_connection) { create(:stripe_customer, customer:, code: "stripe_us") }
+
+      before { stripe_connection }
+
+      context "when the rule is matched by lago_id" do
+        let(:params) do
+          [
+            {
+              lago_id: recurring_transaction_rule.id,
+              trigger: "interval",
+              interval: "weekly",
+              connections: {payment: {code: "stripe_us"}}
+            }
+          ]
+        end
+
+        it "updates the rule in place and pins the connection to it" do
+          rule = result.wallet.reload.recurring_transaction_rules.active.sole
+
+          expect(rule.id).to eq(recurring_transaction_rule.id)
+          expect(rule.billing_object_connections.sole).to have_attributes(
+            category: "payment",
+            behavior: "specific",
+            payment_provider_customer_id: stripe_connection.id
+          )
+        end
+
+        it "does not leak the connections key into the rule attributes" do
+          expect(result).to be_success
+        end
+
+        context "when the rule already has an override for that category" do
+          before do
+            create(
+              :billing_object_connection,
+              owner: recurring_transaction_rule,
+              organization: wallet.organization,
+              category: "payment",
+              behavior: "skip"
+            )
+          end
+
+          it "keeps the surviving rule's row and updates it rather than duplicating" do
+            expect { result }.not_to change(BillingObjectConnection, :count)
+
+            rule = result.wallet.reload.recurring_transaction_rules.active.sole
+            expect(rule.billing_object_connections.sole).to have_attributes(
+              behavior: "specific",
+              payment_provider_customer_id: stripe_connection.id
+            )
+          end
+        end
+      end
+
+      context "when the rule is replaced because no lago_id was sent" do
+        let(:params) do
+          [
+            {
+              trigger: "interval",
+              interval: "weekly",
+              connections: {payment: {code: "stripe_us"}}
+            }
+          ]
+        end
+
+        before do
+          create(
+            :billing_object_connection,
+            owner: recurring_transaction_rule,
+            organization: wallet.organization,
+            category: "payment",
+            behavior: "skip"
+          )
+        end
+
+        it "writes fresh rows for the new rule and leaves the terminated rule's rows alone" do
+          expect { result }.to change(BillingObjectConnection, :count).by(1)
+
+          wallet = result.wallet.reload
+          new_rule = wallet.recurring_transaction_rules.active.sole
+          expect(new_rule.id).not_to eq(recurring_transaction_rule.id)
+
+          expect(new_rule.billing_object_connections.sole).to have_attributes(
+            behavior: "specific",
+            payment_provider_customer_id: stripe_connection.id
+          )
+
+          # The superseded rule is soft-terminated, never destroyed, so dependent: :destroy does
+          # not fire and its override survives on the dead rule.
+          expect(recurring_transaction_rule.reload).to be_terminated
+          expect(recurring_transaction_rule.billing_object_connections.sole.behavior).to eq("skip")
+        end
+      end
+
+      context "when a code does not resolve" do
+        let(:params) do
+          [
+            {
+              lago_id: recurring_transaction_rule.id,
+              trigger: "interval",
+              interval: "weekly",
+              connections: {payment: {code: "nope"}}
+            }
+          ]
+        end
+
+        it "fails with connection_not_found" do
+          expect(result).not_to be_success
+          expect(result.error.messages[:connections]).to include("connection_not_found")
+        end
+      end
+
+      context "when the behavior is invalid" do
+        let(:params) do
+          [
+            {
+              lago_id: recurring_transaction_rule.id,
+              trigger: "interval",
+              interval: "weekly",
+              connections: {payment: {behavior: "nonsense"}}
+            }
+          ]
+        end
+
+        it "fails before writing anything" do
+          expect { result }.not_to change(BillingObjectConnection, :count)
+          expect(result).not_to be_success
+          expect(result.error.messages[:connections]).to include("invalid_connection_behavior")
+        end
+      end
+    end
   end
 end
diff --git a/spec/support/shared_examples/wallet_actions.rb b/spec/support/shared_examples/wallet_actions.rb
index e153631afbeb..2c1b1893b497 100644
--- a/spec/support/shared_examples/wallet_actions.rb
+++ b/spec/support/shared_examples/wallet_actions.rb
@@ -719,6 +719,139 @@
       end
     end
   end
+
+  context "with connections" do
+    let(:stripe_connection) { create(:stripe_customer, customer:, code: "stripe_us") }
+    let(:netsuite_connection) { create(:netsuite_customer, customer:, code: "netsuite_main") }
+    let(:create_params) do
+      {
+        external_customer_id: customer.external_id,
+        rate_amount: "1",
+        name: "Wallet1",
+        currency: "EUR",
+        paid_credits: "10",
+        granted_credits: "10",
+        connections: {
+          payment: {code: "stripe_us"},
+          tax: {behavior: "skip"},
+          accounting: {code: "netsuite_main"},
+          crm: {behavior: "skip"}
+        }
+      }
+    end
+
+    before do
+      organization.enable_feature_flag!(:multi_connection)
+      stripe_connection
+      netsuite_connection
+    end
+
+    it "persists one connection per category" do
+      expect { subject }.to change(BillingObjectConnection, :count).by(4)
+
+      expect(response).to have_http_status(:success)
+
+      wallet = Wallet.find(json[:wallet][:lago_id])
+      expect(wallet.billing_object_connections.pluck(:category)).to match_array(%w[payment tax accounting crm])
+      expect(wallet.effective_payment_connection).to eq(stripe_connection)
+      expect(wallet.effective_accounting_connection).to eq(netsuite_connection)
+      expect(wallet.effective_tax_connection).to be_nil
+    end
+
+    it "keeps the top-level payment_method working alongside connections" do
+      create_params[:payment_method] = {payment_method_type: "provider", payment_method_id: payment_method.id}
+
+      subject
+
+      expect(response).to have_http_status(:success)
+      expect(json[:wallet][:payment_method][:payment_method_id]).to eq(payment_method.id)
+      expect(Wallet.find(json[:wallet][:lago_id]).billing_object_connections.count).to eq(4)
+    end
+
+    context "when a code does not resolve" do
+      let(:create_params) do
+        {
+          external_customer_id: customer.external_id,
+          rate_amount: "1",
+          name: "Wallet1",
+          currency: "EUR",
+          connections: {payment: {code: "unknown_connection"}}
+        }
+      end
+
+      it "returns a validation error" do
+        subject
+
+        expect(response).to have_http_status(:unprocessable_content)
+        expect(json[:error_details][:connections]).to include("connection_not_found")
+      end
+    end
+
+    context "when the behavior is invalid" do
+      let(:create_params) do
+        {
+          external_customer_id: customer.external_id,
+          rate_amount: "1",
+          name: "Wallet1",
+          currency: "EUR",
+          connections: {payment: {behavior: "nonsense"}}
+        }
+      end
+
+      it "returns a validation error" do
+        subject
+
+        expect(response).to have_http_status(:unprocessable_content)
+        expect(json[:error_details][:connections]).to include("invalid_connection_behavior")
+      end
+    end
+
+    context "when the multi_connection flag is disabled" do
+      before { organization.disable_feature_flag!(:multi_connection) }
+
+      it "returns a forbidden error" do
+        subject
+
+        expect(response).to have_http_status(:forbidden)
+      end
+    end
+
+    context "with connections on a recurring transaction rule", :premium do
+      let(:create_params) do
+        {
+          external_customer_id: customer.external_id,
+          rate_amount: "1",
+          name: "Wallet1",
+          currency: "EUR",
+          paid_credits: "10",
+          granted_credits: "10",
+          recurring_transaction_rules: [
+            {
+              trigger: "interval",
+              interval: "monthly",
+              connections: {payment: {code: "stripe_us"}}
+            }
+          ]
+        }
+      end
+
+      it "owns the connection by the rule, not the wallet" do
+        subject
+
+        expect(response).to have_http_status(:success)
+
+        wallet = Wallet.find(json[:wallet][:lago_id])
+        expect(wallet.billing_object_connections).to be_empty
+
+        rule = wallet.recurring_transaction_rules.sole
+        expect(rule.billing_object_connections.sole).to have_attributes(
+          category: "payment",
+          behavior: "specific",
+          payment_provider_customer_id: stripe_connection.id
+        )
+      end
+    end
+  end
 end
 
 RSpec.shared_examples "a wallet create endpoint with billing_entity_id" do
@@ -1252,6 +1385,48 @@
       end
     end
   end
+
+  context "with connections" do
+    let(:stripe_connection) { create(:stripe_customer, customer:, code: "stripe_us") }
+    let(:update_params) { {name: "wallet1", connections: {payment: {code: "stripe_us"}}} }
+
+    before do
+      organization.enable_feature_flag!(:multi_connection)
+      stripe_connection
+    end
+
+    it "pins the connection on the wallet" do
+      expect { subject }.to change(BillingObjectConnection, :count).by(1)
+
+      expect(response).to have_http_status(:success)
+      expect(wallet.reload.effective_payment_connection).to eq(stripe_connection)
+    end
+
+    context "when inherit is sent for a category that has an override" do
+      let(:update_params) { {name: "wallet1", connections: {payment: {behavior: "inherit"}}} }
+
+      before do
+        create(:billing_object_connection, owner: wallet, organization:, category: "payment", behavior: "skip")
+      end
+
+      it "clears the override so resolution falls back to the customer" do
+        expect { subject }.to change(BillingObjectConnection, :count).by(-1)
+
+        expect(response).to have_http_status(:success)
+        expect(wallet.reload.billing_object_connections).to be_empty
+      end
+    end
+
+    context "when the multi_connection flag is disabled" do
+      before { organization.disable_feature_flag!(:multi_connection) }
+
+      it "returns a forbidden error" do
+        subject
+
+        expect(response).to have_http_status(:forbidden)
+      end
+    end
+  end
 end
 
 RSpec.shared_examples "a wallet show endpoint" do