feat(handlers): expose sync_catalogs through sales-catalog-driven specialism

Issue #786: sync_catalogs was defined in ADCP_TOOL_DEFINITIONS and
SyncCatalogsRequest/response types existed, but no specialism mapping
surfaced it to buyers and PlatformHandler had no dispatcher shim.

Changes:
- Add _CATALOG_ADVERTISED_TOOLS frozenset and union it into
  SPECIALISM_TO_ADVERTISED_TOOLS["sales-catalog-driven"]
- Add _CATALOG_ADVERTISED_TOOLS to PlatformHandler.advertised_tools ClassVar
- Add _project_sync_catalogs helper (list-return ergonomic arm)
- Add PlatformHandler.sync_catalogs dispatcher; passes full
  SyncCatalogsRequest to platform (no arg projection) so adopters
  retain access to delete_missing/dry_run/validation_mode
- Add sync_catalogs method declaration to SalesPlatform Protocol with
  discovery-mode semantics documented (req.catalogs is None = read-only)
- Update tests: expected set, parametrize list, three shim tests,
  two per-specialism filter tests, runtime_checkable update

Note: adding sync_catalogs to SalesPlatform Protocol body is a minor
structural consequence for @runtime_checkable isinstance checks —
isinstance(obj, SalesPlatform) now requires sync_catalogs. Production
code uses validate_platform (not isinstance) as its enforcement gate;
no existing framework code is affected. Documented in test update.

#786

Author: claude

src/adcp/decisioning/handler.py

@@ -165,6 +165,8 @@
     SyncAccountsResponse,
     SyncAudiencesRequest,
     SyncAudiencesSuccessResponse,
+    SyncCatalogsRequest,
+    SyncCatalogsSuccessResponse,
     SyncCreativesRequest,
     SyncCreativesSuccessResponse,
     SyncPlansRequest,
@@ -292,6 +294,11 @@
         "si_terminate_session",
     }
 )
