feat(compat): server-side shape-based v2.5 detection (stage 6)

src/adcp/compat/legacy/__init__.py

@@ -32,6 +32,12 @@
 #: Versions handled via the legacy-adapter path. Distinct from
 #: ``COMPATIBLE_ADCP_VERSIONS`` in :mod:`adcp._version`, which lists the
 #: versions the SDK natively validates against.
+#:
+#: **Iteration order is probe-precedence order.** When the dispatcher
+#: runs shape-based detection (Stage 6, ``is_legacy_shape``), it walks
+#: this tuple and breaks on the first match. List versions
+#: highest-to-lowest priority — newer legacy versions before older —
+#: so an ambiguous payload routes to the most-recent matching adapter.
 LEGACY_ADAPTER_VERSIONS: Final[tuple[str, ...]] = ("2.5",)
 
 # Per-version adapter module list. Data, not control flow, so adding a

src/adcp/compat/legacy/types.py

@@ -54,3 +54,15 @@ class AdapterPair:
     tool_name: str
     adapt_request: Callable[[dict[str, Any]], dict[str, Any]]
     normalize_response: Callable[[dict[str, Any]], dict[str, Any]] | None = None
+    # Optional shape probe used by the dispatcher when the buyer didn't
+    # send an ``adcp_version`` / ``adcp_major_version`` envelope (real v2.5
+    # buyers can't — the field didn't exist in the v2.5 schema). The
+    # probe should return ``True`` only on strong, unambiguous v2.5
+    # markers — fields that exist in v2.5 but NOT in v3 (e.g.
+    # ``brand_manifest``, ``creative_ids`` in packages, bare-string
+    # ``format_id``). False positives downgrade a real v3 buyer to v2.5
+    # validation, which is the worst outcome; bias conservatively. Tools
+    # with pass-through requests (``list_creative_formats``,
+    # ``preview_creative``) leave this ``None`` because their request
+    # shape is identical across versions.
+    is_legacy_shape: Callable[[dict[str, Any]], bool] | None = None

src/adcp/compat/legacy/v2_5/create_media_buy.py

@@ -40,9 +40,27 @@ def adapt_request(payload: dict[str, Any]) -> dict[str, Any]:
     return out
 
 
+def is_legacy_shape(payload: dict[str, Any]) -> bool:
+    """v2.5 ``create_media_buy`` markers:
+
+    * top-level ``brand_manifest`` URL (v3 uses ``brand: {domain}``)
+    * any package with ``creative_ids: list[str]`` (v3 uses
+      ``creative_assignments: [{creative_id, ...}]``)
+    """
+    if "brand_manifest" in payload:
+        return True
+    packages = payload.get("packages")
+    if isinstance(packages, list):
+        for pkg in packages:
+            if isinstance(pkg, dict) and isinstance(pkg.get("creative_ids"), list):
+                return True
+    return False
+
+
 ADAPTER = AdapterPair(
     tool_name="create_media_buy",
     adapt_request=adapt_request,
     normalize_response=normalize_media_buy_response,
+    is_legacy_shape=is_legacy_shape,
 )
 register_adapter("2.5", ADAPTER)

src/adcp/compat/legacy/v2_5/get_products.py

