diff --git a/make_post_sell/lib/digest_sender.py b/make_post_sell/lib/digest_sender.py
index e6804f0..8cb5c80 100644
--- a/make_post_sell/lib/digest_sender.py
+++ b/make_post_sell/lib/digest_sender.py
@@ -122,6 +122,7 @@ def run(env, frequency_str, dry_run=False):
     shops = (
         dbsession.query(Shop)
         .filter(Shop.subscriptions_enabled == True)
+        .filter(Shop.environment == 0)  # MPS-14: exclude non-production shops
         .all()
     )
 
diff --git a/make_post_sell/lib/karaoke.py b/make_post_sell/lib/karaoke.py
index d861c2d..9cafc4e 100644
--- a/make_post_sell/lib/karaoke.py
+++ b/make_post_sell/lib/karaoke.py
@@ -362,6 +362,15 @@ def backfill_karaoke_async(shop_id, session_factory, app_settings):
         session = SASession(bind=engine)
 
         def _make_s3():
+            # BYOB: use shop's own bucket if configured (MPS-16)
+            if shop and shop.has_primary_s3:
+                return boto3.session.Session().client(
+                    "s3",
+                    region_name=shop.primary_s3_region,
+                    endpoint_url=shop.primary_s3_endpoint,
+                    aws_access_key_id=shop.primary_s3_access_key,
+                    aws_secret_access_key=shop.primary_s3_secret_key,
+                )
             return boto3.session.Session().client(
                 "s3",
                 region_name=app_settings["bucket.secure_uploads.region"],
@@ -370,14 +379,15 @@ def backfill_karaoke_async(shop_id, session_factory, app_settings):
                 aws_secret_access_key=app_settings["bucket.secure_uploads.secret_key"],
             )
 
-        s3 = _make_s3()
-        bucket = app_settings["bucket.secure_uploads"]
-
         try:
+            # Must load shop before _make_s3 can check has_primary_s3
             shop = session.get(Shop, shop_id)
             if not shop or not shop.unsandbox_public_key or not shop.unsandbox_secret_key:
                 return
 
+            s3 = _make_s3()
+            bucket = shop.primary_s3_bucket if shop.has_primary_s3 else app_settings["bucket.secure_uploads"]
+
             pk, sk = shop.unsandbox_public_key, shop.unsandbox_secret_key
 
             # Query account concurrency from unsandbox API — abort if keys are bad
diff --git a/make_post_sell/lib/s3_mirror.py b/make_post_sell/lib/s3_mirror.py
index ddd0729..343a0e7 100644
--- a/make_post_sell/lib/s3_mirror.py
+++ b/make_post_sell/lib/s3_mirror.py
@@ -229,7 +229,16 @@ def backfill_mirror_async(shop_id, session_factory, app_settings):
         engine = create_engine(db_url)
         session = SASession(bind=engine)
 
