diff --git a/docs/handler-authoring.md b/docs/handler-authoring.md index 5b0c4624c..6d8ee812d 100644 --- a/docs/handler-authoring.md +++ b/docs/handler-authoring.md @@ -55,6 +55,50 @@ Pass ``advertise_all=True`` to ``serve()`` / ``create_mcp_server()`` / storyboards, agents that deliberately signal ``not_supported`` on specific tools). +**Custom handler bases — declare ``advertised_tools``.** The override +filter works perfectly for direct ``ADCPHandler`` subclasses and the +specialized bases (``GovernanceHandler``, ``ContentStandardsHandler``, +etc.). But if you're authoring a *new* specialized base that +introduces its own focused tool set — typically a codegen target like +``adcp.decisioning.handler.PlatformHandler``, or a hand-rolled +``ReadOnlyAnalyticsHandler`` — declare the tool set on the class +itself:: + + from typing import ClassVar + from adcp.server import ADCPHandler + + class ReadOnlyAnalyticsHandler(ADCPHandler): + advertised_tools: ClassVar[set[str]] = { + "get_products", + "get_media_buy_delivery", + } + + # ... method bodies ... + +The framework auto-registers ``ReadOnlyAnalyticsHandler -> +advertised_tools`` at class definition time via +``ADCPHandler.__init_subclass__``. The override filter then runs +against this set instead of inheriting ``ADCPHandler``'s full +surface. + +Equivalent imperative form — same outcome:: + + from adcp.server import register_handler_tools + register_handler_tools("ReadOnlyAnalyticsHandler", {"get_products", ...}) + +Without either declaration, ``serve()`` emits a one-time +``UserWarning`` at boot pointing you at the registration paths. The +warning matters because the alternative — silent over-advertisement +of the full ADCPHandler surface — is exactly the discoverability +gap that bites operators in production: ``tools/list`` returns 57 +tools when the agent only handles 2. + +**What not to build:** don't use ``advertise_all=True`` as a +workaround for missing registration. The flag exists for legitimate +opt-in cases (storyboards, deliberate ``not_supported`` signaling); +using it to silence the registration warning over-advertises every +tool to every buyer. + ## The `_impl` pattern (production-grade) Production agents usually don't put business logic directly on handler diff --git a/src/adcp/server/__init__.py b/src/adcp/server/__init__.py index eaa7f0012..4619554aa 100644 --- a/src/adcp/server/__init__.py +++ b/src/adcp/server/__init__.py @@ -99,6 +99,7 @@ async def get_products(params, context=None): MCPToolSet, create_mcp_tools, get_tools_for_handler, + register_handler_tools, validate_discovery_set, ) from adcp.server.proposal import ProposalBuilder, ProposalNotSupported @@ -167,6 +168,7 @@ async def get_products(params, context=None): "create_mcp_tools", "create_mcp_server", "get_tools_for_handler", + "register_handler_tools", "serve", "validate_discovery_set", # A2A integration diff --git a/src/adcp/server/base.py b/src/adcp/server/base.py index 3f3fdfffa..d099d6b6d 100644 --- a/src/adcp/server/base.py +++ b/src/adcp/server/base.py @@ -231,10 +231,57 @@ class ADCPHandler(ABC, Generic[TContext]): - ContentStandardsHandler: For content standards agents - SponsoredIntelligenceHandler: For sponsored intelligence agents - GovernanceHandler: For governance agents + + **Tool advertisement** (`advertised_tools` class attribute): + + A subclass that introduces a new specialism — i.e., a custom base + that needs its own ``tools/list`` filter rather than inheriting one + from a built-in handler — declares the tool set on the class body:: + + class PlatformHandler(ADCPHandler): + advertised_tools: ClassVar[set[str]] = { + "get_products", + "create_media_buy", + ... + } + + The framework registers ``PlatformHandler -> advertised_tools`` with + :func:`adcp.server.mcp_tools.register_handler_tools` at class + definition time. Subclasses that DON'T introduce a new specialism + (a custom ``MyContentAgent(ContentStandardsHandler)``, for example) + inherit their parent's tool set unchanged — no class attr needed. + + Hand-written equivalent (no ``advertised_tools`` declaration):: + + from adcp.server.mcp_tools import register_handler_tools + register_handler_tools("PlatformHandler", {...}) + + Either path is fine; codegen targets emit the class attribute so the + declaration sits next to the class definition. """ _agent_type: str = "this agent" + def __init_subclass__(cls, **kwargs: Any) -> None: + """Auto-register subclass-declared tool advertisement. + + Reads ``cls.__dict__["advertised_tools"]`` (subclass-defined-only + — inherited values don't trigger re-registration) and routes + through :func:`adcp.server.mcp_tools.register_handler_tools`. + Only fires when the subclass declares the attribute on its own + class body; intermediate subclasses (multi-level hierarchy) + register at the level that introduces the attribute. + + The lazy import avoids a base.py ↔ mcp_tools.py circular — + mcp_tools imports ADCPHandler at module load, so register is + looked up only when a subclass is actually being created. + """ + super().__init_subclass__(**kwargs) + if "advertised_tools" in cls.__dict__: + from adcp.server.mcp_tools import register_handler_tools + + register_handler_tools(cls.__name__, cls.__dict__["advertised_tools"]) + def _not_supported(self, operation: str) -> NotImplementedResponse: """Create a not-supported response that includes the agent type.""" return not_supported(f"{operation} is not supported by {self._agent_type}") diff --git a/src/adcp/server/mcp_tools.py b/src/adcp/server/mcp_tools.py index a2e5a4897..8f4ef348b 100644 --- a/src/adcp/server/mcp_tools.py +++ b/src/adcp/server/mcp_tools.py @@ -19,6 +19,7 @@ from __future__ import annotations import copy +import difflib import logging from collections.abc import Callable, Iterable from typing import Any @@ -1027,6 +1028,58 @@ def validate_discovery_set(tools: Iterable[str]) -> None: assert not _unknown, f"{_handler_name} references unknown tools: {_unknown}" +def register_handler_tools(handler_name: str, tools: Iterable[str]) -> None: + """Register a handler-class-name → tool-set mapping with the framework. + + Public seam. ``get_tools_for_handler`` reads ``_HANDLER_TOOLS`` to + filter ``tools/list`` per handler subclass; without registration, an + ``ADCPHandler`` subclass that introduces a new specialism would fall + through to its parent's tool set (typically ``ADCPHandler``'s + full-spec surface), over-advertising. Codegen targets like + ``adcp.decisioning.handler.PlatformHandler`` register here at class + definition time via ``ADCPHandler.__init_subclass__``; hand-written + custom bases call this directly before ``serve()``. + Idempotent on equal input — calling twice with the same tool set + is a no-op so module re-imports / reload-friendly test harnesses + don't break. + Conflicts raise. Unknown tool names raise with a closest-match + suggestion (typo recovery for adopters working from spec memory). + :param handler_name: The class name of the handler subclass — + typically ``cls.__name__`` from inside ``__init_subclass__``. + :param tools: Iterable of AdCP tool names this handler answers + (members of ``ADCP_TOOL_DEFINITIONS``). Order doesn't matter. + :raises ValueError: when ``handler_name`` is already registered with + a different tool set, or when any tool name is not in the AdCP + spec surface. + """ + incoming = frozenset(tools) + existing = _HANDLER_TOOLS.get(handler_name) + if existing is not None: + if frozenset(existing) == incoming: + return + raise ValueError( + f"register_handler_tools({handler_name!r}, ...) called twice " + f"with different tool sets. Existing: {sorted(existing)}; " + f"incoming: {sorted(incoming)}. The framework can only hold " + "one mapping per handler class — pick the canonical set." + ) + unknown = incoming - _ALL_TOOL_NAMES + if unknown: + suggestions: list[str] = [] + for bad in sorted(unknown): + close = difflib.get_close_matches(bad, _ALL_TOOL_NAMES, n=1) + if close: + suggestions.append(f"{bad!r} (did you mean {close[0]!r}?)") + else: + suggestions.append(repr(bad)) + raise ValueError( + f"register_handler_tools({handler_name!r}, ...): unknown tool " + f"name(s) {', '.join(suggestions)}. Tool names must match the " + "AdCP spec — see ``adcp.server.mcp_tools.ADCP_TOOL_DEFINITIONS``." + ) + _HANDLER_TOOLS[handler_name] = set(incoming) + + # ============================================================================ # Pydantic schema generation — spec-accurate input schemas # ============================================================================ @@ -1325,10 +1378,18 @@ def _apply_pydantic_schemas() -> None: _apply_pydantic_schemas() -_SDK_BASE_CLASS_NAMES: frozenset[str] = frozenset(_HANDLER_TOOLS.keys()) -"""Names of the SDK's own base classes. Used to detect whether a method -is an SDK default (inherited from one of these) or a subclass override. -Kept alongside ``_HANDLER_TOOLS`` so they can't drift.""" +def _is_sdk_base_class(cls_name: str) -> bool: + """True when ``cls_name`` is registered in ``_HANDLER_TOOLS``. + + Used during MRO walks to identify the nearest SDK base whose + method baselines a subclass override. Reads ``_HANDLER_TOOLS`` + live so that handler classes registered after import time — + via :func:`register_handler_tools` or + :meth:`ADCPHandler.__init_subclass__` reading + ``advertised_tools`` — participate in override detection without + requiring a frozen-set rebuild. + """ + return cls_name in _HANDLER_TOOLS def _is_method_overridden(handler_cls: type, method_name: str) -> bool: @@ -1375,7 +1436,7 @@ def _is_method_overridden(handler_cls: type, method_name: str) -> bool: sdk_base: type | None = None base_method: Any | None = None for base in handler_cls.__mro__[1:]: - if base.__name__ not in _SDK_BASE_CLASS_NAMES: + if not _is_sdk_base_class(base.__name__): continue found = base.__dict__.get(method_name) if found is None: diff --git a/src/adcp/server/serve.py b/src/adcp/server/serve.py index 63215e6fb..c7162bb95 100644 --- a/src/adcp/server/serve.py +++ b/src/adcp/server/serve.py @@ -20,6 +20,7 @@ async def get_adcp_capabilities(self, params, context=None): import logging import os +import warnings from collections.abc import Awaitable, Callable from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Literal @@ -27,7 +28,11 @@ async def get_adcp_capabilities(self, params, context=None): logger = logging.getLogger("adcp.server") from adcp.server.base import ADCPHandler, ToolContext -from adcp.server.mcp_tools import create_tool_caller, get_tools_for_handler +from adcp.server.mcp_tools import ( + _HANDLER_TOOLS, + create_tool_caller, + get_tools_for_handler, +) if TYPE_CHECKING: from collections.abc import Sequence @@ -203,6 +208,14 @@ def _log_advertised_tools( Registered at ``INFO`` because operators routinely tail this; the delta at ``DEBUG`` because it's noisy on fully-implemented handlers. + + Also fires a one-time ``UserWarning`` at boot when the handler + class introduces a new specialism (a custom subclass that's not in + the framework's tool registry and doesn't declare + ``advertised_tools``) but ``advertise_all`` is False — closes the + silent over-advertisement gap where adopters see the full + ``ADCPHandler`` tool surface inherited via MRO when they meant to + declare a focused subset. """ registered_set = set(registered) full_defs = get_tools_for_handler(handler, advertise_all=True) @@ -219,6 +232,87 @@ def _log_advertised_tools( if unadvertised and not advertise_all: logger.debug("%s server unadvertised tools: %s", transport, ", ".join(unadvertised)) + # Stacklevel walks: warnings.warn → _warn_if_unregistered_subclass → + # _log_advertised_tools → operator's call site. The MCP path adds one + # extra frame (_register_handler_tools); A2A calls _log_advertised_tools + # directly from create_a2a_server. + caller_stacklevel = 4 if transport == "mcp" else 3 + _warn_if_unregistered_subclass( + handler, advertise_all=advertise_all, stacklevel=caller_stacklevel + ) + + +#: Bases whose tool set is broad-by-design — when an adopter subclass +#: lands on one of these via MRO without registering its own +#: ``advertised_tools``, the result is over-advertisement (the broad +#: base's full set inherited unintentionally). Naming the rule rather +#: than checking ``base.__name__ != "ADCPHandler"`` inline so future +#: broad bases (a hypothetical ``UniversalHandler``) get added to one +#: place — and a reviewer's first question becomes "is this base +#: broad-by-design?" not "what's special about ADCPHandler?". +_BROAD_SURFACE_BASES: frozenset[str] = frozenset({"ADCPHandler"}) + + +def _warn_if_unregistered_subclass( + handler: ADCPHandler[Any], *, advertise_all: bool, stacklevel: int = 4 +) -> None: + """Emit a one-time ``UserWarning`` when a custom handler base bypasses + the tool-discovery registry. + + The trigger: the concrete handler class itself isn't in + ``_HANDLER_TOOLS``, has no ``advertised_tools`` declaration of its + own, and inherits its tool set from a broad-surface base (see + :data:`_BROAD_SURFACE_BASES`) rather than a specialized base like + ``GovernanceHandler``. That combination almost always means the + adopter meant to declare a focused tool set but forgot to register + it; the framework over-advertises by silently falling through to + the broad base's full surface. + + Suppressed when ``advertise_all=True`` — that's the explicit "yes, + advertise everything" opt-in. + """ + if advertise_all: + return + cls = type(handler) + if cls.__name__ in _HANDLER_TOOLS: + return + if "advertised_tools" in cls.__dict__: + # Should already have been auto-registered via __init_subclass__, + # but defensively skip the warning if the attribute exists. + return + # Walk MRO looking for a specialized (non-broad-surface) SDK base. + # If one is found, the adopter is subclassing a focused base and + # inheriting its tool set — that's the documented pattern, no + # warning needed. + has_specialized_parent = any( + base.__name__ in _HANDLER_TOOLS and base.__name__ not in _BROAD_SURFACE_BASES + for base in cls.__mro__ + ) + if has_specialized_parent: + return + # Default stacklevel=4 covers the MCP path (warn → this fn → + # _log_advertised_tools → _register_handler_tools → caller). The A2A + # path lacks _register_handler_tools and passes stacklevel=3. + warnings.warn( + f"Handler class {cls.__name__!r} subclasses ADCPHandler directly " + f"but isn't registered in the framework's tool-discovery " + f"registry. tools/list will inherit the full ADCPHandler tool " + f"surface — this almost always means over-advertising for a " + f"new specialism.\n\n" + f"Pick one:\n" + f" (a) declare ``advertised_tools: set[str] = {{...}}`` on " + f"{cls.__name__} (auto-registers via __init_subclass__)\n" + f" (b) call adcp.server.mcp_tools.register_handler_tools(" + f"{cls.__name__!r}, {{...}}) before serve()\n" + f" (c) pass advertise_all=True to serve() to acknowledge the " + f"full advertisement\n\n" + f"Decisioning-platform adopters: codegen via " + f"`uv run python scripts/generate_decisioning_handler.py` " + f"emits the declaration for you.", + UserWarning, + stacklevel=stacklevel, + ) + async def _dispatch_with_middleware( middleware: tuple[SkillMiddleware, ...] | Sequence[SkillMiddleware], diff --git a/tests/test_advertised_tools_gate.py b/tests/test_advertised_tools_gate.py index 45f877898..f8d68f2ec 100644 --- a/tests/test_advertised_tools_gate.py +++ b/tests/test_advertised_tools_gate.py @@ -369,25 +369,39 @@ async def get_products(self, params, context=None): # --------------------------------------------------------------------------- -# Coupling invariant — _SDK_BASE_CLASS_NAMES must stay in sync with _HANDLER_TOOLS +# SDK-base-class detection reads _HANDLER_TOOLS live (no frozen-set drift) # --------------------------------------------------------------------------- -def test_sdk_base_class_names_stays_in_sync_with_handler_tools(): - """``_SDK_BASE_CLASS_NAMES`` is derived from ``_HANDLER_TOOLS.keys()`` - and the override detector walks the MRO looking for name matches. If - someone adds a new specialized base (e.g. ``RetailMediaHandler``) to - ``_HANDLER_TOOLS`` without touching ``_SDK_BASE_CLASS_NAMES``, the - detector will skip it and silently mis-classify every subclass's - overrides. This test locks the two names together. +def test_is_sdk_base_class_reads_handler_tools_live(): + """``_is_sdk_base_class`` reads ``_HANDLER_TOOLS`` directly so handlers + registered after import time (via :func:`register_handler_tools` or + :meth:`ADCPHandler.__init_subclass__`) participate in override + detection without rebuilding any cached set. Regression coverage for + the prior frozen-set drift bug. """ - from adcp.server.mcp_tools import _HANDLER_TOOLS, _SDK_BASE_CLASS_NAMES - - assert set(_SDK_BASE_CLASS_NAMES) == set(_HANDLER_TOOLS.keys()), ( - "_SDK_BASE_CLASS_NAMES must be derived from _HANDLER_TOOLS.keys(). " - "Adding a new specialized handler base requires updating both." + from adcp.server.mcp_tools import ( + _HANDLER_TOOLS, + _is_sdk_base_class, + register_handler_tools, ) + # Built-in bases recognised. + for name in _HANDLER_TOOLS: + assert _is_sdk_base_class(name), name + + # Unknown name rejected. + assert not _is_sdk_base_class("DefinitelyNotAHandler") + + # Newly registered name picked up immediately, no rebuild. + register_handler_tools("_TestLiveDetectionHandler", {"get_products"}) + try: + assert _is_sdk_base_class("_TestLiveDetectionHandler") + finally: + # Test-cleanup — remove the synthetic registration so other + # tests don't see drift. + _HANDLER_TOOLS.pop("_TestLiveDetectionHandler", None) + # --------------------------------------------------------------------------- # Agent card reflects the filter diff --git a/tests/test_register_handler_tools.py b/tests/test_register_handler_tools.py new file mode 100644 index 000000000..eac353562 --- /dev/null +++ b/tests/test_register_handler_tools.py @@ -0,0 +1,344 @@ +"""Tests for the public ``register_handler_tools`` seam. + +Covers: + +* :func:`adcp.server.mcp_tools.register_handler_tools` idempotency, + conflict detection, unknown-tool typo recovery, and integration with + :func:`adcp.server.mcp_tools.get_tools_for_handler`. +* :meth:`adcp.server.base.ADCPHandler.__init_subclass__` auto-registration + from the ``advertised_tools`` class attribute. +* The boot-time ``UserWarning`` in + :func:`adcp.server.serve._warn_if_unregistered_subclass`. + +Tests intentionally use direct imports of the (private) +``_HANDLER_TOOLS`` registry for cleanup. Production code reaches it +only through ``register_handler_tools``. +""" + +from __future__ import annotations + +import warnings +from typing import Any, ClassVar + +import pytest + +from adcp.server import ADCPHandler +from adcp.server.mcp_tools import ( + _HANDLER_TOOLS, + get_tools_for_handler, + register_handler_tools, +) +from adcp.server.serve import _warn_if_unregistered_subclass + + +@pytest.fixture +def cleanup_handler_registry(): + """Snapshot ``_HANDLER_TOOLS`` and restore on teardown so tests don't + leak synthetic registrations into each other. + """ + snapshot = {k: set(v) for k, v in _HANDLER_TOOLS.items()} + yield + _HANDLER_TOOLS.clear() + _HANDLER_TOOLS.update(snapshot) + + +# ---- register_handler_tools — idempotency + conflict detection ---- + + +def test_register_handler_tools_idempotent_on_equal_input(cleanup_handler_registry): + """Calling twice with the same set is a no-op — module re-imports + and reloadable test harnesses don't break.""" + register_handler_tools("_TestIdempotent", {"get_products", "create_media_buy"}) + register_handler_tools("_TestIdempotent", {"create_media_buy", "get_products"}) + assert _HANDLER_TOOLS["_TestIdempotent"] == {"get_products", "create_media_buy"} + + +def test_register_handler_tools_raises_on_conflicting_set(cleanup_handler_registry): + """Calling twice with different sets raises so accidental drift is + caught at registration, not at first dispatch.""" + register_handler_tools("_TestConflict", {"get_products"}) + with pytest.raises(ValueError, match="called twice"): + register_handler_tools("_TestConflict", {"create_media_buy"}) + + +def test_register_handler_tools_rejects_unknown_tool(cleanup_handler_registry): + """Unknown tool names raise — adopters can't smuggle non-spec + surface through the registry.""" + with pytest.raises(ValueError, match="unknown tool"): + register_handler_tools("_TestUnknown", {"definitely_not_a_real_tool"}) + + +def test_register_handler_tools_suggests_close_match(cleanup_handler_registry): + """Typo recovery — ``difflib.get_close_matches`` surfaces the most + likely intended name on the error message so adopters working from + spec memory can fix the typo without a manual scan.""" + with pytest.raises(ValueError, match="did you mean 'create_media_buy'"): + register_handler_tools("_TestTypo", {"create_media_buyy"}) + + +def test_register_handler_tools_integrates_with_get_tools_for_handler( + cleanup_handler_registry, +): + """After registration, ``get_tools_for_handler`` returns the + registered set (filtered by override-detection — direct + ``ADCPHandler`` subclass with override of one tool advertises that + one tool plus protocol/discovery).""" + + class _CustomBase(ADCPHandler): + async def get_products(self, params, context=None): + return {"products": []} + + register_handler_tools("_CustomBase", {"get_products", "create_media_buy"}) + tools = {t["name"] for t in get_tools_for_handler(_CustomBase())} + assert "get_products" in tools + # create_media_buy is in the registered set but the subclass didn't + # override it, so override-detection filters it out (advertise_all=False). + assert "create_media_buy" not in tools + + +def test_register_handler_tools_advertise_all_returns_full_registered_set( + cleanup_handler_registry, +): + """``advertise_all=True`` skips the override filter and returns + every registered tool — exactly the explicit "yes, advertise + everything" opt-in.""" + + class _AdvertiseAllBase(ADCPHandler): + pass + + register_handler_tools("_AdvertiseAllBase", {"get_products", "create_media_buy"}) + tools = {t["name"] for t in get_tools_for_handler(_AdvertiseAllBase(), advertise_all=True)} + assert "get_products" in tools + assert "create_media_buy" in tools + + +# ---- ADCPHandler.__init_subclass__ — auto-registration via class attr ---- + + +def test_init_subclass_auto_registers_from_advertised_tools(cleanup_handler_registry): + """A subclass declaring ``advertised_tools`` as a class attribute + is auto-registered at class creation time — codegen targets emit + this and adopters never need to call ``register_handler_tools`` + explicitly.""" + + class _AutoRegisteredHandler(ADCPHandler): + advertised_tools: ClassVar[set[str]] = {"get_products", "create_media_buy"} + + assert "_AutoRegisteredHandler" in _HANDLER_TOOLS + assert _HANDLER_TOOLS["_AutoRegisteredHandler"] == { + "get_products", + "create_media_buy", + } + + +def test_init_subclass_skips_inherited_advertised_tools(cleanup_handler_registry): + """Inherited ``advertised_tools`` does NOT trigger re-registration — + multi-level subclasses don't shadow their parent's registered set + with the same set under a new class name.""" + + class _ParentBase(ADCPHandler): + advertised_tools: ClassVar[set[str]] = {"get_products"} + + class _ChildBase(_ParentBase): + # No advertised_tools of its own. Inherits via MRO but + # __init_subclass__ checks cls.__dict__ specifically so this + # subclass isn't registered. + pass + + assert "_ParentBase" in _HANDLER_TOOLS + assert "_ChildBase" not in _HANDLER_TOOLS + + +def test_init_subclass_three_level_chain_with_intermediate_declaration( + cleanup_handler_registry, +): + """Three-level inheritance where the *intermediate* base declares + ``advertised_tools`` registers AT the intermediate level. The leaf + inherits via MRO and isn't separately registered. Regression for + middle-of-the-chain registration.""" + + class _Root(ADCPHandler): + # No advertised_tools — root level. + pass + + class _Intermediate(_Root): + # Declares its own — registers at this level. + advertised_tools: ClassVar[set[str]] = {"get_products", "create_media_buy"} + + class _Leaf(_Intermediate): + # Inherits from intermediate; no own declaration. + pass + + assert "_Root" not in _HANDLER_TOOLS + assert "_Intermediate" in _HANDLER_TOOLS + assert _HANDLER_TOOLS["_Intermediate"] == {"get_products", "create_media_buy"} + assert "_Leaf" not in _HANDLER_TOOLS + + +def test_init_subclass_forwards_kwargs_to_super(cleanup_handler_registry): + """``__init_subclass__`` calls ``super().__init_subclass__(**kwargs)`` + so PEP 487-style metaclass kwargs (e.g. from + ``__init_subclass__`` declared on a custom metaclass) flow through. + Without this, a future subclass using ``class X(ADCPHandler, mixin_kw=...)`` + would lose the kwarg silently. Regression coverage by declaring a + bystander base that inspects ``cls`` on subclass creation.""" + seen: list[type] = [] + + class _Bystander: + def __init_subclass__(cls, **kwargs: Any) -> None: + super().__init_subclass__(**kwargs) + seen.append(cls) + + class _MultiInheritHandler(ADCPHandler, _Bystander): + advertised_tools: ClassVar[set[str]] = {"get_products"} + + # Bystander's __init_subclass__ saw the new class — proves the + # super() chain is intact even with ADCPHandler.__init_subclass__ + # in the chain. Without `super().__init_subclass__(**kwargs)`, the + # bystander would never fire. + assert _MultiInheritHandler in seen + + +def test_register_handler_tools_accepts_empty_iterable(cleanup_handler_registry): + """An empty tool set is allowed — represents "this handler claims + zero tool surface." Registering it makes the handler-name known to + the registry without claiming any AdCP verbs. Useful for handlers + that exist purely for typed test-context plumbing.""" + register_handler_tools("_EmptySetHandler", set()) + assert _HANDLER_TOOLS["_EmptySetHandler"] == set() + + +def test_init_subclass_idempotent_on_module_reload(cleanup_handler_registry): + """Re-evaluating the same class body (test reload, ipython rerun) + re-triggers ``__init_subclass__`` with the same tool set — must not + raise. Idempotency on equal input.""" + + def _define(): + class _ReloadHandler(ADCPHandler): + advertised_tools: ClassVar[set[str]] = {"get_products"} + + return _ReloadHandler + + _define() + _define() # would raise on conflict if registry were strict + assert "_ReloadHandler" in _HANDLER_TOOLS + + +# ---- serve() boot UserWarning ---- + + +def test_warn_if_unregistered_subclass_fires_on_unregistered_direct_subclass( + cleanup_handler_registry, +): + """An ``ADCPHandler`` subclass with no registry entry, no + ``advertised_tools`` declaration, and no specialized parent + triggers the boot-time UserWarning.""" + + class _UnregisteredHandler(ADCPHandler): + pass + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always", UserWarning) + _warn_if_unregistered_subclass(_UnregisteredHandler(), advertise_all=False) + matched = [w for w in caught if "_UnregisteredHandler" in str(w.message)] + assert len(matched) == 1 + assert "advertised_tools" in str(matched[0].message) + assert "advertise_all=True" in str(matched[0].message) + + +def test_warn_if_unregistered_subclass_suppressed_by_advertise_all( + cleanup_handler_registry, +): + """``advertise_all=True`` silences the warning — explicit opt-in to + full advertisement.""" + + class _IntentionallyAllHandler(ADCPHandler): + pass + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always", UserWarning) + _warn_if_unregistered_subclass(_IntentionallyAllHandler(), advertise_all=True) + matched = [w for w in caught if "_IntentionallyAllHandler" in str(w.message)] + assert matched == [] + + +def test_warn_if_unregistered_subclass_suppressed_when_advertised_tools_set( + cleanup_handler_registry, +): + """A subclass declaring ``advertised_tools`` is auto-registered via + ``__init_subclass__``, so the warning never fires for it.""" + + class _DeclaresAdvertisedHandler(ADCPHandler): + advertised_tools: ClassVar[set[str]] = {"get_products"} + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always", UserWarning) + _warn_if_unregistered_subclass(_DeclaresAdvertisedHandler(), advertise_all=False) + matched = [w for w in caught if "_DeclaresAdvertisedHandler" in str(w.message)] + assert matched == [] + + +def test_warn_if_unregistered_subclass_suppressed_for_specialized_parent( + cleanup_handler_registry, +): + """Subclassing a specialized base (e.g. ``GovernanceHandler``) is + the documented pattern; the subclass inherits the parent's tool + set via MRO and no warning should fire even when the subclass + itself isn't separately registered.""" + from adcp.server import GovernanceHandler + + # Concrete subclass providing the handle_* methods GovernanceHandler + # declares abstract. Implementations are minimal — the test only + # cares that the class instantiates so we can call the warning + # check on an instance. + class _CustomGovernanceAgent(GovernanceHandler): + async def handle_check_governance(self, params, context=None): + return {} + + async def handle_report_plan_outcome(self, params, context=None): + return {} + + async def handle_get_plan_audit_logs(self, params, context=None): + return {} + + async def handle_sync_plans(self, params, context=None): + return {} + + async def handle_get_creative_features(self, params, context=None): + return {} + + async def handle_create_property_list(self, params, context=None): + return {} + + async def handle_get_property_list(self, params, context=None): + return {} + + async def handle_list_property_lists(self, params, context=None): + return {} + + async def handle_update_property_list(self, params, context=None): + return {} + + async def handle_delete_property_list(self, params, context=None): + return {} + + async def handle_create_collection_list(self, params, context=None): + return {} + + async def handle_get_collection_list(self, params, context=None): + return {} + + async def handle_list_collection_lists(self, params, context=None): + return {} + + async def handle_update_collection_list(self, params, context=None): + return {} + + async def handle_delete_collection_list(self, params, context=None): + return {} + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always", UserWarning) + _warn_if_unregistered_subclass(_CustomGovernanceAgent(), advertise_all=False) + matched = [w for w in caught if "_CustomGovernanceAgent" in str(w.message)] + assert matched == []