+_CATALOG_ADVERTISED_TOOLS: frozenset[str] = frozenset(
+    {
+        "sync_catalogs",
+    }
+)
 _GOVERNANCE_ADVERTISED_TOOLS: frozenset[str] = frozenset(
     {
         "check_governance",
@@ -394,7 +401,9 @@
     "sales-guaranteed": _SALES_ADVERTISED_TOOLS | _ACCOUNT_ADVERTISED_TOOLS,
     "sales-broadcast-tv": _SALES_ADVERTISED_TOOLS | _ACCOUNT_ADVERTISED_TOOLS,
     "sales-social": _SALES_ADVERTISED_TOOLS | _ACCOUNT_ADVERTISED_TOOLS,
-    "sales-catalog-driven": _SALES_ADVERTISED_TOOLS | _ACCOUNT_ADVERTISED_TOOLS,
+    "sales-catalog-driven": (
+        _SALES_ADVERTISED_TOOLS | _ACCOUNT_ADVERTISED_TOOLS | _CATALOG_ADVERTISED_TOOLS
+    ),
     "sales-proposal-mode": _SALES_ADVERTISED_TOOLS | _ACCOUNT_ADVERTISED_TOOLS,
     # Creative — Builder + AdServer. Builder claims expose
     # build_creative + optional preview_creative; AdServer adds
@@ -794,6 +803,27 @@ def _project_sync_audiences(result: Any) -> Any:
     return result
 
 
+def _project_sync_catalogs(result: Any) -> Any:
+    """Project the adopter's ``sync_catalogs`` return into the wire envelope shape.
+
+    Adopters may return a list of catalog-result rows (ergonomic form) or a
+    fully-shaped :class:`SyncCatalogsSuccessResponse`. The wire envelope per
+    ``schemas/cache/media-buy/sync-catalogs-response.json`` is
+    ``{catalogs: [rows]}``. This helper wraps the list case.
+
+    Discovery mode (``req.catalogs is None``) returns existing synced catalogs
+    per spec — the platform method must handle ``req.catalogs is None`` as a
+    read-only path.
+    """
+    if isinstance(result, list):
+        return {
+            "catalogs": [
+                r.model_dump(mode="json") if hasattr(r, "model_dump") else r for r in result
+            ]
+        }
+    return result
+
+
 def _project_sync_accounts(result: Any) -> Any:
     """Project the adopter's ``upsert`` return into the
     ``sync_accounts`` wire envelope.
@@ -1014,6 +1044,7 @@ class PlatformHandler(ADCPHandler[ToolContext]):
         | set(_CREATIVE_ADVERTISED_TOOLS)
         | set(_SIGNALS_ADVERTISED_TOOLS)
         | set(_AUDIENCE_ADVERTISED_TOOLS)
+        | set(_CATALOG_ADVERTISED_TOOLS)
         | set(_GOVERNANCE_ADVERTISED_TOOLS)
         | set(_BRAND_RIGHTS_ADVERTISED_TOOLS)
         | set(_CONTENT_STANDARDS_ADVERTISED_TOOLS)
@@ -2376,6 +2407,41 @@ async def sync_audiences(  # type: ignore[override]
         self._maybe_auto_emit_sync_completion("sync_audiences", params, projected)
         return cast("SyncAudiencesSuccessResponse", projected)
 
+    async def sync_catalogs(  # type: ignore[override]
+        self,
+        params: SyncCatalogsRequest,
+        context: ToolContext | None = None,
+    ) -> SyncCatalogsSuccessResponse:
+        """Sync product catalogs with the platform.
+
+        The platform method receives the full :class:`SyncCatalogsRequest`
+        so adopters can inspect ``req.catalogs``, ``req.delete_missing``,
+        ``req.dry_run``, and ``req.validation_mode``. Discovery mode
+        (``req.catalogs is None``) returns existing synced catalogs without
+        modification — the platform method must handle ``req.catalogs is None``
+        as a read-only path.
+
+        Two return arms per the per-specialism Protocol: a list of
+        :class:`SyncCatalogResult` rows (ergonomic form) or a fully-shaped
+        :class:`SyncCatalogsSuccessResponse`. The shim projects the list arm
+        to the wire envelope ``{catalogs: [...]}`` so adopters can return
+        the ergonomic form.
+        """
+        tool_ctx = context or ToolContext()
+        account = await self._resolve_account(getattr(params, "account", None), tool_ctx)
+        ctx = self._build_ctx(tool_ctx, account)
+        result = await _invoke_platform_method(
+            self._platform,
+            "sync_catalogs",
+            params,
+            ctx,
+            executor=self._executor,
+            registry=self._registry,
+        )
+        projected = _project_sync_catalogs(result)
+        self._maybe_auto_emit_sync_completion("sync_catalogs", params, projected)
+        return cast("SyncCatalogsSuccessResponse", projected)
+
     # ----- CampaignGovernancePlatform -----
 
     async def check_governance(  # type: ignore[override]

src/adcp/decisioning/specialisms/sales.py

@@ -25,8 +25,12 @@
 * :meth:`provide_performance_feedback`
 * :meth:`list_creative_formats`
 * :meth:`list_creatives`
-* :meth:`sync_catalogs` — required when claiming
-  ``sales-catalog-driven``
+
+Required only when claiming ``sales-catalog-driven``:
+
+* :meth:`sync_catalogs` — catalog sync and discovery. ``validate_platform``
+  hard-fails at server boot if a ``sales-catalog-driven`` platform doesn't
+  implement this.
 
 The framework's :func:`validate_platform` walks ``capabilities.specialisms``
 and confirms each specialism's required methods exist on the platform
@@ -64,6 +68,8 @@
         ListCreativesResponse,
         ProvidePerformanceFeedbackRequest,
         ProvidePerformanceFeedbackResponse,
+        SyncCatalogsRequest,
+        SyncCatalogsSuccessResponse,
         SyncCreativesRequest,
         SyncCreativesSuccessResponse,
         UpdateMediaBuyRequest,
@@ -269,3 +275,36 @@ def list_creatives(
         Required when claiming any ``sales-*`` specialism in v6.0 rc.1+.
         """
         ...
+
+    # ---- Required when claiming ``sales-catalog-driven`` ----
+
+    def sync_catalogs(
+        self,
+        req: SyncCatalogsRequest,
+        ctx: RequestContext[TMeta],
+    ) -> MaybeAsync[SyncCatalogsSuccessResponse]:
+        """Sync product catalogs with the platform.
+
+        **Required** when claiming ``sales-catalog-driven``.
+        ``validate_platform`` hard-fails at server boot when this method is
+        absent on a ``sales-catalog-driven`` platform.
+
+        Discovery mode: when ``req.catalogs is None``, return the account's
+        existing synced catalogs without modification (read-only path per
+        AdCP spec). Check ``req.catalogs`` before applying any mutations::
+
+            def sync_catalogs(self, req, ctx):
+                if req.catalogs is None:
+                    return self._get_existing_catalogs(ctx.account_id)
+                return self._upsert_catalogs(req.catalogs, ctx)
+
+        **Important:** ``req.delete_missing=True`` with ``req.catalogs=None``
+        is spec-undefined — reject it with
+        ``AdcpError("INVALID_REQUEST", field="catalogs")`` rather than
+        silently deleting buyer-managed catalogs.
+
+        Return a list of :class:`~adcp.types.SyncCatalogResult` rows
+        (ergonomic form) or a fully-shaped
+        :class:`~adcp.types.SyncCatalogsSuccessResponse`.
+        """
+        ...

tests/test_decisioning_advertised_per_specialism.py

@@ -117,6 +117,29 @@ def build_creative(self, req, ctx):
         return {"creative_manifest": {"creative_id": "cr_1"}}
 
 
+class _CatalogDrivenPlatform(DecisioningPlatform):
+    capabilities = DecisioningCapabilities(specialisms=["sales-catalog-driven"])
+    accounts = SingletonAccounts(account_id="catalog-driven")
+
+    def get_products(self, req, ctx):
+        return {"products": []}
+
+    def create_media_buy(self, req, ctx):
+        return {"media_buy_id": "x", "status": "active"}
+
+    def update_media_buy(self, media_buy_id, patch, ctx):
+        return {"media_buy_id": media_buy_id, "status": "active"}
+
+    def sync_creatives(self, req, ctx):
+        return {"creatives": []}
+
+    def get_media_buy_delivery(self, req, ctx):
+        return {"media_buy_deliveries": []}
+
+    def sync_catalogs(self, req, ctx):
+        return {"catalogs": []}
+
+
 def test_sales_only_does_not_advertise_creative_or_signals_tools(executor) -> None:
     """Regression: sales-only adopter saw acquire_rights, build_creative,
     check_governance, etc. in tools/list. After the per-specialism
@@ -333,3 +356,40 @@ def test_class_level_inspection_preserves_full_universe() -> None:
     assert "get_products" in tools
     assert "build_creative" in tools
     assert "acquire_rights" in tools
+
+
+def test_catalog_driven_advertises_sync_catalogs(executor) -> None:
+    """``sales-catalog-driven`` must expose ``sync_catalogs`` via
+    ``tools/list``. This was the missing wire-up reported in issue #786:
+    the tool existed and types were defined, but no specialism mapping
+    surfaced it to buyers."""
+    handler = PlatformHandler(
+        _CatalogDrivenPlatform(),
+        executor=executor,
+        registry=InMemoryTaskRegistry(),
+    )
+    tools = {tool["name"] for tool in get_tools_for_handler(handler)}
+
+    assert "sync_catalogs" in tools, (
+        "sales-catalog-driven platform must advertise sync_catalogs; "
+        "check SPECIALISM_TO_ADVERTISED_TOOLS['sales-catalog-driven']"
+    )
+    # Sales tools also present.
+    assert "get_products" in tools
+    assert "create_media_buy" in tools
+
+
+def test_non_catalog_sales_does_not_advertise_sync_catalogs(executor) -> None:
+    """``sync_catalogs`` is specific to ``sales-catalog-driven`` — no
+    other sales-* specialism should surface it."""
+    handler = PlatformHandler(
+        _SalesOnlyPlatform(),
+        executor=executor,
+        registry=InMemoryTaskRegistry(),
+    )
+    tools = {tool["name"] for tool in get_tools_for_handler(handler)}
+
+    assert "sync_catalogs" not in tools, (
+        "sales-non-guaranteed must not advertise sync_catalogs; "
+        "sync_catalogs is only for sales-catalog-driven"
+    )

tests/test_decisioning_handler_shims.py

@@ -84,6 +84,8 @@ def test_advertised_tools_covers_every_specialism_wire_tool() -> None:
         "activate_signal",
         # Audience
         "sync_audiences",
+        # Catalog (sales-catalog-driven only)
+        "sync_catalogs",
         # Governance
         "check_governance",
         "sync_plans",
@@ -131,6 +133,7 @@ def test_advertised_tools_covers_every_specialism_wire_tool() -> None:
         "get_signals",
         "activate_signal",
         "sync_audiences",
+        "sync_catalogs",
         "check_governance",
         "sync_plans",
         "report_plan_outcome",
@@ -256,6 +259,136 @@ def sync_audiences(self, audiences, ctx):
     assert received_audiences == fake_audiences
 
 
+@pytest.mark.asyncio
+async def test_sync_catalogs_shim_passes_full_request_to_platform(executor) -> None:
+    """The ``sync_catalogs`` shim passes the full ``SyncCatalogsRequest``
+    to the platform method (no arg projection). Adopters receive the full
+    request so they can inspect ``req.catalogs``, ``req.delete_missing``,
+    ``req.dry_run``, and ``req.validation_mode``."""
+    received_req = []
+
+    class _CatalogAgent(DecisioningPlatform):
+        capabilities = DecisioningCapabilities(specialisms=["sales-catalog-driven"])
+        accounts = SingletonAccounts(account_id="hello")
+
+        def get_products(self, req, ctx):
+            return {"products": []}
+
+        def create_media_buy(self, req, ctx):
+            return {"media_buy_id": "x", "status": "active"}
+
+        def update_media_buy(self, media_buy_id, patch, ctx):
+            return {"media_buy_id": media_buy_id, "status": "active"}
+
+        def sync_creatives(self, req, ctx):
+            return {"creatives": []}
+
+        def get_media_buy_delivery(self, req, ctx):
+            return {"media_buy_deliveries": []}
+
+        def sync_catalogs(self, req, ctx):
+            received_req.append(req)
+            return {"catalogs": []}
+
+    handler = PlatformHandler(
+        _CatalogAgent(),
+        executor=executor,
+        registry=InMemoryTaskRegistry(),
+    )
+    from adcp.types import SyncCatalogsRequest
+
+    fake_catalogs = [{"type": "product", "catalog_id": "feed-1"}]
+    req = SyncCatalogsRequest.model_construct(catalogs=fake_catalogs)
+    await handler.sync_catalogs(req, ToolContext())
+    assert len(received_req) == 1
+    assert received_req[0] is req
+
+
+@pytest.mark.asyncio
+async def test_sync_catalogs_discovery_mode_passes_none_catalogs(executor) -> None:
+    """Discovery mode (``req.catalogs is None``) passes the request
+    through intact — the platform receives ``req`` with ``catalogs=None``
+    and can distinguish discovery from an explicit empty push."""
+    received_catalogs_value = []
+
+    class _CatalogAgent(DecisioningPlatform):
+        capabilities = DecisioningCapabilities(specialisms=["sales-catalog-driven"])
+        accounts = SingletonAccounts(account_id="hello")
+
+        def get_products(self, req, ctx):
+            return {"products": []}
+
+        def create_media_buy(self, req, ctx):
+            return {"media_buy_id": "x", "status": "active"}
+
+        def update_media_buy(self, media_buy_id, patch, ctx):
+            return {"media_buy_id": media_buy_id, "status": "active"}
+
+        def sync_creatives(self, req, ctx):
+            return {"creatives": []}
+
+        def get_media_buy_delivery(self, req, ctx):
+            return {"media_buy_deliveries": []}
+
+        def sync_catalogs(self, req, ctx):
+            received_catalogs_value.append(req.catalogs)
+            return {"catalogs": []}
+
+    handler = PlatformHandler(
+        _CatalogAgent(),
+        executor=executor,
+        registry=InMemoryTaskRegistry(),
+    )
+    from adcp.types import SyncCatalogsRequest
+
+    req = SyncCatalogsRequest.model_construct(catalogs=None)
+    await handler.sync_catalogs(req, ToolContext())
+    assert received_catalogs_value == [None]
+
+
+@pytest.mark.asyncio
+async def test_sync_catalogs_list_return_projected_to_envelope(executor) -> None:
+    """When the platform returns a list of catalog results (ergonomic arm),
+    the shim wraps it in ``{catalogs: [...]}``."""
+
+    class _CatalogAgent(DecisioningPlatform):
+        capabilities = DecisioningCapabilities(specialisms=["sales-catalog-driven"])
+        accounts = SingletonAccounts(account_id="hello")
+
+        def get_products(self, req, ctx):
+            return {"products": []}
+
+        def create_media_buy(self, req, ctx):
+            return {"media_buy_id": "x", "status": "active"}
+
+        def update_media_buy(self, media_buy_id, patch, ctx):
+            return {"media_buy_id": media_buy_id, "status": "active"}
+
+        def sync_creatives(self, req, ctx):
+            return {"creatives": []}
+
+        def get_media_buy_delivery(self, req, ctx):
+            return {"media_buy_deliveries": []}
+
+        def sync_catalogs(self, req, ctx):
+            return [{"catalog_id": "feed-1", "action": "created", "item_count": 100}]
+
+    handler = PlatformHandler(
+        _CatalogAgent(),
+        executor=executor,
+        registry=InMemoryTaskRegistry(),
+    )
+    from adcp.types import SyncCatalogsRequest
+
+    req = SyncCatalogsRequest.model_construct(
+        catalogs=[{"type": "product", "catalog_id": "feed-1"}]
+    )
+    result = await handler.sync_catalogs(req, ToolContext())
+    assert result == {
+        "catalogs": [{"catalog_id": "feed-1", "action": "created", "item_count": 100}]
+    }
+
+
 @pytest.mark.asyncio
 async def test_check_governance_shim_routes_to_platform(executor) -> None:
     class _GovernanceAgent(DecisioningPlatform):