-        def _make_src():
+        def _make_src(shop):
+            # BYOB: if shop has its own primary bucket, mirror from there (MPS-16)
+            if shop and shop.has_primary_s3:
+                return boto3.session.Session().client(
+                    "s3",
+                    region_name=shop.primary_s3_region,
+                    endpoint_url=shop.primary_s3_endpoint,
+                    aws_access_key_id=shop.primary_s3_access_key,
+                    aws_secret_access_key=shop.primary_s3_secret_key,
+                )
             return boto3.session.Session().client(
                 "s3",
                 region_name=app_settings["bucket.secure_uploads.region"],
@@ -243,14 +252,14 @@ def backfill_mirror_async(shop_id, session_factory, app_settings):
             if not shop or not shop.has_s3_mirror:
                 return
 
-            src = _make_src()
+            src = _make_src(shop)
             dst = _make_mirror_client(
                 shop.mirror_s3_endpoint,
                 shop.mirror_s3_region,
                 shop.mirror_s3_access_key,
                 shop.mirror_s3_secret_key,
             )
-            src_bucket = app_settings["bucket.secure_uploads"]
+            src_bucket = shop.primary_s3_bucket if shop.has_primary_s3 else app_settings["bucket.secure_uploads"]
             dst_bucket = shop.mirror_s3_bucket
 
             # List all objects under the shop's prefix
diff --git a/make_post_sell/models/product.py b/make_post_sell/models/product.py
index 3fe1fc4..ca98ecc 100644
--- a/make_post_sell/models/product.py
+++ b/make_post_sell/models/product.py
@@ -677,9 +677,12 @@ def get_products_by_keywords(dbsession, keywords, shop=None):
     for keyword in keywords:
         keyword_filter = Product.title.ilike(f"%{keyword}%")
         # the product _must_ be public (1).
-        products = (
-            product_query.filter(keyword_filter).filter(Product.visibility == 1).all()
-        )
+        query = product_query.filter(keyword_filter).filter(Product.visibility == 1)
+        # Exclude non-production shops from search results (MPS-14)
+        if not shop:
+            from .shop import Shop
+            query = query.join(Shop, Product.shop_id == Shop.id).filter(Shop.environment == 0)
+        products = query.all()
 
         for product in products:
             if product.id not in scores:
diff --git a/make_post_sell/templates/shop_settings.j2 b/make_post_sell/templates/shop_settings.j2
index 3e5c136..797bf8c 100644
--- a/make_post_sell/templates/shop_settings.j2
+++ b/make_post_sell/templates/shop_settings.j2
@@ -1212,6 +1212,10 @@ Existing sales honored for download buy purchasers.
 
     <p>Configure your own S3-compatible storage bucket. When enabled, all media uploads and CDN URLs will use your bucket instead of the default MPS storage.</p>
 
+    {% if request.shop.is_trial_active %}
+    <p><strong>Trial tip:</strong> Setting up your own storage bucket during your trial ensures your media is always under your control. Any S3-compatible provider works (DigitalOcean Spaces, AWS S3, Backblaze B2, etc.).</p>
+    {% endif %}
+
     <form method="post" action="/s/{{ request.shop.id }}/settings" onsubmit="submit.disabled = true; return true;">
       <input type="hidden" name="form_section" value="bucket-settings" />
 
diff --git a/make_post_sell/tests/test_functional.py b/make_post_sell/tests/test_functional.py
index 7939db4..320ae3a 100644
--- a/make_post_sell/tests/test_functional.py
+++ b/make_post_sell/tests/test_functional.py
@@ -4675,7 +4675,11 @@ class TestBucketSettings(_AuthenticatedBase):
         )
         res = res.follow()
         flash = self._get_flash_messages(res)
-        self.assertIn("Storage bucket settings updated", flash)
+        # Connection test runs on save — may fail in test env but settings are saved
+        self.assertTrue(
+            "connection test failed" in flash or "Storage bucket settings" in flash,
+            f"Unexpected flash: {flash}"
+        )
         self.dbsession.refresh(shop)
         self.assertTrue(shop.primary_s3_enabled)
         self.assertEqual(shop.primary_s3_bucket, "my-test-bucket")
diff --git a/make_post_sell/tests/test_integration.py b/make_post_sell/tests/test_integration.py
index dadd55f..d635f40 100644
--- a/make_post_sell/tests/test_integration.py
+++ b/make_post_sell/tests/test_integration.py
@@ -3689,3 +3689,199 @@ class TestGiftCardIntegration(DatabaseIntegrationTests):
         self.assertEqual(cart.gift_card_purchases_total_in_cents, 2500)
 
         transaction.commit()
