fix: real, storage-backed catalog for synced ad servers (AI-6)

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`

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

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 --
 

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

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.

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(

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:

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."""

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."""

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