diff --git a/CHANGELOG.md b/CHANGELOG.md
index eb2ec09f..5216f0ce 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -65,6 +65,43 @@ All notable changes to the IAB Tech Lab Seller Agent are documented here.
 
 ### Fixed
 
+- `GET /products` (and everything reading the catalog — avails, pricing,
+  package resolution, quotes) now reflects real ad-server inventory
+  (GAM/FreeWheel/S3) once synced, read fresh from storage on every call
+  (AI-6). Previously a synced product only ever lived in the flow's
+  per-request state and one worker process's in-memory cache
+  (`infra/docker/Dockerfile` runs uvicorn with `--workers 2`), so the
+  catalog kept serving the 13-product static default indefinitely after
+  a real sync — matching how synced packages already worked. Synced
+  products persist under a dedicated `synced_product:{id}` storage key,
+  separate from the generic `product:{id}` key `negotiation_service`
+  already writes for an unrelated reason (a narrower proposal-time
+  snapshot), and are pruned when no longer returned by the ad server.
+  CSV-mode and no-ad-server-configured behavior is unchanged.
+- `GAMRestClient` can now actually connect to Google Ad Manager (AI-6).
+  It called `googleapiclient.discovery.build("admanager", "v1", ...)`
+  with no `discoveryServiceUrl`; Ad Manager self-hosts its own discovery
+  document rather than registering with the generic, centrally-indexed
+  discovery service `build()` queries by default (and it isn't in the
+  locally bundled discovery docs either), so every real connection
+  attempt raised `UnknownApiNameOrVersion` and `sync_from_ad_server()`'s
+  broad exception handler silently degraded to mock synced packages —
+  real GAM inventory sync had never actually succeeded. Fixed by pointing
+  `build()` at `admanager.googleapis.com/$discovery/rest?version=v1`
+  directly (with `static_discovery=False`). Verified against a live GAM
+  network: `list_inventory()` returned 100 real ad units, a real
+  `POST /packages/sync` completed with zero warnings (previously always
+  warned and fell back to mocks), and `GET /products` served 50 real
+  synced products end to end.
+- `infra/docker/Dockerfile` now installs the `gam` extra alongside
+  `redis` (AI-6): the production image only ever installed
+  `".[redis]"`, so `googleads`/`google-api-python-client`/`google-auth`
+  were never present, and a real GAM deployment would `ImportError`
+  before ever reaching `GAMRestClient.connect()` — even with the
+  discovery-URL fix above in place. Rebuilt the real Dockerfile from
+  scratch with no manual installs and re-ran the full live-GAM
+  verification (100 real ad units, clean sync, 50 real products served)
+  to confirm the production image now works end to end unmodified.
 - MCP `list_packages` no longer crashes (AI-12). It constructed
   `MediaKitService()` with no arguments, but the class requires
   `storage`/`pricing_engine`, so every call raised `TypeError`
diff --git a/infra/docker/Dockerfile b/infra/docker/Dockerfile
index 8fcfe645..eb32f05e 100644
--- a/infra/docker/Dockerfile
+++ b/infra/docker/Dockerfile
@@ -27,7 +27,7 @@ ENV PATH="/opt/venv/bin:$PATH"
 
 COPY . .
 RUN pip install --no-cache-dir --upgrade pip \
-    && pip install --no-cache-dir -e ".[redis]"
+    && pip install --no-cache-dir -e ".[redis,gam]"
 
 # ---------------------------------------------------------------------------
 # Stage 2 — Runtime