+
+
+class TestTrialIntegration(DatabaseIntegrationTests):
+    """MPS-15: Integration tests for trial system with real ORM objects."""
+
+    def _make_shop(self, **kwargs):
+        shop = Shop(
+            name="Trial Shop",
+            phone_number="555-555-5555",
+            billing_address="123 Test St",
+            description="A trial shop",
+        )
+        shop.domain_name = "trial.test.com"
+        for k, v in kwargs.items():
+            setattr(shop, k, v)
+        self.dbsession.add(shop)
+        self.dbsession.flush()
+        return shop
+
+    def test_grandfathered_shop_not_expired(self):
+        """Pre-trial shops (NULL trial_started_timestamp) are never expired."""
+        shop = self._make_shop(trial_started_timestamp=None)
+        self.assertFalse(shop.is_trial_expired)
+        self.assertFalse(shop.is_trial_active)
+        self.assertTrue(shop.is_active)
+        transaction.commit()
+
+    def test_active_trial(self):
+        """Shop within 21-day trial window is active."""
+        now_ms = int(time.time() * 1000)
+        shop = self._make_shop(trial_started_timestamp=now_ms)
+        self.assertTrue(shop.is_trial_active)
+        self.assertFalse(shop.is_trial_expired)
+        self.assertTrue(shop.is_active)
+        self.assertIn(shop.trial_days_remaining, (20, 21))  # depends on sub-day timing
+        transaction.commit()
+
+    def test_expired_trial(self):
+        """Shop past 21-day window is expired."""
+        expired_ms = int(time.time() * 1000) - (22 * 24 * 60 * 60 * 1000)
+        shop = self._make_shop(trial_started_timestamp=expired_ms)
+        self.assertFalse(shop.is_trial_active)
+        self.assertTrue(shop.is_trial_expired)
+        self.assertFalse(shop.is_active)
+        self.assertEqual(shop.trial_days_remaining, 0)
+        transaction.commit()
+
+    def test_paid_plan_overrides_trial(self):
+        """Paid plan makes shop active regardless of trial status."""
+        expired_ms = int(time.time() * 1000) - (22 * 24 * 60 * 60 * 1000)
+        shop = self._make_shop(trial_started_timestamp=expired_ms, plan_active=True)
+        self.assertFalse(shop.is_trial_active)
+        self.assertFalse(shop.is_trial_expired)
+        self.assertTrue(shop.is_active)
+        transaction.commit()
+
+    def test_trial_with_products(self):
+        """Products in a trial shop are accessible but cannot be created when expired."""
+        expired_ms = int(time.time() * 1000) - (22 * 24 * 60 * 60 * 1000)
+        shop = self._make_shop(trial_started_timestamp=expired_ms)
+
+        # Existing products are still in the database and readable
+        product = Product(title="Existing Product", description="Created before trial expired")
+        product.shop_id = shop.id
+        product.price_in_cents = 1000
+        self.dbsession.add(product)
+        self.dbsession.flush()
+
+        # Shop is expired but product still exists
+        self.assertTrue(shop.is_trial_expired)
+        self.assertEqual(product.shop_id, shop.id)
+        transaction.commit()
+
+    def test_environment_with_trial(self):
+        """Environment and trial are independent — non-prod shop can have trial."""
+        now_ms = int(time.time() * 1000)
+        shop = self._make_shop(trial_started_timestamp=now_ms, environment=1)
+        self.assertTrue(shop.is_trial_active)
+        self.assertTrue(shop.is_staging)
+        self.assertTrue(shop.is_non_production)
+        transaction.commit()
+
+
+class TestBYOBIntegration(DatabaseIntegrationTests):
+    """MPS-16: Integration tests for BYOB (Bring Your Own Bucket) with real ORM objects."""
+
+    def _make_shop(self, **kwargs):
+        shop = Shop(
+            name="BYOB Shop",
+            phone_number="555-555-5555",
+            billing_address="123 Test St",
+            description="A BYOB shop",
+        )
+        shop.domain_name = "byob.test.com"
+        for k, v in kwargs.items():
+            setattr(shop, k, v)
+        self.dbsession.add(shop)
+        self.dbsession.flush()
+        return shop
+
+    def test_has_primary_s3_false_by_default(self):
+        """New shops do not have BYOB enabled."""
+        shop = self._make_shop()
+        self.assertFalse(shop.has_primary_s3)
+        transaction.commit()
+
+    def test_has_primary_s3_requires_all_fields(self):
+        """BYOB requires all fields AND enabled flag."""
+        shop = self._make_shop(
+            primary_s3_endpoint="https://nyc3.digitaloceanspaces.com",
+            primary_s3_region="nyc3",
+            primary_s3_bucket="my-bucket",
+            primary_s3_access_key="AKIATEST",
+            primary_s3_secret_key="secret123",
+            primary_s3_cdn_endpoint="https://my-bucket.nyc3.cdn.digitaloceanspaces.com",
+            primary_s3_enabled=False,  # not enabled
+        )
+        self.assertFalse(shop.has_primary_s3)
+
+        shop.primary_s3_enabled = True
+        self.assertTrue(shop.has_primary_s3)
+        transaction.commit()
+
+    def test_has_primary_s3_missing_field(self):
+        """BYOB is false if any required field is missing."""
+        shop = self._make_shop(
+            primary_s3_endpoint="https://nyc3.digitaloceanspaces.com",
+            primary_s3_region="nyc3",
+            primary_s3_bucket="my-bucket",
+            primary_s3_access_key="AKIATEST",
+            primary_s3_secret_key="",  # missing
+            primary_s3_cdn_endpoint="https://my-bucket.nyc3.cdn.digitaloceanspaces.com",
+            primary_s3_enabled=True,
+        )
+        self.assertFalse(shop.has_primary_s3)
+        transaction.commit()
+
+    def test_media_cdn_endpoint_byob(self):
+        """media_cdn_endpoint returns shop's CDN when BYOB is active."""
+        shop = self._make_shop(
+            primary_s3_endpoint="https://nyc3.digitaloceanspaces.com",
+            primary_s3_region="nyc3",
+            primary_s3_bucket="my-bucket",
+            primary_s3_access_key="AKIATEST",
+            primary_s3_secret_key="secret123",
+            primary_s3_cdn_endpoint="https://custom-cdn.example.com",
+            primary_s3_enabled=True,
+        )
+        self.assertEqual(shop.media_cdn_endpoint, "https://custom-cdn.example.com")
+
+        # Disable BYOB — CDN returns None
+        shop.primary_s3_enabled = False
+        self.assertIsNone(shop.media_cdn_endpoint)
+        transaction.commit()
+
+    def test_byob_with_mirror(self):
+        """BYOB and mirror can coexist on the same shop."""
+        shop = self._make_shop(
+            primary_s3_endpoint="https://nyc3.digitaloceanspaces.com",
+            primary_s3_region="nyc3",
+            primary_s3_bucket="primary-bucket",
+            primary_s3_access_key="AKIATEST",
+            primary_s3_secret_key="secret123",
+            primary_s3_cdn_endpoint="https://primary-cdn.example.com",
+            primary_s3_enabled=True,
+            mirror_s3_endpoint="https://sfo3.digitaloceanspaces.com",
+            mirror_s3_region="sfo3",
+            mirror_s3_bucket="mirror-bucket",
+            mirror_s3_access_key="AKIAMIRROR",
+            mirror_s3_secret_key="mirrorsecret",
+            mirror_s3_enabled=True,
+        )
+        self.assertTrue(shop.has_primary_s3)
+        self.assertTrue(shop.has_s3_mirror)
+        transaction.commit()
+
+    def test_byob_persists_after_flush(self):
+        """BYOB fields round-trip through database correctly."""
+        shop = self._make_shop(
+            primary_s3_endpoint="https://nyc3.digitaloceanspaces.com",
+            primary_s3_region="nyc3",
+            primary_s3_bucket="test-bucket",
+            primary_s3_access_key="AKIATEST",
+            primary_s3_secret_key="secret123",
+            primary_s3_cdn_endpoint="https://cdn.example.com",
+            primary_s3_enabled=True,
+        )
+        shop_id = shop.id
+        self.dbsession.flush()
+
+        # Re-fetch from DB
+        fetched = self.dbsession.query(Shop).get(shop_id)
+        self.assertTrue(fetched.has_primary_s3)
+        self.assertEqual(fetched.primary_s3_bucket, "test-bucket")
+        self.assertEqual(fetched.primary_s3_cdn_endpoint, "https://cdn.example.com")
+        transaction.commit()
diff --git a/make_post_sell/views/__init__.py b/make_post_sell/views/__init__.py
index c6e21a8..91f85b1 100644
--- a/make_post_sell/views/__init__.py
+++ b/make_post_sell/views/__init__.py
@@ -88,6 +88,30 @@ def shop_owner_required(
     return wrapped
 
 
+# view decorator.
+def trial_active_required(
+    flash_msg="Your 21-day trial has expired. Choose a plan to continue.",
+    flash_level="error",
+):
+    """Block write operations when shop trial has expired.
+
+    Grandfathered shops (NULL trial_started_timestamp) and paid shops pass through.
+    Only blocks shops that have an expired trial with no active plan.
+    """
+
+    def wrapped(fn):
+        def inner(request):
+            shop = request.shop
+            if shop and shop.is_trial_expired:
+                request.session.flash((flash_msg, flash_level))
+                return HTTPFound(get_referer_or_home(request))
+            return fn(request)
+
+        return inner
+
+    return wrapped
+
+
 # view decorator.
 def shop_editor_required(
     flash_msg="You must have a shop editor role to access that.",
diff --git a/make_post_sell/views/cart.py b/make_post_sell/views/cart.py
index 543c6de..a930df5 100644
--- a/make_post_sell/views/cart.py
+++ b/make_post_sell/views/cart.py
@@ -4,6 +4,7 @@ from . import (
     user_required,
     get_referer_or_home,
     shop_is_ready_required,
+    trial_active_required,
 )
 
 from ..models.cart import get_cart_by_id
@@ -524,6 +525,7 @@ def cart_handling_option(request):
     redirect_to_route_name="join-or-log-in",
 )
 @shop_is_ready_required()
+@trial_active_required()
 def cart_checkout(request):
     stripe_user_shop = request.shop.stripe_user_shop(request.user)
     paypal_user_shop = request.shop.paypal_user_shop(request.user)
@@ -680,6 +682,7 @@ def cart_checkout(request):
 )
 @user_required()
 @shop_is_ready_required()