@@ -229,9 +229,18 @@ def normalize_response(response: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def is_legacy_shape(payload: dict[str, Any]) -> bool:
+    """v2.5 ``get_products`` carries either ``brand_manifest`` (URL
+    string field that v3 doesn't have) or ``promoted_offerings``
+    (nested object replaced by ``catalog`` in v3). Either is a
+    strong signal."""
+    return "brand_manifest" in payload or "promoted_offerings" in payload
+
+
 ADAPTER = AdapterPair(
     tool_name="get_products",
     adapt_request=adapt_request,
     normalize_response=normalize_response,
+    is_legacy_shape=is_legacy_shape,
 )
 register_adapter("2.5", ADAPTER)

src/adcp/compat/legacy/v2_5/sync_creatives.py

@@ -133,5 +133,23 @@ def adapt_request(payload: dict[str, Any]) -> dict[str, Any]:
     return out
 
 
-ADAPTER = AdapterPair(tool_name="sync_creatives", adapt_request=adapt_request)
+def is_legacy_shape(payload: dict[str, Any]) -> bool:
+    """v2.5 ``sync_creatives`` has at least one creative whose
+    ``format_id`` is a bare string (v3 always emits the structured
+    ``{agent_url, id}`` form). Strong signal — v3 wouldn't emit this
+    even with ``adcp_version`` omitted."""
+    creatives = payload.get("creatives")
+    if not isinstance(creatives, list):
+        return False
+    for creative in creatives:
+        if isinstance(creative, dict) and isinstance(creative.get("format_id"), str):
+            return True
+    return False
+
+
+ADAPTER = AdapterPair(
+    tool_name="sync_creatives",
+    adapt_request=adapt_request,
+    is_legacy_shape=is_legacy_shape,
+)
 register_adapter("2.5", ADAPTER)

src/adcp/compat/legacy/v2_5/update_media_buy.py

@@ -36,9 +36,22 @@ def adapt_request(payload: dict[str, Any]) -> dict[str, Any]:
     return out
 
 
+def is_legacy_shape(payload: dict[str, Any]) -> bool:
+    """v2.5 ``update_media_buy``: any package with ``creative_ids:
+    list[str]`` is the marker (no ``brand_manifest`` on updates)."""
+    packages = payload.get("packages")
+    if not isinstance(packages, list):
+        return False
+    for pkg in packages:
+        if isinstance(pkg, dict) and isinstance(pkg.get("creative_ids"), list):
+            return True
+    return False
+
+
 ADAPTER = AdapterPair(
     tool_name="update_media_buy",
     adapt_request=adapt_request,
     normalize_response=normalize_media_buy_response,
+    is_legacy_shape=is_legacy_shape,
 )
 register_adapter("2.5", ADAPTER)

src/adcp/server/mcp_tools.py

@@ -2027,14 +2027,47 @@ async def call_tool(params: dict[str, Any], context: ToolContext | None = None)
             )
             wire_version = None
 
-        # Legacy-version routing: if the buyer claims a version handled
-        # via the adapter path (e.g. ``"2.5"``), validate the params
-        # against the legacy schema first, *then* translate to the
-        # current shape. Pre-adapter validation surfaces structural
-        # errors with the legacy schema's field paths — far easier
-        # for the buyer to act on than a v3 field-path error after a
-        # confusing translation. Post-adapter validation (further
-        # down) catches translator bugs against the SDK pin.
+        # Shape-based legacy detection (issue: real v2.5 buyers can't
+        # send ``adcp_version`` — the field didn't exist in the v2.5
+        # schema). When the envelope is empty and a legacy adapter
+        # registers an ``is_legacy_shape`` probe, run it. A match
+        # promotes ``wire_version`` to the probe's version so the
+        # adapter path below fires normally. Bias is conservative:
+        # probes return ``True`` only on fields v3 doesn't emit
+        # (``brand_manifest``, ``creative_ids``, bare-string
+        # ``format_id``). False positives downgrade a real v3 buyer
+        # to legacy validation, which is the worst outcome.
+        if wire_version is None:
+            for candidate in LEGACY_ADAPTER_VERSIONS:
+                candidate_adapter = get_legacy_adapter(candidate, method_name)
+                if candidate_adapter is None:
+                    continue
+                probe = candidate_adapter.is_legacy_shape
+                if probe is None:
+                    continue
+                try:
+                    matched = probe(params) if isinstance(params, dict) else False
+                except Exception:  # noqa: BLE001 — defensive: probes are pure-ish
+                    matched = False
+                if matched:
+                    logger.info(
+                        "Detected %s wire shape for %s (no envelope version "
+                        "supplied); routing through legacy adapter.",
+                        candidate,
+                        method_name,
+                    )
+                    wire_version = candidate
+                    break
+
+        # Legacy-version routing: if the buyer claims (or shape-detected)
+        # a version handled via the adapter path (e.g. ``"2.5"``),
+        # validate the params against the legacy schema first, *then*
+        # translate to the current shape. Pre-adapter validation
+        # surfaces structural errors with the legacy schema's field
+        # paths — far easier for the buyer to act on than a v3
+        # field-path error after a confusing translation. Post-adapter
+        # validation (further down) catches translator bugs against
+        # the SDK pin.
         legacy_adapter: Any = None
         if wire_version in LEGACY_ADAPTER_VERSIONS:
             legacy_adapter = get_legacy_adapter(wire_version, method_name)