diff --git a/src/ad_seller/clients/ad_server_base.py b/src/ad_seller/clients/ad_server_base.py
index 81c5dbe7..b07cf0e4 100644
--- a/src/ad_seller/clients/ad_server_base.py
+++ b/src/ad_seller/clients/ad_server_base.py
@@ -258,10 +258,14 @@ async def update_deal(
     async def list_inventory(
         self,
         *,
-        limit: int = 100,
+        limit: Optional[int] = 100,
         filter_str: Optional[str] = None,
     ) -> list[AdServerInventoryItem]:
-        """List ad units (GAM) or ad slots (FreeWheel)."""
+        """List ad units (GAM) or ad slots (FreeWheel).
+
+        Returns at most ``limit`` items; ``limit=None`` returns all of them
+        (a full sync needs the whole inventory, not the first page).
+        """
 
     # -- Audience Operations --
 
diff --git a/src/ad_seller/clients/csv_adapter.py b/src/ad_seller/clients/csv_adapter.py
index 2b0060e8..636f9399 100644
--- a/src/ad_seller/clients/csv_adapter.py
+++ b/src/ad_seller/clients/csv_adapter.py
@@ -262,7 +262,7 @@ async def disconnect(self) -> None:
     async def list_inventory(
         self,
         *,
-        limit: int = 100,
+        limit: Optional[int] = 100,
         filter_str: Optional[str] = None,
     ) -> list[AdServerInventoryItem]:
         """Read inventory.csv and return AdServerInventoryItem list."""
@@ -306,7 +306,7 @@ async def list_inventory(
             item.__dict__["raw"] = raw
 
             results.append(item)
-            if len(results) >= limit:
+            if limit is not None and len(results) >= limit:
                 break
 
         return results
diff --git a/src/ad_seller/clients/freewheel_adapter.py b/src/ad_seller/clients/freewheel_adapter.py
index 23a8e284..ca175afe 100644
--- a/src/ad_seller/clients/freewheel_adapter.py
+++ b/src/ad_seller/clients/freewheel_adapter.py
@@ -158,7 +158,7 @@ async def _reconnect_bc(self) -> None:
     async def list_inventory(
         self,
         *,
-        limit: int = 100,
+        limit: Optional[int] = 100,
         filter_str: Optional[str] = None,
     ) -> list[AdServerInventoryItem]:
         """List inventory from FreeWheel Streaming Hub.
diff --git a/src/ad_seller/clients/gam_adapter.py b/src/ad_seller/clients/gam_adapter.py
index f1b491d6..437a48c1 100644
--- a/src/ad_seller/clients/gam_adapter.py
+++ b/src/ad_seller/clients/gam_adapter.py
@@ -322,13 +322,29 @@ async def update_deal(
     async def list_inventory(
         self,
         *,
-        limit: int = 100,
+        limit: Optional[int] = 100,
         filter_str: Optional[str] = None,
     ) -> list[AdServerInventoryItem]:
-        ad_units, _ = await self._rest.list_ad_units(
-            page_size=limit,
-            filter_str=filter_str,
-        )
+        # The REST API caps a page at 100 ad units, so follow
+        # nextPageToken until it runs out or ``limit`` units are collected
+        # (``limit=None`` reads every page). Reading only the first page made
+        # a full sync see at most 100 units and prune the rest.
+        ad_units = []
+        page_token: Optional[str] = None
+        seen_tokens: set[str] = set()
+        while limit is None or len(ad_units) < limit:
+            page_size = 100 if limit is None else min(100, limit - len(ad_units))
+            page, page_token = await self._rest.list_ad_units(
+                page_size=page_size,
+                page_token=page_token,
+                filter_str=filter_str,
+            )
+            ad_units.extend(page)
+            if not page or not page_token or page_token in seen_tokens:
+                break
+            seen_tokens.add(page_token)
+        if limit is not None:
+            ad_units = ad_units[:limit]
 
         return [
             AdServerInventoryItem(
diff --git a/src/ad_seller/clients/gam_rest_client.py b/src/ad_seller/clients/gam_rest_client.py
index c2993696..48fb9586 100644
--- a/src/ad_seller/clients/gam_rest_client.py
+++ b/src/ad_seller/clients/gam_rest_client.py
@@ -79,11 +79,20 @@ async def connect(self) -> None:
                 scopes=["https://www.googleapis.com/auth/admanager"],
             )
 
-            # Build the GAM API service
+            # Build the GAM API service. Ad Manager self-hosts its own
+            # discovery document (admanager.googleapis.com/$discovery/rest)
+            # rather than registering with the generic, centrally-indexed
+            # discovery service googleapiclient.discovery.build() queries
+            # by default (and it isn't in the locally bundled discovery
+            # docs either) -- without an explicit discoveryServiceUrl this
+            # raises UnknownApiNameOrVersion for every real connection
+            # attempt, silently degrading every GAM sync to mock data.
             self._service = build(
                 "admanager",
                 "v1",
                 credentials=self._credentials,
+                discoveryServiceUrl="https://admanager.googleapis.com/$discovery/rest?version=v1",
+                static_discovery=False,
                 cache_discovery=False,
             )
         except ImportError:
diff --git a/src/ad_seller/clients/s3_csv_adapter.py b/src/ad_seller/clients/s3_csv_adapter.py
index f0f0bb4f..ece8a944 100644
--- a/src/ad_seller/clients/s3_csv_adapter.py
+++ b/src/ad_seller/clients/s3_csv_adapter.py
@@ -179,7 +179,7 @@ async def disconnect(self) -> None:
     async def list_inventory(
         self,
         *,
-        limit: int = 100,
+        limit: Optional[int] = 100,
         filter_str: Optional[str] = None,
     ) -> list[AdServerInventoryItem]:
         """List inventory items from S3 CSV files."""
diff --git a/src/ad_seller/flows/product_setup_flow.py b/src/ad_seller/flows/product_setup_flow.py
index 76a4af14..9421b5b4 100644
--- a/src/ad_seller/flows/product_setup_flow.py
+++ b/src/ad_seller/flows/product_setup_flow.py
@@ -163,8 +163,11 @@ async def sync_from_ad_server(self) -> None:
             # servers (GAM/FreeWheel), which is also where excluding
             # archived inventory by default actually matters.
             filter_str = "status:ACTIVE" if client.ad_server_type.value != "csv" else None
+            # limit=None: a sync must see the whole inventory. The default
+            # (100) silently truncated larger GAM networks and S3 files, and
+            # the prune below then deleted everything past the cut.
             async with client:
-                items = await client.list_inventory(filter_str=filter_str)
+                items = await client.list_inventory(filter_str=filter_str, limit=None)
 
             # Group items by inferred type and create packages
             grouped: dict[str, list] = {}
@@ -177,45 +180,99 @@ async def sync_from_ad_server(self) -> None:
             # Canonical item→product mapping lives in catalog_service
             # (shared with the CSV-mode catalog build) so flow-seeded
             # products and API catalog products cannot diverge.
-            from ..services.catalog_service import product_from_inventory_item
+            from ..services.catalog_service import (
+                persist_synced_product,
+                product_from_inventory_item,
+            )
 
-            for item in items:
-                product_def = product_from_inventory_item(item)
-                self.state.products[product_def.product_id] = product_def
+            # Persist every product, but roll back precisely on any
+            # mid-loop failure (AI-6 follow-up): persist_synced_product
+            # writes straight to storage per item, so without this, an
+            # exception on item N left items 1..N-1 of THIS run already
+            # committed even though the sync is reported as failed --
+            # contradicting the "a transient failure never silently
+            # changes the synced layer" invariant above. Capturing each
+            # key's PRIOR value (not just blindly deleting it) matters
+            # because a re-synced item can share an id with one already
+            # persisted by an earlier successful sync; rollback must
+            # restore that old value, not erase still-valid data.
+            persisted: list[tuple[str, Optional[dict[str, Any]]]] = []
+
+            async def _rollback_products() -> None:
+                for product_id, prior in persisted:
+                    key = f"synced_product:{product_id}"
+                    if prior is None:
+                        await storage.delete(key)
+                    else:
+                        await storage.set(key, prior)
+                    self.state.products.pop(product_id, None)
+
+            try:
+                for item in items:
+                    product_def = product_from_inventory_item(item)
+                    key = f"synced_product:{product_def.product_id}"
+                    prior = await storage.get(key)
+                    self.state.products[product_def.product_id] = product_def
+                    # Persist so GET /products reflects real ad-server inventory
+                    # for non-CSV ad servers too (AI-6) -- state.products alone
+                    # is per-flow-instance and never outlives this request.
+                    await persist_synced_product(product_def)
+                    persisted.append((product_def.product_id, prior))
+            except Exception:
+                await _rollback_products()
+                raise
 
             logger.info("Created %d products from ad server inventory", len(self.state.products))
 
-            for inv_type, inv_items in grouped.items():
-                ad_formats = self._classify_ad_formats_from_type(inv_type)
-                device_types = self._classify_device_types_from_type(inv_type)
-                base_cpm = self._estimate_base_cpm(inv_type)
-
-                package = Package(
-                    package_id=self._synced_package_id(client.ad_server_type.value, inv_type),
-                    name=f"{inv_type.replace('_', ' ').title()} - Synced",
-                    description=f"Synced {inv_type} inventory ({len(inv_items)} ad units)",
-                    layer=PackageLayer.SYNCED,
-                    status=PackageStatus.ACTIVE,
-                    placements=[
-                        PackagePlacement(
-                            product_id=item.id,
-                            product_name=item.name,
-                            ad_formats=ad_formats,
-                            device_types=device_types,
-                        )
-                        for item in inv_items
-                    ],
-                    ad_formats=ad_formats,
-                    device_types=device_types,
-                    cat=["IAB1"],
-                    cattax=2,
-                    base_price=base_cpm,
-                    floor_price=round(base_cpm * 0.7, 2),
-                    ad_server_source=client.ad_server_type.value,
-                    is_featured=inv_type == "ctv",
-                )
-
-                await self._upsert_synced_package(storage, package)
+            # Same rollback discipline for packages -- identical risk,
+            # same fix shape.
+            upserted: list[tuple[str, Optional[dict[str, Any]]]] = []
+            try:
+                for inv_type, inv_items in grouped.items():
+                    ad_formats = self._classify_ad_formats_from_type(inv_type)
+                    device_types = self._classify_device_types_from_type(inv_type)
+                    base_cpm = self._estimate_base_cpm(inv_type)
+
+                    package = Package(
+                        package_id=self._synced_package_id(client.ad_server_type.value, inv_type),
+                        name=f"{inv_type.replace('_', ' ').title()} - Synced",
+                        description=f"Synced {inv_type} inventory ({len(inv_items)} ad units)",
+                        layer=PackageLayer.SYNCED,
+                        status=PackageStatus.ACTIVE,
+                        placements=[
+                            PackagePlacement(
+                                product_id=item.id,
+                                product_name=item.name,
+                                ad_formats=ad_formats,
+                                device_types=device_types,
+                            )
+                            for item in inv_items
+                        ],
+                        ad_formats=ad_formats,
+                        device_types=device_types,
+                        cat=["IAB1"],
+                        cattax=2,
+                        base_price=base_cpm,
+                        floor_price=round(base_cpm * 0.7, 2),
+                        ad_server_source=client.ad_server_type.value,
+                        is_featured=inv_type == "ctv",
+                    )
+
+                    prior_package = await storage.get_package(package.package_id)
+                    await self._upsert_synced_package(storage, package)
+                    upserted.append((package.package_id, prior_package))
+            except Exception:
+                for package_id, prior_package in upserted:
+                    if prior_package is None:
+                        await storage.delete_package(package_id)
+                    else:
+                        await storage.set_package(package_id, prior_package)
+                    if package_id in self.state.synced_segments:
+                        self.state.synced_segments.remove(package_id)
+                # The products loop already committed this run's products;
+                # a failed run must not leave them behind either.
+                await _rollback_products()
+                raise
 
             logger.info("Synced %d packages from ad server", len(grouped))
 
@@ -230,14 +287,40 @@ async def sync_from_ad_server(self) -> None:
             self.state.ad_server_sync_failed = True
             return
 
-        await self._finish_sync()
-
-    async def _finish_sync(self) -> None:
-        """Terminal step of every sync path: prune the stale synced layer."""
+        await self._finish_sync(real_sync_completed=True)
+
+    async def _finish_sync(self, real_sync_completed: bool = False) -> None:
+        """Terminal step of a sync that reflects current state: prune the stale synced layer.
+
+        Runs on exactly two paths: the mock path (no ad server configured,
+        so mocks are the honest steady state) and the real ad-server
+        success path. It deliberately does NOT run when the real sync
+        raises: ``sync_from_ad_server`` returns before reaching it, because
+        a failed run leaves ``self.state.synced_segments`` /
+        ``self.state.products`` empty and the prune would then delete every
+        real package and product from the last successful sync. A transient
+        failure therefore keeps serving the last-good synced catalog, and
+        the next successful sync converges it (issue #34). Do not "fix"
+        the failure path to call this.
+
+        ``real_sync_completed`` is only ever True from the real ad-server
+        success path, never the mock path -- a zero-item real sync is a
+        genuine "no inventory right now" result that must be served as an
+        honestly empty catalog, not mistaken for "never synced" (AI-6
+        follow-up); the mock path has no such distinction to make, since
+        it never persists ``synced_product:*`` entries at all.
+        """
+        from ..services.catalog_service import (
+            mark_synced_catalog_completed,
+            prune_stale_synced_products,
+        )
         from ..storage.factory import get_storage
 
         storage = await get_storage()
         await self._prune_stale_synced_packages(storage)
+        await prune_stale_synced_products(set(self.state.products.keys()))
+        if real_sync_completed:
+            await mark_synced_catalog_completed()
 
     async def _create_mock_synced_packages(self) -> None:
         """Create mock Layer 1 packages for development without ad server creds."""
diff --git a/src/ad_seller/interfaces/api/deps.py b/src/ad_seller/interfaces/api/deps.py
index 91a5d019..87a1ccbc 100644
--- a/src/ad_seller/interfaces/api/deps.py
+++ b/src/ad_seller/interfaces/api/deps.py
@@ -238,16 +238,19 @@ async def _verified_buyer_context(
 
 
 async def get_product_catalog() -> dict[str, Any]:
-    """Return the cached static product catalog.
+    """Return the product catalog.
 
     Resolves through ``interfaces.api.main._get_static_product_catalog``
     AT CALL TIME so existing tests that patch that attribute (and reset
     ``main._STATIC_PRODUCT_CATALOG``) keep governing every endpoint.
     The lazy import avoids the main ↔ routers circular import.
 
-    Async since the AI-14 follow-up: any stored inventory-type override
-    is applied inside ``catalog_service.get_static_product_catalog()``
-    itself, which needs to consult storage.
+    Async for two reasons that both land in
+    ``catalog_service.get_static_product_catalog()``: since AI-6, when a
+    real ad server is configured and synced, this reads persisted products
+    from storage on every call rather than an in-memory cache; since the
+    AI-14 follow-up, any stored inventory-type override is applied inside
+    that same function, which also needs to consult storage.
     """
     from . import main
 
diff --git a/src/ad_seller/interfaces/api/main.py b/src/ad_seller/interfaces/api/main.py
index 131e5250..a3800c24 100644
--- a/src/ad_seller/interfaces/api/main.py
+++ b/src/ad_seller/interfaces/api/main.py
@@ -174,13 +174,16 @@ async def lifespan(application):
 # patchable name, and endpoint code resolves the catalog through it at
 # call time (see deps.get_product_catalog).
 #
-# This no longer CACHES the catalog itself — an inventory-type override
-# must be reflected live on every call (AI-14 follow-up), and caching the
-# dict here would silently pin it to whatever the first call returned.
+# This no longer CACHES the catalog itself — two reasons, both requiring a
+# live read on every call rather than a dict pinned to whatever the first
+# call returned: AI-6, a synced ad server's catalog must be read live from
+# storage (see catalog_service.get_static_product_catalog); the AI-14
+# follow-up, an inventory-type override must be reflected live too.
 # `_STATIC_PRODUCT_CATALOG` is kept only as the legacy test-reset signal:
 # setting it to None forces one `catalog_service.reset_catalog_cache()`
 # call, which existing test fixtures rely on for a deterministic rebuild
-# between tests.
+# between tests (CSV/static catalogs are still cached inside
+# catalog_service itself).
 _STATIC_PRODUCT_CATALOG: Optional[dict[str, Any]] = None
 
 
diff --git a/src/ad_seller/interfaces/api/routers/media_kit.py b/src/ad_seller/interfaces/api/routers/media_kit.py
index 247f677c..b24610bc 100644
--- a/src/ad_seller/interfaces/api/routers/media_kit.py
+++ b/src/ad_seller/interfaces/api/routers/media_kit.py
@@ -173,8 +173,9 @@ async def search_media_kit(
 async def _resolve_package_products(product_ids: list[str], storage):
     """Resolve product ids to ProductDefinitions, catalog-first.
 
-    ``GET /products`` serves the in-memory static catalog
-    (``deps.get_product_catalog()``), which is never persisted to storage —
+    ``GET /products`` serves ``deps.get_product_catalog()``: the static or
+    CSV catalog, or (since AI-6) the synced ad-server catalog stored under
+    ``synced_product:*`` keys, none of which ``storage.get_product`` reads —
     so package endpoints must resolve ids against that same catalog first,
     falling back to ``storage.get_product`` for products persisted by the
     negotiation flow (issue #34: storage-only resolution silently produced
diff --git a/src/ad_seller/interfaces/cli/main.py b/src/ad_seller/interfaces/cli/main.py
index 274f0f21..f3ba4c79 100644
--- a/src/ad_seller/interfaces/cli/main.py
+++ b/src/ad_seller/interfaces/cli/main.py
@@ -54,7 +54,8 @@ def init(
     # Read the default catalog from the single cached service source (EP-3.2)
     # rather than spinning up ProductSetupFlow / an OpenDirect MCP session.
     # Bridged via _run_blocking since this Typer command is synchronous but
-    # the catalog accessor is async (applies inventory-type overrides).
+    # the catalog accessor is async (AI-6: live reads for synced ad servers;
+    # also applies inventory-type overrides).
     products = _run_blocking(catalog_service.get_static_product_catalog())["products"]
 
     console.print(f"[green]✓[/green] Organization '{organization_name}' initialized")
diff --git a/src/ad_seller/services/catalog_service.py b/src/ad_seller/services/catalog_service.py
index d691608b..4a0f8630 100644
--- a/src/ad_seller/services/catalog_service.py
+++ b/src/ad_seller/services/catalog_service.py
@@ -364,10 +364,13 @@ def build_csv_product_catalog() -> dict[str, Any]:
 
     data_dir = get_settings().csv_data_dir
 
+    # limit=None: serve every row. The adapter default (100) silently
+    # dropped the rest, while the sync (which reads every row) still built
+    # packages that pointed at them.
     async def _load() -> list[Any]:
         client = CSVAdServerClient(data_dir=data_dir)
         async with client:
-            return await client.list_inventory()
+            return await client.list_inventory(limit=None)
 
     try:
         items = _run_blocking(_load())
@@ -415,37 +418,178 @@ def build_static_product_catalog() -> dict[str, Any]:
     }
 
 
+def _real_ad_server_configured(settings: Any) -> bool:
+    """True when a real (non-CSV) ad server is configured (AI-6).
+
+    Mirrors ``ProductSetupFlow.sync_from_ad_server``'s own "is this a real
+    ad server, not the mock/dev fallback" condition, minus the CSV case
+    (``_csv_mode_active`` handles that separately).
+    """
+    return bool(
+        settings.gam_network_code
+        or settings.freewheel_sh_mcp_url
+        or settings.ad_server_type == "s3"
+    )
+
+
+async def persist_synced_product(product: Any) -> None:
+    """Persist a product built from a real ad-server sync (AI-6).
+
+    Stored under a dedicated ``synced_product:{id}`` key, deliberately
+    NOT the generic ``product:{id}`` key ``negotiation_service`` and
+    ``media_kit_service`` already write/read for an unrelated reason (a
+    per-product snapshot at proposal time, in a narrower shape via
+    ``serialize_product``, so a later cold lookup doesn't 404). Reusing
+    that key would make :func:`build_synced_product_catalog` treat an
+    incidental negotiation snapshot of a STATIC-catalog product as
+    evidence a real ad-server sync had happened, and ``ProductDefinition
+    (**data)`` would likely fail to reconstruct from the narrower
+    snapshot shape anyway.
+    """
+    from ..storage.factory import get_storage
+
+    storage = await get_storage()
+    await storage.set(f"synced_product:{product.product_id}", product.model_dump(mode="json"))
+
+
+async def prune_stale_synced_products(seen_product_ids: set[str]) -> None:
+    """Delete synced products not re-seeded by the current sync run (AI-6).
+
+    Mirrors ``ProductSetupFlow._prune_stale_synced_packages`` (issue #34):
+    only products actually returned by the ad server on this run survive,
+    so a product removed from the ad server stops being served instead of
+    lingering forever as phantom inventory.
+    """
+    from ..storage.factory import get_storage
+
+    storage = await get_storage()
+    for key in await storage.keys("synced_product:*"):
+        product_id = key.removeprefix("synced_product:")
+        if product_id not in seen_product_ids:
+            await storage.delete(key)
+            logger.info("Pruned stale synced product: %s", product_id)
+
+
+_SYNCED_CATALOG_COMPLETED_KEY = "synced_catalog:completed"
+
+
+async def mark_synced_catalog_completed() -> None:
+    """Record that a real ad-server sync has completed at least once (AI-6).
+
+    A zero-item sync is a real, successful sync — the seller's ad server
+    genuinely has no (matching) inventory right now — and must be told
+    apart from "no sync has ever run": both leave zero ``synced_product:*``
+    keys, but only the latter should fall back to the static demo catalog.
+    Set once a real sync's item loop completes, even with zero items;
+    never cleared (a seller who has synced once stays in "trust storage,
+    not the static fallback" mode from then on).
+    """
+    from ..storage.factory import get_storage
+
+    storage = await get_storage()
+    await storage.set(_SYNCED_CATALOG_COMPLETED_KEY, True)
+
+
+async def build_synced_product_catalog() -> Optional[dict[str, Any]]:
+    """Build the catalog from persisted ad-server-synced products (AI-6).
+
+    Returns ``None`` only when no real ad-server sync has ever run — no
+    ``synced_product:*`` keys exist AND no sync has ever completed — so
+    the caller falls back to the static default catalog — the same
+    "honest, never fabricated" fallback CSV mode already uses on a read
+    failure.
+
+    Either signal alone is enough evidence of a real sync:
+    - ``synced_product:*`` keys may exist even mid-sync, before
+      ``mark_synced_catalog_completed`` runs (each item is persisted as
+      it's processed, in a loop that precedes the completion marker —
+      see ``ProductSetupFlow.sync_from_ad_server``), including the case
+      where a sync crashed partway through.
+    - the completion marker may be set with zero keys when a *completed*
+      sync genuinely found no inventory — a real, honest result that
+      must be served as an empty catalog (``products: {}``), not mistaken
+      for "never synced" and backfilled with stale demo products.
+    """
+    from ..models.flow_state import ProductDefinition
+    from ..storage.factory import get_storage
+
+    storage = await get_storage()
+    keys = await storage.keys("synced_product:*")
+    if not keys and not await storage.exists(_SYNCED_CATALOG_COMPLETED_KEY):
+        return None
+
+    products: dict[str, Any] = {}
+    for key in keys:
+        data = await storage.get(key)
+        if not data:
+            continue
+        product_def = ProductDefinition(**data)
+        products[product_def.product_id] = product_def
+
+    return {
+        "products": products,
+        "inventory_types": sorted({p.inventory_type for p in products.values()}),
+    }
+
+
 async def get_static_product_catalog() -> dict[str, Any]:
     """Return the seller's product catalog without running the flow.
 
     In CSV mode (``AD_SERVER_TYPE=csv``) the catalog is built from the CSV
-    inventory; in every other mode it is the static default catalog
-    (byte-identical to the pre-CSV-wiring behavior).
-
-    The underlying product list is cached — repeated reads return stable
-    product_ids (issue #34) — but any stored inventory-type override is
-    applied fresh on every call via :func:`apply_inventory_type_overrides_batch`.
-    This is now the ONE place every consumer (REST, MCP, CLI, chat,
-    negotiation, quote/pricing) sees an override applied. Previously only
-    two REST routes applied it themselves, so `GET /products` could show
-    an overridden type while `POST /quotes` priced off the original one —
-    a real mispricing, not just a display inconsistency (AI-14 follow-up).
-    The cache itself is never mutated: when nothing is overridden, the
-    identical cached dict is returned (no allocation, and `is` identity
-    across calls is preserved); when something is, a shallow copy with
-    the overridden products substituted is returned instead.
+    inventory, cached once per process (issue #34).
+
+    When a real ad server (GAM/FreeWheel/S3) is configured AND has been
+    synced at least once, the catalog reflects those real, persisted
+    products (AI-6) — read fresh from storage on EVERY call, never
+    cached, so every worker process agrees immediately after a sync
+    (matching how packages already work) instead of only whichever
+    worker happened to run the sync seeing it. Storage reads are cheap
+    local I/O, not a live ad-server call, so this doesn't reintroduce the
+    per-request hang risk that CSV-only caching was built to avoid.
+
+    Otherwise (no ad server configured, or configured but never yet
+    synced) the static default catalog is served, cached once, exactly
+    as before.
+
+    Whichever source produced the catalog dict, any stored inventory-type
+    override is applied fresh on every call via
+    :func:`apply_inventory_type_overrides_batch` (AI-14 follow-up). This is
+    the ONE place every consumer (REST, MCP, CLI, chat, negotiation,
+    quote/pricing) sees an override applied — previously only two REST
+    routes applied it themselves, so `GET /products` could show an
+    overridden type while `POST /quotes` priced off the original one, a
+    real mispricing, not just a display inconsistency. A cached source
+    (CSV or static) is never mutated: when nothing is overridden, the
+    identical cached dict is returned (no allocation, `is` identity
+    preserved); when something is, a shallow copy with the overridden
+    products substituted is returned instead. The synced-from-real-ad-server
+    source is never cached in the first place, so overrides apply there the
+    same way.
     """
     global _CATALOG_CACHE
-    if _CATALOG_CACHE is None:
-        if _csv_mode_active():
+
+    if _csv_mode_active():
+        if _CATALOG_CACHE is None:
             _CATALOG_CACHE = build_csv_product_catalog()
-        else:
-            _CATALOG_CACHE = build_static_product_catalog()
+        catalog = _CATALOG_CACHE
+    else:
+        from ..config import get_settings
 
-    products = await apply_inventory_type_overrides_batch(_CATALOG_CACHE["products"])
-    if products is _CATALOG_CACHE["products"]:
-        return _CATALOG_CACHE
-    return {**_CATALOG_CACHE, "products": products}
+        synced = None
+        if _real_ad_server_configured(get_settings()):
+            synced = await build_synced_product_catalog()
+
+        if synced is not None:
+            catalog = synced
+        else:
+            if _CATALOG_CACHE is None:
+                _CATALOG_CACHE = build_static_product_catalog()
+            catalog = _CATALOG_CACHE
+
+    products = await apply_inventory_type_overrides_batch(catalog["products"])
+    if products is catalog["products"]:
+        return catalog
+    return {**catalog, "products": products}
 
 
 def reset_catalog_cache() -> None:
diff --git a/tests/unit/test_csv_adapter.py b/tests/unit/test_csv_adapter.py
index 38d63198..c18fc106 100644
--- a/tests/unit/test_csv_adapter.py
+++ b/tests/unit/test_csv_adapter.py
@@ -244,6 +244,12 @@ async def test_applies_limit(self, sample_client: CSVAdServerClient) -> None:
         items = await sample_client.list_inventory(limit=3)
         assert len(items) == 3
 
+    async def test_limit_none_returns_every_row(self, sample_client: CSVAdServerClient) -> None:
+        """limit=None (what a full sync passes) returns all rows."""
+        everything = await sample_client.list_inventory(limit=10_000)
+        items = await sample_client.list_inventory(limit=None)
+        assert len(items) == len(everything) > 3
+
     async def test_applies_filter_str(self, sample_client: CSVAdServerClient) -> None:
         """filter_str filters by substring match on name (case-insensitive)."""
         items = await sample_client.list_inventory(filter_str="sports")
diff --git a/tests/unit/test_csv_catalog_coherence.py b/tests/unit/test_csv_catalog_coherence.py
index 37b9e98e..0bc2bf17 100644
--- a/tests/unit/test_csv_catalog_coherence.py
+++ b/tests/unit/test_csv_catalog_coherence.py
@@ -203,6 +203,22 @@ async def test_csv_catalog_ids_deterministic_across_rebuild(self, csv_mode):
         second_ids = set((await catalog_service.get_static_product_catalog())["products"])
         assert first_ids == second_ids == CSV_IDS
 
+    async def test_csv_catalog_serves_every_row_past_the_adapter_default(self, csv_mode):
+        """The adapter's default limit is 100; the catalog must not inherit
+        it, or rows past 100 404 while the sync still packages them."""
+        rows = [
+            f"inv-bulk-{i:03d},Bulk Display {i:03d},,ACTIVE,300x250,banner,2,display,"
+            "IAB1,10.00,USD,US,Bulk row"
+            for i in range(150)
+        ]
+        (csv_mode / "inventory.csv").write_text(CSV_HEADER + "\n" + "\n".join(rows) + "\n")
+        _reset_caches()
+
+        catalog = await catalog_service.get_static_product_catalog()
+
+        assert len(catalog["products"]) == 150
+        assert "inv-bulk-149" in catalog["products"]
+
 
 # =============================================================================
 # Non-CSV modes pinned byte-identical
diff --git a/tests/unit/test_gam_clients.py b/tests/unit/test_gam_clients.py
index 4da99a24..d0a34ce7 100644
--- a/tests/unit/test_gam_clients.py
+++ b/tests/unit/test_gam_clients.py
@@ -339,3 +339,73 @@ def test_cpv_not_supported(self):
         """Test CPV is not a valid GAM cost type."""
         valid_types = [e.value for e in GAMCostType]
         assert "CPV" not in valid_types
+
+
+class TestGAMAdapterListInventoryPaging:
+    """GAMAdServerClient.list_inventory must follow nextPageToken.
+
+    The REST API returns at most 100 ad units per page. Reading only the
+    first page made a sync of a larger network see 100 units and prune the
+    rest.
+    """
+
+    @staticmethod
+    def _adapter(total_units: int, repeat_token: bool = False):
+        from types import SimpleNamespace
+
+        from ad_seller.clients.gam_adapter import GAMAdServerClient
+
+        units = [
+            SimpleNamespace(
+                id=f"u{i}", name=f"Unit {i}", parent_id=None, status="ACTIVE", ad_unit_sizes=[]
+            )
+            for i in range(total_units)
+        ]
+        calls: list[dict] = []
+
+        async def list_ad_units(page_size=100, page_token=None, filter_str=None):
+            calls.append({"page_size": page_size, "page_token": page_token})
+            start = int(page_token or 0)
+            end = start + min(page_size, 100)
+            if repeat_token:
+                return units[start:end], "0"
+            return units[start:end], str(end) if end < len(units) else None
+
+        with (
+            patch("ad_seller.clients.gam_adapter.GAMSoapClient"),
+            patch("ad_seller.clients.gam_adapter.GAMRestClient"),
+        ):
+            adapter = GAMAdServerClient()
+        adapter._rest.list_ad_units = list_ad_units
+        return adapter, calls
+
+    async def test_limit_none_reads_every_page(self):
+        adapter, calls = self._adapter(250)
+
+        items = await adapter.list_inventory(limit=None, filter_str="status:ACTIVE")
+
+        assert [i.id for i in items] == [f"u{i}" for i in range(250)]
+        assert [c["page_token"] for c in calls] == [None, "100", "200"]
+
+    async def test_limit_caps_across_pages(self):
+        adapter, calls = self._adapter(250)
+
+        items = await adapter.list_inventory(limit=120)
+
+        assert len(items) == 120
+        assert [c["page_size"] for c in calls] == [100, 20]
+
+    async def test_default_limit_still_returns_one_page(self):
+        adapter, calls = self._adapter(250)
+
+        items = await adapter.list_inventory()
+
+        assert len(items) == 100
+        assert len(calls) == 1
+
+    async def test_repeated_page_token_stops_instead_of_looping(self):
+        adapter, calls = self._adapter(250, repeat_token=True)
+
+        await adapter.list_inventory(limit=None)
+
+        assert len(calls) == 2
diff --git a/tests/unit/test_inventory_sync_persistence.py b/tests/unit/test_inventory_sync_persistence.py
index 7bd82eca..6df4545b 100644
--- a/tests/unit/test_inventory_sync_persistence.py
+++ b/tests/unit/test_inventory_sync_persistence.py
@@ -85,7 +85,7 @@ async def __aenter__(self):
     async def __aexit__(self, *exc):
         return False
 
-    async def list_inventory(self, filter_str=None):
+    async def list_inventory(self, filter_str=None, limit=None):
         return self._items
 
 
@@ -252,7 +252,7 @@ async def __aenter__(self):
     async def __aexit__(self, *exc):
         return False
 
-    async def list_inventory(self, filter_str=None):
+    async def list_inventory(self, filter_str=None, limit=None):
         self.received_filter_str = filter_str
         return self._items
 
@@ -275,7 +275,7 @@ async def __aenter__(self):
     async def __aexit__(self, *exc):
         return False
 
-    async def list_inventory(self, filter_str=None):
+    async def list_inventory(self, filter_str=None, limit=None):
         raise self._error
 
 
diff --git a/tests/unit/test_issue34_catalog_fixes.py b/tests/unit/test_issue34_catalog_fixes.py
index 85c60dce..4147446c 100644
--- a/tests/unit/test_issue34_catalog_fixes.py
+++ b/tests/unit/test_issue34_catalog_fixes.py
@@ -358,7 +358,7 @@ async def __aenter__(self):
     async def __aexit__(self, *exc):
         return False
 
-    async def list_inventory(self, filter_str=None):
+    async def list_inventory(self, filter_str=None, limit=None):
         return self._items
 
 
diff --git a/tests/unit/test_synced_product_catalog.py b/tests/unit/test_synced_product_catalog.py
new file mode 100644
index 00000000..746328df
--- /dev/null
+++ b/tests/unit/test_synced_product_catalog.py
@@ -0,0 +1,553 @@
+# Author: Green Mountain Systems AI Inc.
+# Donated to IAB Tech Lab
+
+"""AI-6: real, storage-backed catalog for synced ad servers.
+
+Defect: ``GET /products`` (and everything reading the catalog) served the
+static 13-product default catalog even after a real ad-server sync
+(GAM/FreeWheel/S3) had run. ``ProductSetupFlow.state.products`` is
+per-flow-instance and never outlives the request, and the catalog
+service's in-memory cache is per-worker-process (``infra/docker/Dockerfile``
+runs uvicorn with ``--workers 2``), so at most one of two worker processes
+ever saw synced data.
+
+Fix under test: synced products are persisted to a dedicated
+``synced_product:{id}`` storage key -- deliberately NOT the generic
+``product:{id}`` key ``negotiation_service``/``media_kit_service`` already
+write/read for an unrelated reason (a proposal-time snapshot, narrower
+shape) -- and served fresh from storage on every catalog read when a real
+ad server is configured, matching how packages already work (issue #34).
+"""
+
+from types import SimpleNamespace
+from typing import Any, Optional
+from unittest.mock import patch
+
+import pytest
+
+from ad_seller.models.core import DealType, PricingModel
+from ad_seller.models.flow_state import ProductDefinition
+from ad_seller.services import catalog_service
+from ad_seller.storage.base import StorageBackend
+
+
+class InMemoryStorage(StorageBackend):
+    """Fully in-memory storage backend for tests."""
+
+    def __init__(self) -> None:
+        self._data: dict[str, Any] = {}
+
+    async def connect(self) -> None:
+        pass
+
+    async def disconnect(self) -> None:
+        self._data.clear()
+
+    async def get(self, key: str) -> Optional[Any]:
+        return self._data.get(key)
+
+    async def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
+        self._data[key] = value
+
+    async def delete(self, key: str) -> bool:
+        return self._data.pop(key, None) is not None
+
+    async def exists(self, key: str) -> bool:
+        return key in self._data
+
+    async def keys(self, pattern: str = "*") -> list[str]:
+        import fnmatch
+
+        return [k for k in self._data if fnmatch.fnmatch(k, pattern)]
+
+
+@pytest.fixture
+def storage() -> InMemoryStorage:
+    return InMemoryStorage()
+
+
+@pytest.fixture(autouse=True)
+def _reset_catalog_cache():
+    """Reset the static-catalog cache between tests (issue #34 convention)."""
+    catalog_service.reset_catalog_cache()
+    yield
+    catalog_service.reset_catalog_cache()
+
+
+def _product(
+    product_id: str = "prod-a",
+    inventory_type: str = "display",
+    base_cpm: float = 10.0,
+) -> ProductDefinition:
+    return ProductDefinition(
+        product_id=product_id,
+        name=f"Product {product_id}",
+        inventory_type=inventory_type,
+        supported_deal_types=[DealType.PREFERRED_DEAL],
+        supported_pricing_models=[PricingModel.CPM],
+        base_cpm=base_cpm,
+        floor_cpm=round(base_cpm * 0.85, 2),
+    )
+
+
+# =============================================================================
+# persist_synced_product / prune_stale_synced_products
+# =============================================================================
+
+
+class TestPersistSyncedProduct:
+    async def test_writes_to_dedicated_synced_product_key(self, storage):
+        product = _product("prod-a")
+        with patch("ad_seller.storage.factory.get_storage", return_value=storage):
+            await catalog_service.persist_synced_product(product)
+
+        assert await storage.exists("synced_product:prod-a")
+        # Never the generic key negotiation_service/media_kit_service use.
+        assert not await storage.exists("product:prod-a")
+
+    async def test_does_not_treat_negotiation_snapshot_as_a_sync(self, storage):
+        """Storage-key-namespace isolation: an unrelated ``product:{id}``
+        write (negotiation_service's proposal-time snapshot, via
+        ``serialize_product``'s narrower 7-field shape) must never be read
+        as evidence a real ad-server sync happened."""
+        await storage.set(
+            "product:prod-b",
+            {"product_id": "prod-b", "name": "Snapshot only", "inventory_type": "display"},
+        )
+
+        with patch("ad_seller.storage.factory.get_storage", return_value=storage):
+            catalog = await catalog_service.build_synced_product_catalog()
+
+        assert catalog is None, "a generic product:{id} snapshot must not count as a sync"
+
+
+class TestPruneStaleSyncedProducts:
+    async def test_removes_products_not_in_seen_set(self, storage):
+        await storage.set("synced_product:keep", _product("keep").model_dump(mode="json"))
+        await storage.set("synced_product:drop", _product("drop").model_dump(mode="json"))
+
+        with patch("ad_seller.storage.factory.get_storage", return_value=storage):
+            await catalog_service.prune_stale_synced_products({"keep"})
+
+        assert await storage.exists("synced_product:keep")
+        assert not await storage.exists("synced_product:drop")
+
+    async def test_noop_when_all_seen(self, storage):
+        await storage.set("synced_product:a", _product("a").model_dump(mode="json"))
+
+        with patch("ad_seller.storage.factory.get_storage", return_value=storage):
+            await catalog_service.prune_stale_synced_products({"a"})
+
+        assert await storage.exists("synced_product:a")
+
+
+# =============================================================================
+# build_synced_product_catalog
+# =============================================================================
+
+
+class TestBuildSyncedProductCatalog:
+    async def test_returns_none_when_nothing_synced(self, storage):
+        with patch("ad_seller.storage.factory.get_storage", return_value=storage):
+            assert await catalog_service.build_synced_product_catalog() is None
+
+    async def test_reconstructs_products_from_storage(self, storage):
+        product = _product("prod-x", inventory_type="ctv", base_cpm=30.0)
+        await storage.set("synced_product:prod-x", product.model_dump(mode="json"))
+
+        with patch("ad_seller.storage.factory.get_storage", return_value=storage):
+            catalog = await catalog_service.build_synced_product_catalog()
+
+        assert catalog is not None
+        assert set(catalog["products"]) == {"prod-x"}
+        assert catalog["products"]["prod-x"].base_cpm == 30.0
+        assert catalog["inventory_types"] == ["ctv"]
+
+
+# =============================================================================
+# _real_ad_server_configured
+# =============================================================================
+
+
+class TestRealAdServerConfigured:
+    @pytest.mark.parametrize(
+        "kwargs,expected",
+        [
+            (
+                {
+                    "gam_network_code": "12345",
+                    "freewheel_sh_mcp_url": None,
+                    "ad_server_type": "none",
+                },
+                True,
+            ),
+            (
+                {
+                    "gam_network_code": None,
+                    "freewheel_sh_mcp_url": "https://fw.example",
+                    "ad_server_type": "none",
+                },
+                True,
+            ),
+            (
+                {"gam_network_code": None, "freewheel_sh_mcp_url": None, "ad_server_type": "s3"},
+                True,
+            ),
+            (
+                {"gam_network_code": None, "freewheel_sh_mcp_url": None, "ad_server_type": "csv"},
+                False,
+            ),
+            (
+                {"gam_network_code": None, "freewheel_sh_mcp_url": None, "ad_server_type": "none"},
+                False,
+            ),
+        ],
+    )
+    def test_matrix(self, kwargs, expected):
+        settings = SimpleNamespace(**kwargs)
+        assert catalog_service._real_ad_server_configured(settings) is expected
+
+
+# =============================================================================
+# get_static_product_catalog — the live-read invariant
+# =============================================================================
+
+
+class TestGetStaticProductCatalogLiveRead:
+    async def test_serves_synced_catalog_live_every_call_not_cached(self, storage):
+        """The core AI-6 invariant: a product persisted AFTER the first
+        catalog read must appear on the very next read -- no restart, no
+        explicit cache-reset call -- proving reads are live, not cached."""
+        settings = SimpleNamespace(
+            gam_network_code="12345", freewheel_sh_mcp_url=None, ad_server_type="none"
+        )
+
+        with (
+            patch("ad_seller.storage.factory.get_storage", return_value=storage),
+            patch("ad_seller.config.get_settings", return_value=settings),
+        ):
+            first = await catalog_service.get_static_product_catalog()
+
+            await catalog_service.persist_synced_product(
+                _product("prod-live", inventory_type="native", base_cpm=42.0)
+            )
+
+            second = await catalog_service.get_static_product_catalog()
+
+        # Before any sync: honest fallback to the static default catalog.
+        assert len(first["products"]) == len(catalog_service.DEFAULT_PRODUCT_CONFIGS)
+        # Immediately after persisting, with NO cache reset in between.
+        assert set(second["products"]) == {"prod-live"}
+        assert second["products"]["prod-live"].base_cpm == 42.0
+
+    async def test_non_real_ad_server_ignores_synced_products(self, storage):
+        """A synced_product key with no real ad server configured (e.g. a
+        stale key from a prior config) must not leak into the served
+        catalog -- only CSV/static-default apply."""
+        settings = SimpleNamespace(
+            gam_network_code=None, freewheel_sh_mcp_url=None, ad_server_type="none"
+        )
+        await storage.set("synced_product:orphan", _product("orphan").model_dump(mode="json"))
+
+        with (
+            patch("ad_seller.storage.factory.get_storage", return_value=storage),
+            patch("ad_seller.config.get_settings", return_value=settings),
+        ):
+            catalog = await catalog_service.get_static_product_catalog()
+
+        assert "orphan" not in catalog["products"]
+        assert len(catalog["products"]) == len(catalog_service.DEFAULT_PRODUCT_CONFIGS)
+
+
+# =============================================================================
+# End-to-end: ProductSetupFlow.sync_from_ad_server persists + prunes
+# =============================================================================
+
+
+class _FakeInventoryItem:
+    def __init__(self, id: str, name: str, raw: Optional[dict] = None):
+        self.id = id
+        self.name = name
+        self.raw = raw or {}
+
+
+class _FakeAdServerClient:
+    ad_server_type = SimpleNamespace(value="gam")
+
+    def __init__(self, items):
+        self._items = items
+        self.limit_requested = "not called"
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    # Same default as the real adapters, so a caller that omits limit is caught.
+    async def list_inventory(self, filter_str=None, limit=100):
+        self.limit_requested = limit
+        return self._items
+
+
+def _flow_settings(**overrides) -> SimpleNamespace:
+    defaults = {
+        "gam_network_code": "12345",
+        "freewheel_sh_mcp_url": None,
+        "ad_server_type": "none",
+        "seller_organization_id": "test-seller-org",
+        "seller_organization_name": "Test Seller",
+        # AI-8 added this to real Settings after this fixture was written;
+        # _estimate_base_cpm() reads it unconditionally.
+        "default_price_floor_cpm": 10.0,
+    }
+    defaults.update(overrides)
+    return SimpleNamespace(**defaults)
+
+
+async def _run_sync(storage, settings, client) -> None:
+    with (
+        patch("ad_seller.flows.product_setup_flow.get_settings", return_value=settings),
+        patch("ad_seller.storage.factory.get_storage", return_value=storage),
+        patch("ad_seller.clients.ad_server_base.get_ad_server_client", lambda: client),
+    ):
+        from ad_seller.flows.product_setup_flow import ProductSetupFlow
+
+        flow = ProductSetupFlow()
+        await flow.sync_from_ad_server()
+
+
+class TestSyncPersistsAndPrunesProducts:
+    async def test_sync_requests_the_whole_inventory(self, storage):
+        """The adapters' default limit is 100; a sync that inherited it
+        would see only the first 100 items and prune the rest."""
+        client = _FakeAdServerClient([_FakeInventoryItem("gam-001", "Homepage Display Banner")])
+        await _run_sync(storage, _flow_settings(), client)
+
+        assert client.limit_requested is None
+
+    async def test_sync_persists_products_to_storage(self, storage):
+        client = _FakeAdServerClient(
+            [
+                _FakeInventoryItem("gam-001", "Homepage Display Banner", {"floor_price_cpm": 8.0}),
+                _FakeInventoryItem("gam-002", "CTV Premium App", {"floor_price_cpm": 20.0}),
+            ]
+        )
+        await _run_sync(storage, _flow_settings(), client)
+
+        assert await storage.exists("synced_product:gam-001")
+        assert await storage.exists("synced_product:gam-002")
+
+    async def test_resync_with_fewer_items_prunes_the_dropped_product(self, storage):
+        settings = _flow_settings()
+        client_a = _FakeAdServerClient(
+            [
+                _FakeInventoryItem("gam-001", "Homepage Display Banner"),
+                _FakeInventoryItem("gam-002", "CTV Premium App"),
+            ]
+        )
+        await _run_sync(storage, settings, client_a)
+
+        client_b = _FakeAdServerClient([_FakeInventoryItem("gam-001", "Homepage Display Banner")])
+        await _run_sync(storage, settings, client_b)
+
+        assert await storage.exists("synced_product:gam-001")
+        assert not await storage.exists("synced_product:gam-002"), (
+            "product dropped from the ad server must be pruned, not left as phantom inventory"
+        )
+
+    async def test_catalog_reflects_synced_products_live_after_flow_sync(self, storage):
+        """End-to-end: GET /products' data source reflects a sync immediately."""
+        settings = _flow_settings()
+        client = _FakeAdServerClient([_FakeInventoryItem("gam-999", "Pre-Roll Video Unit")])
+        await _run_sync(storage, settings, client)
+
+        with (
+            patch("ad_seller.storage.factory.get_storage", return_value=storage),
+            patch("ad_seller.config.get_settings", return_value=settings),
+        ):
+            catalog = await catalog_service.get_static_product_catalog()
+
+        assert set(catalog["products"]) == {"gam-999"}
+
+    async def test_resync_with_zero_items_serves_an_honestly_empty_catalog(self, storage):
+        """A real sync finding zero inventory must not resurrect demo products.
+
+        Regression test: a zero-item sync is a genuine, successful result
+        (the ad server really has nothing matching right now), not the
+        absence of a sync. Both leave zero ``synced_product:*`` keys, so
+        without a separate "a real sync has completed" marker, the catalog
+        builder couldn't tell the two apart and silently fell back to the
+        13-product static demo catalog -- fake inventory reappearing in
+        place of an honest empty one.
+        """
+        settings = _flow_settings()
+        client_a = _FakeAdServerClient([_FakeInventoryItem("gam-999", "Pre-Roll Video Unit")])
+        await _run_sync(storage, settings, client_a)
+
+        client_b = _FakeAdServerClient([])
+        await _run_sync(storage, settings, client_b)
+
+        assert not await storage.exists("synced_product:gam-999")
+
+        with (
+            patch("ad_seller.storage.factory.get_storage", return_value=storage),
+            patch("ad_seller.config.get_settings", return_value=settings),
+        ):
+            catalog = await catalog_service.get_static_product_catalog()
+
+        assert catalog["products"] == {}, (
+            "a real sync that found zero inventory must serve an empty catalog, "
+            "not fall back to the static demo products"
+        )
+
+
+class TestMidSyncCrashRollsBackPartialWrites:
+    """A sync that crashes partway through persisting items must leave
+    storage exactly as the last successful sync left it (AI-6 follow-up).
+
+    Defect: ``persist_synced_product``/``_upsert_synced_package`` write
+    straight to storage one item at a time inside the loop. An exception
+    partway through left every item already written in THIS run
+    permanently in storage, even though the sync is reported as failed
+    and the code's own docstring claims a failure "never silently
+    changes" the synced layer. Found live: synced A+B cleanly, then a
+    sync that persists C and crashes before D: A and B survived as
+    expected, but C leaked in anyway.
+    """
+
+    async def test_item_persisted_before_a_later_crash_is_rolled_back(self, storage):
+        settings = _flow_settings()
+        client_a = _FakeAdServerClient(
+            [_FakeInventoryItem("A", "Product A"), _FakeInventoryItem("B", "Product B")]
+        )
+        await _run_sync(storage, settings, client_a)
+
+        client_b = _FakeAdServerClient(
+            [
+                _FakeInventoryItem("C", "Product C"),
+                _FakeInventoryItem("D", "Product D"),
+                _FakeInventoryItem("E", "Product E"),
+            ]
+        )
+        real_persist = catalog_service.persist_synced_product
+
+        async def flaky_persist(product):
+            if product.product_id == "D":
+                raise RuntimeError("simulated crash persisting D")
+            await real_persist(product)
+
+        with patch("ad_seller.services.catalog_service.persist_synced_product", flaky_persist):
+            await _run_sync(storage, settings, client_b)
+
+        assert await storage.exists("synced_product:A")
+        assert await storage.exists("synced_product:B")
+        assert not await storage.exists("synced_product:C"), (
+            "C was persisted by the crashed run before D failed -- it must be "
+            "rolled back, not left behind despite the sync being reported as failed"
+        )
+        assert not await storage.exists("synced_product:D")
+        assert not await storage.exists("synced_product:E")
+
+    async def test_resynced_existing_item_restores_its_prior_value_on_rollback(self, storage):
+        """Rollback must restore, not blindly delete, an id that already
+        existed -- a re-synced item can share an id with one a previous
+        successful sync already persisted."""
+        settings = _flow_settings()
+        client_a = _FakeAdServerClient(
+            [_FakeInventoryItem("A", "Product A", {"floor_price_cpm": 8.0})]
+        )
+        await _run_sync(storage, settings, client_a)
+        original = await storage.get("synced_product:A")
+        assert original["floor_cpm"] == pytest.approx(8.0 * 0.85)
+
+        client_b = _FakeAdServerClient(
+            [
+                _FakeInventoryItem("A", "Product A", {"floor_price_cpm": 99.0}),
+                _FakeInventoryItem("D", "Product D"),
+            ]
+        )
+        real_persist = catalog_service.persist_synced_product
+
+        async def flaky_persist(product):
+            if product.product_id == "D":
+                raise RuntimeError("simulated crash persisting D")
+            await real_persist(product)
+
+        with patch("ad_seller.services.catalog_service.persist_synced_product", flaky_persist):
+            await _run_sync(storage, settings, client_b)
+
+        restored = await storage.get("synced_product:A")
+        assert restored == original, (
+            "A existed before this crashed run and must be restored to its "
+            "prior value, not deleted or left with the crashed run's new value"
+        )
+        assert not await storage.exists("synced_product:D")
+
+    async def test_packages_loop_crash_rolls_back_this_runs_products(self, storage):
+        """The products loop commits before the packages loop runs, so a
+        crash in the packages loop must roll back this run's products too,
+        not only its packages."""
+        settings = _flow_settings()
+        client_a = _FakeAdServerClient(
+            [_FakeInventoryItem("A", "Product A", {"floor_price_cpm": 8.0})]
+        )
+        await _run_sync(storage, settings, client_a)
+        original = await storage.get("synced_product:A")
+
+        client_b = _FakeAdServerClient(
+            [
+                _FakeInventoryItem("A", "Product A", {"floor_price_cpm": 99.0}),
+                _FakeInventoryItem("C", "Product C"),
+            ]
+        )
+
+        async def crashing_upsert(self, storage, package):
+            raise RuntimeError("simulated crash upserting a package")
+
+        with patch(
+            "ad_seller.flows.product_setup_flow.ProductSetupFlow._upsert_synced_package",
+            crashing_upsert,
+        ):
+            await _run_sync(storage, settings, client_b)
+
+        assert await storage.get("synced_product:A") == original, (
+            "A must be restored to the last successful sync's value"
+        )
+        assert not await storage.exists("synced_product:C"), (
+            "C was persisted by the products loop of the failed run -- it must be "
+            "rolled back when the packages loop fails"
+        )
+
+
+class TestSyncWithRealCsvAdapter:
+    """The flow tests above use _FakeAdServerClient, which accepts any
+    ``limit``. This runs the sync through the real CSVAdServerClient so a
+    flow/adapter contract mismatch (the flow passing ``limit=None``) fails
+    here instead of only on a live server."""
+
+    async def test_csv_sync_completes_and_persists_every_row(self, storage, tmp_path):
+        import shutil
+        from pathlib import Path
+
+        from ad_seller.clients.csv_adapter import CSVAdServerClient
+
+        sample = Path(__file__).parents[2] / "data" / "csv" / "samples" / "ctv_streaming"
+        shutil.copy(sample / "inventory.csv", tmp_path / "inventory.csv")
+        client = CSVAdServerClient(data_dir=str(tmp_path))
+        async with client:
+            rows = await client.list_inventory(limit=10_000)
+
+        settings = _flow_settings(gam_network_code=None, ad_server_type="csv")
+        with (
+            patch("ad_seller.flows.product_setup_flow.get_settings", return_value=settings),
+            patch("ad_seller.storage.factory.get_storage", return_value=storage),
+            patch("ad_seller.clients.ad_server_base.get_ad_server_client", lambda: client),
+        ):
+            from ad_seller.flows.product_setup_flow import ProductSetupFlow
+
+            flow = ProductSetupFlow()
+            await flow.sync_from_ad_server()
+
+        assert not flow.state.ad_server_sync_failed, flow.state.warnings
+        assert len(await storage.keys("synced_product:*")) == len(rows)