+@trial_active_required()
 def cart_complete_checkout(request):
     stripe_enabled = request.stripe_enabled
 
@@ -828,6 +831,7 @@ def cart_complete_checkout(request):
 )
 @user_required()
 @shop_is_ready_required()
+@trial_active_required()
 def paypal_complete_checkout(request):
     """Complete checkout using PayPal payment."""
     if not request.paypal_enabled:
@@ -1110,6 +1114,7 @@ def adyen_create_session(request):
 )
 @user_required()
 @shop_is_ready_required()
+@trial_active_required()
 def adyen_complete_checkout(request):
     """Complete checkout using Adyen payment."""
     if not getattr(request, "adyen_enabled", False):
diff --git a/make_post_sell/views/gift_card.py b/make_post_sell/views/gift_card.py
index ca0bc5b..fae1955 100644
--- a/make_post_sell/views/gift_card.py
+++ b/make_post_sell/views/gift_card.py
@@ -4,6 +4,7 @@ from pyramid.httpexceptions import HTTPFound
 from . import (
     user_required,
     shop_owner_required,
+    trial_active_required,
     get_referer_or_home,
 )
 
@@ -53,6 +54,7 @@ def gift_card_page(request):
 
 @view_config(route_name="gift_card_add_to_cart", request_method="POST", require_csrf=True)
 @user_required()
