diff --git a/fluxapay/src/lib.rs b/fluxapay/src/lib.rs
index cad126d..c73f1c6 100644
--- a/fluxapay/src/lib.rs
+++ b/fluxapay/src/lib.rs
@@ -366,6 +366,8 @@ pub enum Error {
     MaxRetriesExceeded = 54,
     /// Issue #505: Invalid payment status transition attempted.
     InvalidStatusTransition = 54,
+    /// Issue #450: Customer called `claim_refund` before an operator approved it.
+    RefundNotApproved = 56,
     /// Issue #437: DEX router is not in the allowed routers list.
     RouterNotAllowed = 56,
     /// Issue #436: Aggregate route output is less than minimum output amount.
@@ -2062,6 +2064,30 @@ impl RefundManager {
         Ok(())
     }
 
+    /// Issue #450: Customer self-serves an operator-approved refund.
+    ///
+    /// Callable only by the original refund requester, and only once an
+    /// operator has called `approve_refund`. `process_refund` remains
+    /// available for operators/oracles who need to execute a refund
+    /// directly without waiting for the customer to claim it.
+    pub fn claim_refund(env: Env, requester: Address, refund_id: String) -> Result<(), Error> {
+        requester.require_auth();
+        Self::require_not_paused(&env)?;
+        Self::require_not_blacklisted(&env, &requester)?;
+
+        let refund = Self::get_refund_internal(&env, &refund_id)?;
+
+        if refund.requester != requester {
+            return Err(Error::Unauthorized);
+        }
+        if refund.status != RefundStatus::Pending {
+            return Err(Error::RefundAlreadyProcessed);
+        }
+        if !refund.approved {
+            return Err(Error::RefundNotApproved);
+        }
+
+        Self::process_refund_internal(&env, &requester, refund_id)
     /// Admin-configurable refund expiry window in seconds (Issue #170).
     /// Applies to refunds created after this call.
     pub fn set_refund_expiry(env: Env, admin: Address, secs: u64) -> Result<(), Error> {
diff --git a/fluxapay/src/merchant_auth.rs b/fluxapay/src/merchant_auth.rs
index 3241b76..397b3a1 100644
--- a/fluxapay/src/merchant_auth.rs
+++ b/fluxapay/src/merchant_auth.rs
@@ -237,6 +237,14 @@ impl MerchantPreAuth {
                 .period_start
                 .saturating_add(periods_elapsed * auth.period_secs);
             auth.pulled_this_period = 0;
+
+            env.events().publish(
+                (
+                    Symbol::new(&env, "MERCHANT_AUTH"),
+                    Symbol::new(&env, "PERIOD_RESET"),
+                ),
+                (customer.clone(), merchant.clone(), auth.period_start),
+            );
         }
 
         // ── Limit check ───────────────────────────────────────────────────────
@@ -316,3 +324,124 @@ impl MerchantPreAuth {
         Ok(auth.limit_per_period.saturating_sub(pulled).max(0))
     }
 }
+
+#[cfg(test)]
+mod period_reset_tests {
+    use crate::{PaymentProcessor, PaymentProcessorClient};
+    use soroban_sdk::{
+        testutils::{Address as _, Events as _, Ledger as _},
+        token, Address, Env,
+    };
+
+    fn setup(env: &Env) -> (Address, PaymentProcessorClient<'_>) {
+        let contract_id = env.register(PaymentProcessor, ());
+        let client = PaymentProcessorClient::new(env, &contract_id);
+        let admin = Address::generate(env);
+        client.initialize_payment_processor(&admin);
+        (admin, client)
+    }
+
+    fn setup_authorization(
+        env: &Env,
+        client: &PaymentProcessorClient<'_>,
+        limit_per_period: i128,
+        period_secs: u64,
+    ) -> (Address, Address, Address) {
+        let customer = Address::generate(env);
+        let merchant = Address::generate(env);
+        let token_admin = Address::generate(env);
+        let token = env
+            .register_stellar_asset_contract_v2(token_admin)
+            .address();
+
+        token::StellarAssetClient::new(env, &token).mint(&customer, &1_000_000_000i128);
+
+        client.pre_authorize_merchant(
+            &customer,
+            &merchant,
+            &token,
+            &limit_per_period,
+            &period_secs,
+        );
+
+        (customer, merchant, token)
+    }
+
+    #[test]
+    fn test_period_reset_on_first_pull_of_new_period() {
+        let env = Env::default();
+        env.mock_all_auths();
+        let (_admin, client) = setup(&env);
+        let (customer, merchant, _token) = setup_authorization(&env, &client, 1_000i128, 100u64);
+
+        client.pull_payment(&merchant, &customer, &1_000i128);
+        // Fully spent for this period.
+        let result = client.try_pull_payment(&merchant, &customer, &1i128);
+        assert!(result.is_err());
+
+        // Advance past the period boundary — the first pull of the new
+        // period must reset `pulled_this_period` before applying the limit.
+        env.ledger().with_mut(|li| li.timestamp += 101);
+        let pulled = client.pull_payment(&merchant, &customer, &500i128);
+        assert_eq!(pulled, 500i128);
+
+        let auth = client.get_merchant_authorization(&customer, &merchant);
+        assert_eq!(auth.pulled_this_period, 500i128);
+    }
+
+    #[test]
+    fn test_limit_enforced_within_period() {
+        let env = Env::default();
+        env.mock_all_auths();
+        let (_admin, client) = setup(&env);
+        let (customer, merchant, _token) = setup_authorization(&env, &client, 1_000i128, 100u64);
+
+        client.pull_payment(&merchant, &customer, &700i128);
+        let result = client.try_pull_payment(&merchant, &customer, &400i128);
+        assert_eq!(
+            result,
+            Err(Ok(crate::merchant_auth::MerchantAuthError::LimitExceeded))
+        );
+
+        // Remaining budget in the same period is still pullable.
+        let pulled = client.pull_payment(&merchant, &customer, &300i128);
+        assert_eq!(pulled, 1_000i128);
+    }
+
+    #[test]
+    fn test_multi_period_pulls_all_succeed() {
+        let env = Env::default();
+        env.mock_all_auths();
+        let (_admin, client) = setup(&env);
+        let (customer, merchant, _token) = setup_authorization(&env, &client, 1_000i128, 100u64);
+
+        for _ in 0..3 {
+            let pulled = client.pull_payment(&merchant, &customer, &1_000i128);
+            assert_eq!(pulled, 1_000i128);
+            env.ledger().with_mut(|li| li.timestamp += 101);
+        }
+    }
+
+    #[test]
+    fn test_period_reset_event_emitted_on_rollover() {
+        let env = Env::default();
+        env.mock_all_auths();
+        let (_admin, client) = setup(&env);
+        let (customer, merchant, _token) = setup_authorization(&env, &client, 1_000i128, 100u64);
+
+        // Pull within the same period — no reset, no PERIOD_RESET event.
+        client.pull_payment(&merchant, &customer, &500i128);
+        let events_before = env.events().all().len();
+
+        // Roll over into a new period — this pull must emit PERIOD_RESET
+        // in addition to the usual CHARGED event.
+        env.ledger().with_mut(|li| li.timestamp += 101);
+        client.pull_payment(&merchant, &customer, &200i128);
+        let events_after = env.events().all().len();
+
+        assert!(
+            events_after > events_before + 1,
+            "expected an extra MERCHANT_AUTH/PERIOD_RESET event on period rollover"
+        );
+    }
+}
diff --git a/fluxapay/src/test.rs b/fluxapay/src/test.rs
index 5958594..96bd868 100644
--- a/fluxapay/src/test.rs
+++ b/fluxapay/src/test.rs
@@ -328,6 +328,130 @@ fn test_create_stream_fails_for_blacklisted_sender() {
     assert_eq!(result, Err(Ok(StreamError::Unauthorized)));
 }
 
+#[test]
+fn test_pause_stream_checkpoints_accrual_and_sets_paused() {
+    let env = Env::default();
+    env.mock_all_auths();
+    let (_admin, client) = setup_payment_processor(&env);
+
+    let token_admin = Address::generate(&env);
+    let token = env
+        .register_stellar_asset_contract_v2(token_admin.clone())
+        .address();
+    let sender = Address::generate(&env);
+    let recipient = Address::generate(&env);
+    let stream_id = String::from_str(&env, "pause_stream_1");
+
+    token::StellarAssetClient::new(&env, &token).mint(&sender, &1_000_000i128);
+    client.create_stream(&sender, &recipient, &token, &10i128, &1_000i128, &stream_id);
+
+    env.ledger().with_mut(|li| li.timestamp += 50);
+
+    client.pause_stream(&sender, &stream_id);
+
+    let stream = client.get_stream(&stream_id);
+    assert_eq!(stream.status, StreamStatus::Paused);
+    // 50 seconds at rate 10/s should have been checkpointed.
+    assert_eq!(stream.accrued_at_checkpoint, 500i128);
+    assert_eq!(stream.last_checkpoint_at, env.ledger().timestamp());
+}
+
+#[test]
+fn test_double_pause_stream_rejected() {
+    let env = Env::default();
+    env.mock_all_auths();
+    let (_admin, client) = setup_payment_processor(&env);
+
+    let token_admin = Address::generate(&env);
+    let token = env
+        .register_stellar_asset_contract_v2(token_admin.clone())
+        .address();
+    let sender = Address::generate(&env);
+    let recipient = Address::generate(&env);
+    let stream_id = String::from_str(&env, "double_pause_stream_1");
+
+    token::StellarAssetClient::new(&env, &token).mint(&sender, &1_000_000i128);
+    client.create_stream(&sender, &recipient, &token, &10i128, &1_000i128, &stream_id);
+    client.pause_stream(&sender, &stream_id);
+
+    let result = client.try_pause_stream(&sender, &stream_id);
+    assert_eq!(result, Err(Ok(StreamError::StreamNotActive)));
+}
+
+#[test]
+fn test_resume_stream_restarts_accrual_from_correct_point() {
+    let env = Env::default();
+    env.mock_all_auths();
+    let (_admin, client) = setup_payment_processor(&env);
+
+    let token_admin = Address::generate(&env);
+    let token = env
+        .register_stellar_asset_contract_v2(token_admin.clone())
+        .address();
+    let sender = Address::generate(&env);
+    let recipient = Address::generate(&env);
+    let stream_id = String::from_str(&env, "resume_stream_1");
+
+    token::StellarAssetClient::new(&env, &token).mint(&sender, &1_000_000i128);
+    client.create_stream(&sender, &recipient, &token, &10i128, &1_000i128, &stream_id);
+
+    env.ledger().with_mut(|li| li.timestamp += 50);
+    client.pause_stream(&sender, &stream_id);
+
+    // Time passes while paused — must not accrue.
+    env.ledger().with_mut(|li| li.timestamp += 200);
+    client.resume_stream(&sender, &stream_id);
+
+    let stream = client.get_stream(&stream_id);
+    assert_eq!(stream.status, StreamStatus::Active);
+    // Accrual while paused must not be counted; only the pre-pause 50s * 10/s.
+    assert_eq!(stream.accrued_at_checkpoint, 500i128);
+    assert_eq!(stream.last_checkpoint_at, env.ledger().timestamp());
+}
+
+#[test]
+fn test_resume_non_paused_stream_rejected() {
+    let env = Env::default();
+    env.mock_all_auths();
+    let (_admin, client) = setup_payment_processor(&env);
+
+    let token_admin = Address::generate(&env);
+    let token = env
+        .register_stellar_asset_contract_v2(token_admin.clone())
+        .address();
+    let sender = Address::generate(&env);
+    let recipient = Address::generate(&env);
+    let stream_id = String::from_str(&env, "resume_active_stream_1");
+
+    token::StellarAssetClient::new(&env, &token).mint(&sender, &1_000_000i128);
+    client.create_stream(&sender, &recipient, &token, &10i128, &1_000i128, &stream_id);
+
+    let result = client.try_resume_stream(&sender, &stream_id);
+    assert_eq!(result, Err(Ok(StreamError::StreamNotPaused)));
+}
+
+#[test]
+fn test_pause_resume_stream_unauthorized_for_non_sender() {
+    let env = Env::default();
+    env.mock_all_auths();
+    let (_admin, client) = setup_payment_processor(&env);
+
+    let token_admin = Address::generate(&env);
+    let token = env
+        .register_stellar_asset_contract_v2(token_admin.clone())
+        .address();
+    let sender = Address::generate(&env);
+    let recipient = Address::generate(&env);
+    let stranger = Address::generate(&env);
+    let stream_id = String::from_str(&env, "pause_unauthorized_stream_1");
+
+    token::StellarAssetClient::new(&env, &token).mint(&sender, &1_000_000i128);
+    client.create_stream(&sender, &recipient, &token, &10i128, &1_000i128, &stream_id);
+
+    let result = client.try_pause_stream(&stranger, &stream_id);
+    assert_eq!(result, Err(Ok(StreamError::Unauthorized)));
+}
+
 #[test]
 fn test_batch_withdraw_to_custom_routing() {
     let env = Env::default();
@@ -897,6 +1021,9 @@ fn test_process_refund_rejects_after_expiry() {
     env.mock_all_auths();
     let (admin, client) = setup_refund_manager(&env);
 
+    let payment_id = String::from_str(&env, "payment_claim_1");
+    let merchant_id = Address::generate(&env);
+    let refund_amount = 1000i128;
     let payment_id = String::from_str(&env, "payment_expiry_bad");
     let merchant_id = Address::generate(&env);
     let requester = Address::generate(&env);
@@ -910,6 +1037,7 @@ fn test_process_refund_rejects_after_expiry() {
 
     let refund_id = client.create_refund(
         &payment_id,
+        &refund_amount,
         &1000i128,
         &String::from_str(&env, "Reason"),
         &requester,
@@ -918,6 +1046,48 @@ fn test_process_refund_rejects_after_expiry() {
     let operator = Address::generate(&env);
     client.grant_role(&admin, &role_settlement_operator(&env), &operator);
 
+    client.approve_refund(&operator, &refund_id);
+    let refund = client.get_refund(&refund_id);
+    assert!(refund.approved);
+    assert_eq!(refund.status, RefundStatus::Pending);
+
+    client.claim_refund(&requester, &refund_id);
+
+    let refund = client.get_refund(&refund_id);
+    assert_eq!(refund.status, RefundStatus::Completed);
+}
+
+#[test]
+fn test_claim_refund_before_approval_fails() {
+    let env = Env::default();
+    env.mock_all_auths();
+    let (_admin, client) = setup_refund_manager(&env);
+
+    let payment_id = String::from_str(&env, "payment_claim_2");
+    let merchant_id = Address::generate(&env);
+    let refund_amount = 1000i128;
+    let requester = Address::generate(&env);
+
+    client.register_payment(
+        &payment_id,
+        &merchant_id,
+        &5000i128,
+        &Symbol::new(&env, "USDC"),
+    );
+
+    let refund_id = client.create_refund(
+        &payment_id,
+        &refund_amount,
+        &String::from_str(&env, "Reason"),
+        &requester,
+    );
+
+    let result = client.try_claim_refund(&requester, &refund_id);
+    assert_eq!(result, Err(Ok(Error::RefundNotApproved)));
+}
+
+#[test]
+fn test_claim_refund_by_non_requester_fails() {
     // 60 days later — past the default 30-day expiry window.
     env.ledger()
         .set_timestamp(env.ledger().timestamp() + 60 * 24 * 60 * 60);
@@ -932,6 +1102,11 @@ fn test_expire_refund_clears_pending_expired_refund() {
     env.mock_all_auths();
     let (admin, client) = setup_refund_manager(&env);
 
+    let payment_id = String::from_str(&env, "payment_claim_3");
+    let merchant_id = Address::generate(&env);
+    let refund_amount = 1000i128;
+    let requester = Address::generate(&env);
+    let stranger = Address::generate(&env);
     let payment_id = String::from_str(&env, "payment_expire_cleanup");
     let merchant_id = Address::generate(&env);
     let requester = Address::generate(&env);
@@ -945,6 +1120,7 @@ fn test_expire_refund_clears_pending_expired_refund() {
 
     let refund_id = client.create_refund(
         &payment_id,
+        &refund_amount,
         &1000i128,
         &String::from_str(&env, "Reason"),
         &requester,
@@ -952,6 +1128,47 @@ fn test_expire_refund_clears_pending_expired_refund() {
 
     let operator = Address::generate(&env);
     client.grant_role(&admin, &role_settlement_operator(&env), &operator);
+    client.approve_refund(&operator, &refund_id);
+
+    let result = client.try_claim_refund(&stranger, &refund_id);
+    assert_eq!(result, Err(Ok(Error::Unauthorized)));
+}
+
+#[test]
+fn test_double_claim_refund_blocked() {
+    let env = Env::default();
+    env.mock_all_auths();
+    let (admin, client) = setup_refund_manager(&env);
+
+    let payment_id = String::from_str(&env, "payment_claim_4");
+    let merchant_id = Address::generate(&env);
+    let refund_amount = 1000i128;
+    let requester = Address::generate(&env);
+
+    client.register_payment(
+        &payment_id,
+        &merchant_id,
+        &5000i128,
+        &Symbol::new(&env, "USDC"),
+    );
+
+    let refund_id = client.create_refund(
+        &payment_id,
+        &refund_amount,
+        &String::from_str(&env, "Reason"),
+        &requester,
+    );
+
+    let operator = Address::generate(&env);
+    client.grant_role(&admin, &role_settlement_operator(&env), &operator);
+    client.approve_refund(&operator, &refund_id);
+    client.claim_refund(&requester, &refund_id);
+
+    let result = client.try_claim_refund(&requester, &refund_id);
+    assert_eq!(result, Err(Ok(Error::RefundAlreadyProcessed)));
+}
+
+#[test]
 
     env.ledger()
         .set_timestamp(env.ledger().timestamp() + 60 * 24 * 60 * 60);
diff --git a/sdk/README.md b/sdk/README.md
index 8094244..c78b26b 100644
--- a/sdk/README.md
+++ b/sdk/README.md
@@ -231,6 +231,43 @@ const dispute = await client.getDispute("dispute_001");
 const paymentDisputes = await client.getPaymentDisputes("pay_123");
 ```
 
+## Merchant Pre-Authorization (Pull Billing)
+
+`MerchantPreAuth` lets a customer grant a merchant permission to pull up to a
+fixed amount per billing period — useful for SaaS-style recurring charges
+without requiring a fresh signature on every charge.
+
+```typescript
+// Customer grants the merchant a $50/30-day pull allowance.
+const auth = await client.preAuthorizeMerchant({
+  customer: "GCUSTOMER...",
+  merchant: "GMERCHANT...",
+  token: "CUSDC...",
+  limitPerPeriod: 50_000_000n, // 50 USDC (7 decimals)
+  periodSecs: 2_592_000n, // 30 days
+});
+
+// Merchant pulls a charge against the authorization. Returns the
+// cumulative amount pulled so far in the current period.
+const pulledThisPeriod = await client.pullFromAuthorization(
+  "GMERCHANT...",
+  "GCUSTOMER...",
+  10_000_000n, // 10 USDC
+);
+
+// Look up the current authorization (null if none exists).
+const current = await client.getAuthorization("GCUSTOMER...", "GMERCHANT...");
+
+// Customer revokes the authorization at any time.
+await client.revokeAuthorization("GCUSTOMER...", "GMERCHANT...");
+```
+
+Billing periods reset automatically: once `now >= period_start + period_secs`,
+the next `pullFromAuthorization` call resets `pulled_this_period` to 0 and
+emits a `MERCHANT_AUTH/PERIOD_RESET` event before applying the pull, so a new
+period always starts with the full `limitPerPeriod` available regardless of
+how many periods were skipped with no activity.
+
 ## RefundManagerClient
 
 The `RefundManagerClient` provides methods for managing refunds on a dedicated RefundManager contract:
diff --git a/sdk/src/index.ts b/sdk/src/index.ts
index 8fdf57b..8e14ff1 100644
--- a/sdk/src/index.ts
+++ b/sdk/src/index.ts
@@ -186,6 +186,32 @@ export interface RegisterMerchantParams {
   feeConfig?: FeeConfig;
 }
 
+/**
+ * A customer's pre-authorization for a merchant to pull recurring payments,
+ * mirroring `MerchantAuthorization` in `fluxapay/src/merchant_auth.rs`.
+ */
+export interface MerchantAuthorization {
+  customer: string;
+  merchant: string;
+  token: string;
+  limit_per_period: bigint;
+  period_secs: bigint;
+  period_start: bigint;
+  pulled_this_period: bigint;
+  active: boolean;
+  created_at: bigint;
+}
+
+/** Error codes from `fluxapay/src/merchant_auth.rs::MerchantAuthError`. */
+export const MerchantAuthError = {
+  1: { message: "AuthorizationNotFound" },
+  2: { message: "AuthorizationInactive" },
+  3: { message: "LimitExceeded" },
+  4: { message: "InvalidAmount" },
+  5: { message: "Unauthorized" },
+  6: { message: "AuthorizationAlreadyExists" },
+} as const;
+
 export interface UpdateMerchantParams {
   merchantId: string;
   businessName?: string;
@@ -704,6 +730,100 @@ export class FluxapayClient {
     );
   }
 
+  // ── Merchant pre-authorization (pull billing, #454) ─────────────────────────
+  //
+  // These delegate to `pre_authorize_merchant` / `pull_payment` /
+  // `revoke_merchant_authorization` / `get_merchant_authorization`, entry
+  // points already exposed on `PaymentProcessor` (see
+  // `fluxapay/src/lib.rs`). They're invoked via a loose cast because the
+  // checked-in `contracts/fluxapay` bindings predate these entry points;
+  // regenerating bindings with `npm run generate` (see `scripts/generate-sdk.sh`)
+  // against a freshly built contract will pick up proper typings, at which
+  // point the `as any` casts below can be removed.
+
+  /**
+   * Customer grants a merchant permission to pull up to `limitPerPeriod`
+   * tokens per `periodSecs`-second billing window.
+   */
+  async preAuthorizeMerchant(params: {
+    customer: string;
+    merchant: string;
+    token: string;
+    limitPerPeriod: bigint;
+    periodSecs: bigint;
+  }): Promise<MerchantAuthorization> {
+    return withMappedContractError(async () => {
+      const tx = await (this.contract as any).pre_authorize_merchant({
+        customer: params.customer,
+        merchant: params.merchant,
+        token: params.token,
+        limit_per_period: params.limitPerPeriod,
+        period_secs: params.periodSecs,
+      });
+      return tx.result;
+    });
+  }
+
+  /**
+   * Merchant pulls `amount` tokens from `customer` against an existing
+   * pre-authorization. Returns the cumulative amount pulled this period.
+   */
+  async pullFromAuthorization(
+    merchant: string,
+    customer: string,
+    amount: bigint,
+  ): Promise<bigint> {
+    return withMappedContractError(async () => {
+      const tx = await (this.contract as any).pull_payment({
+        merchant,
+        customer,
+        amount,
+      });
+      return tx.result;
+    });
+  }
+
+  /**
+   * Customer revokes a previously granted merchant authorization.
+   */
+  async revokeAuthorization(customer: string, merchant: string): Promise<void> {
+    return withMappedContractError(async () => {
+      const tx = await (this.contract as any).revoke_merchant_authorization({
+        customer,
+        merchant,
+      });
+      return tx.result;
+    });
+  }
+
+  /**
+   * Fetch the stored authorization for a (customer, merchant) pair, or
+   * `null` if none exists.
+   */
+  async getAuthorization(
+    customer: string,
+    merchant: string,
+  ): Promise<MerchantAuthorization | null> {
+    try {
+      return await withMappedContractError(async () => {
+        const tx = await (this.contract as any).get_merchant_authorization({
+          customer,
+          merchant,
+        });
+        return tx.result;
+      });
+    } catch (error) {
+      // Note: `FLUXAPAY_CONTRACT_ERROR_MAP` only covers the main `Error`
+      // enum, whose code space overlaps with `MerchantAuthError`'s — code 1
+      // means `AuthorizationNotFound` here, not the mapped "Unauthorized"
+      // name (see docs/error-codes.md). Check the raw code, not the name.
+      if (error instanceof FluxapayError && error.code === 1) {
+        return null;
+      }
+      throw error;
+    }
+  }
+
   /**
    * Get all refunds for a payment
    */