+@trial_active_required()
 def gift_card_add_to_cart(request):
     """Add a gift card to the active cart."""
     shop = request.shop
diff --git a/make_post_sell/views/product.py b/make_post_sell/views/product.py
index 1376e0f..2c018d3 100644
--- a/make_post_sell/views/product.py
+++ b/make_post_sell/views/product.py
@@ -5,6 +5,7 @@ from pyramid.httpexceptions import HTTPFound
 from . import (
     user_required,
     shop_editor_required,
+    trial_active_required,
     get_referer_or_home,
 )
 
@@ -132,6 +133,7 @@ def product(request):
 @view_config(route_name="content_new", renderer="content_new.j2")
 @user_required()
 @shop_editor_required()
+@trial_active_required()
 def product_new(request):
     title = request.params.get("title", "").strip()
     description = request.params.get("description", "").strip()
@@ -242,6 +244,7 @@ def product_edit_description(request):
 @view_config(route_name="content_edit", renderer="product_edit.j2")
 @view_config(route_name="content_edit2", renderer="product_edit.j2")
 @shop_editor_required()
+@trial_active_required()
 def product_edit(request):
     product_modified = False
     product = request.product
diff --git a/make_post_sell/views/shop.py b/make_post_sell/views/shop.py
index ee8b03a..551e847 100644
--- a/make_post_sell/views/shop.py
+++ b/make_post_sell/views/shop.py
@@ -1080,13 +1080,25 @@ def shop_settings(request):
             if env_value not in (0, 1, 2):
                 request.session.flash(("Invalid environment value.", "error"))
             elif env_value != 0 and shop.environment == 0:
-                # Changing from production to non-production
-                shop.environment = env_value
-                request.session.flash((
-                    f"Shop environment changed to {shop.environment_label}. "
-                    "This shop is now hidden from public search and feeds.",
-                    "success",
-                ))
+                # Changing from production to non-production — check allowance
+                # Count OTHER production shops (excluding this one being changed)
+                prod_count = sum(1 for s in request.user.shops if s.is_production and s.id != shop.id)
+                non_prod_count = sum(1 for s in request.user.shops if s.is_non_production)
+                # Each production shop allows 2 non-prod shops; minimum 2 non-prod allowed
+                max_non_prod = max(2, prod_count * 2)
+                if non_prod_count >= max_non_prod:
+                    request.session.flash((
+                        "You need more production shops before creating additional dev/stage shops. "
+                        "Each production shop includes 2 free dev/stage shops.",
+                        "error",
+                    ))
+                else:
+                    shop.environment = env_value
+                    request.session.flash((
+                        f"Shop environment changed to {shop.environment_label}. "
+                        "This shop is now hidden from public search and feeds.",
+                        "success",
+                    ))
             elif env_value == 0 and shop.environment != 0:
                 # Changing from non-production to production
                 shop.environment = env_value
@@ -1135,7 +1147,23 @@ def shop_settings(request):
                     shop.primary_s3_enabled = ps3_enabled
                     changed = True
                 if changed:
-                    request.session.flash(("Storage bucket settings updated.", "success"))
+                    if ps3_enabled:
+                        # Test connection when enabling
+                        try:
+                            import boto3
+                            test_client = boto3.session.Session().client(
+                                "s3",
+                                region_name=ps3_region,
+                                endpoint_url=ps3_endpoint,
+                                aws_access_key_id=ps3_access_key,
+                                aws_secret_access_key=ps3_secret_key,
+                            )
+                            test_client.list_objects_v2(Bucket=ps3_bucket, MaxKeys=0)
+                            request.session.flash(("Storage bucket settings saved and connection verified.", "success"))
+                        except Exception as e:
+                            request.session.flash((f"Bucket settings saved but connection test failed: {e}", "error"))
+                    else:
+                        request.session.flash(("Storage bucket settings updated.", "success"))
 
     # If we processed any form submission, respond accordingly
     if form_section: