diff --git a/src/adcp/__init__.py b/src/adcp/__init__.py
index 737680099..2445170d2 100644
--- a/src/adcp/__init__.py
+++ b/src/adcp/__init__.py
@@ -25,6 +25,7 @@
 from adcp.capabilities import (  # noqa: F401
     FeatureResolver,
     build_synthetic_capabilities,
+    looks_like_v3_capabilities,
     validate_capabilities,
 )
 from adcp.client import ADCPClient, ADCPMultiAgentClient, Checkpoint
diff --git a/src/adcp/capabilities.py b/src/adcp/capabilities.py
index a1502ae4d..f0248a1b0 100644
--- a/src/adcp/capabilities.py
+++ b/src/adcp/capabilities.py
@@ -87,6 +87,70 @@
 del _task, _feature
 
 
+def _is_plain_object(value: Any) -> bool:
+    """Return True iff ``value`` is a non-array dict.
+
+    Mirrors the JS ``isPlainObject`` helper used by ``looks_like_v3_capabilities``.
+    Excludes lists so that ``adcp: []`` or ``media_buy: []`` don't get mistaken
+    for v3 envelope blocks just because ``isinstance(_, dict)`` would have
+    happened to return False anyway — kept for symmetry with the JS check
+    so future contributors don't reintroduce a ``isinstance(_, (dict, list))``
+    false-positive.
+    """
+    return isinstance(value, dict)
+
+
+def looks_like_v3_capabilities(data: Any) -> bool:
+    """Heuristic: does this ``get_adcp_capabilities`` response look v3-shaped?
+
+    Used by ``ADCPClient.refresh_capabilities`` when the response fails strict
+    schema validation but is structurally non-empty. The question the heuristic
+    answers is "is this a v3 agent with a wire-shape bug, or a v2 agent that
+    happens to advertise the tool?". Falling back to v2 in the former case
+    masks the original bug behind cascading v2.5-schema-not-found errors;
+    treating it as v3 surfaces the wire-shape bug at its source.
+
+    Affirmative v3 signals (any one is enough):
+
+    - ``adcp`` block (only v3 servers carry the
+      ``{ major_versions, idempotency, ... }`` envelope)
+    - ``supported_protocols`` array (v3-only top-level field)
+    - any v3 protocol-level capability block (``account``, ``media_buy``,
+      ``signals``, ``creative``, ``brand``, ``governance``,
+      ``sponsored_intelligence``, ``compliance_testing``)
+
+    v2 servers don't expose ``get_adcp_capabilities`` at all (the tool itself
+    is a v3-only addition), so reaching this function with a non-empty payload
+    already strongly implies v3 — but the structural check belt-and-suspenders
+    against genuinely empty / null responses.
+
+    Args:
+        data: Raw response payload (typically a dict, but accepts any value
+            so callers don't have to narrow before calling).
+
+    Returns:
+        True if any v3 signal is present; False for empty, null, non-dict,
+        or shape-mismatched inputs.
+    """
+    if not _is_plain_object(data):
+        return False
+    if _is_plain_object(data.get("adcp")):
+        return True
+    if isinstance(data.get("supported_protocols"), list):
+        return True
+    v3_blocks = (
+        "account",
+        "media_buy",
+        "signals",
+        "creative",
+        "brand",
+        "governance",
+        "sponsored_intelligence",
+        "compliance_testing",
+    )
+    return any(_is_plain_object(data.get(block)) for block in v3_blocks)
+
+
 def build_synthetic_capabilities(
     supported_protocols: list[str],
     *,
@@ -171,7 +235,7 @@ def supports(self, feature: str) -> bool:
 
         # Targeting check: "targeting.geo_countries"
         if feature.startswith("targeting."):
-            attr_name = feature[len("targeting."):]
+            attr_name = feature[len("targeting.") :]
             if caps.media_buy is None or caps.media_buy.execution is None:
                 return False
             targeting = caps.media_buy.execution.targeting
@@ -307,8 +371,7 @@ def validate_capabilities(
         for method_name in handler_methods:
             if not hasattr(handler, method_name):
                 warnings.append(
-                    f"Feature '{feature}' is declared but handler has no "
-                    f"'{method_name}' method"
+                    f"Feature '{feature}' is declared but handler has no " f"'{method_name}' method"
                 )
                 continue
 
diff --git a/src/adcp/client.py b/src/adcp/client.py
index 9ac23b874..73f149408 100644
--- a/src/adcp/client.py
+++ b/src/adcp/client.py
@@ -22,7 +22,7 @@
     from mcp import ClientSession
 
 from adcp._version import resolve_adcp_version
-from adcp.capabilities import TASK_FEATURE_MAP, FeatureResolver
+from adcp.capabilities import TASK_FEATURE_MAP, FeatureResolver, looks_like_v3_capabilities
 from adcp.exceptions import ADCPError, ADCPWebhookSignatureError
 from adcp.protocols.a2a import A2AAdapter
 from adcp.protocols.base import ProtocolAdapter
@@ -1042,8 +1042,22 @@ async def fetch_capabilities(self) -> GetAdcpCapabilitiesResponse:
     async def refresh_capabilities(self) -> GetAdcpCapabilitiesResponse:
         """Fetch capabilities from the seller, bypassing cache.
 
+        On strict-schema validation failure the raw response is inspected with
+        ``looks_like_v3_capabilities``: if the agent is structurally v3-shaped,
+        a wire-shape bug is surfaced loudly with the original validation error
+        rather than silently downgrading to v2 (the v2 fallback would then ask
+        for v2.5 schemas, which aren't shipped — one missing field would
+        cascade into "AdCP schema data for version v2.5 not found"). Genuinely
+        non-v3 responses still fall through to the transport-error path.
+
         Returns:
             The seller's capabilities response.
+
+        Raises:
+            ADCPError: On transport failure, or when the response is
+                v3-shaped but fails schema validation. The error message
+                explicitly references v3 in the latter case so the underlying
+                wire-shape bug doesn't get blamed on a v2.5-schema cascade.
         """
         result = await self.get_adcp_capabilities(GetAdcpCapabilitiesRequest())
         if result.success and result.data is not None:
@@ -1051,6 +1065,48 @@ async def refresh_capabilities(self) -> GetAdcpCapabilitiesResponse:
             self._feature_resolver = FeatureResolver(result.data)
             self._capabilities_fetched_at = time.monotonic()
             return self._capabilities
+
+        # The typed call discards the raw payload on parse failure (only the
+        # error string survives). Distinguish parse-failure (worth shape-
+        # checking) from transport-failure (no data ever arrived) by the
+        # error prefix produced by ProtocolAdapter._parse_response. Only on
+        # parse-failure do we re-fetch the raw dict from the adapter to
+        # inspect its shape; transport failures fall straight through to
+        # the original error path.
+        raw_data: Any = None
+        is_parse_failure = result.error is not None and result.error.startswith(
+            "Failed to parse response:"
+        )
+        if is_parse_failure:
+            raw_result = await self.adapter.get_adcp_capabilities(
+                GetAdcpCapabilitiesRequest().model_dump(mode="json", exclude_none=True)
+            )
+            raw_data = raw_result.data
+            if isinstance(raw_data, list) and len(raw_data) == 1 and isinstance(raw_data[0], dict):
+                # MCP content array — unwrap a single-item content envelope
+                # so the heuristic sees the same shape the parser would.
+                raw_data = raw_data[0]
+
+        if looks_like_v3_capabilities(raw_data):
+            logger.warning(
+                "[AdCP] Agent %r returned a get_adcp_capabilities response that "
+                "failed validation, but the response is structurally v3-shaped. "
+                "The agent has a wire-shape bug — that's the thing to fix. "
+                "(has_error=%s, has_data=%s)",
+                self.agent_config.id,
+                bool(result.error),
+                raw_data is not None,
+            )
+            raise ADCPError(
+                f"v3 capabilities response from agent {self.agent_config.id!r} "
+                f"failed schema validation: {result.error or result.message}. "
+                f"The response is structurally v3-shaped (carries `adcp`, "
+                f"`supported_protocols`, or a v3 protocol block) — fix the "
+                f"agent's wire shape rather than downgrading to v2.",
+                agent_id=self.agent_config.id,
+                agent_uri=self.agent_config.agent_uri,
+            )
+
         raise ADCPError(
             f"Failed to fetch capabilities: {result.error or result.message}",
             agent_id=self.agent_config.id,
diff --git a/tests/test_looks_like_v3.py b/tests/test_looks_like_v3.py
new file mode 100644
index 000000000..6e144364f
--- /dev/null
+++ b/tests/test_looks_like_v3.py
@@ -0,0 +1,277 @@
+"""Tests for the v3-shape detection heuristic and its wire-up in
+``ADCPClient.refresh_capabilities``.
+
+The heuristic exists so a single failed schema validation on
+``get_adcp_capabilities`` doesn't silently re-classify a v3 agent as v2 —
+which downstream tooling turns into the cascade of confusing
+"AdCP schema data for version v2.5 not found" errors. Port of JS commit
+27bd79d (#1201). See issue #461.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from adcp import ADCPClient
+from adcp.capabilities import looks_like_v3_capabilities
+from adcp.exceptions import ADCPError
+from adcp.types.core import AgentConfig, Protocol, TaskResult, TaskStatus
+
+
+def _make_config() -> AgentConfig:
+    return AgentConfig(
+        id="test-seller",
+        agent_uri="https://seller.example.com",
+        protocol=Protocol.A2A,
+    )
+
+
+# ============================================================================
+# looks_like_v3_capabilities — pure heuristic
+# ============================================================================
+
+
+class TestLooksLikeV3CapabilitiesPositive:
+    """The heuristic recognizes any one v3-shape signal."""
+
+    def test_detects_adcp_envelope_block(self):
+        assert looks_like_v3_capabilities({"adcp": {"major_versions": [3]}}) is True
+
+    def test_detects_supported_protocols_array(self):
+        assert looks_like_v3_capabilities({"supported_protocols": ["signals"]}) is True
+
+    def test_detects_empty_supported_protocols_array(self):
+        # Field presence alone is a v3 signal — v2 doesn't have this top-level field.
+        assert looks_like_v3_capabilities({"supported_protocols": []}) is True
+
+    def test_detects_account_block(self):
+        assert looks_like_v3_capabilities({"account": {"require_operator_auth": True}}) is True
+
+    def test_detects_media_buy_block(self):
+        assert looks_like_v3_capabilities({"media_buy": {"features": {}}}) is True
+
+    def test_detects_signals_block(self):
+        assert looks_like_v3_capabilities({"signals": {"catalog_signals": True}}) is True
+
+    def test_detects_creative_block(self):
+        assert looks_like_v3_capabilities({"creative": {"supports_compliance": True}}) is True
+
+    def test_detects_brand_block(self):
+        assert looks_like_v3_capabilities({"brand": {"rights": True}}) is True
+
+    def test_detects_governance_block(self):
+        assert looks_like_v3_capabilities({"governance": {"spend_authority": True}}) is True
+
+    def test_detects_sponsored_intelligence_block(self):
+        assert looks_like_v3_capabilities({"sponsored_intelligence": {"offerings": []}}) is True
+
+    def test_detects_compliance_testing_block(self):
+        assert looks_like_v3_capabilities({"compliance_testing": {"scenarios": []}}) is True
+
+    def test_detects_partial_v3_response_with_one_missing_field(self):
+        """The exact case that surfaced this issue: a v3 agent missing one
+        required field — adcp envelope present, supported_protocols present,
+        account present but missing supported_billing.
+        """
+        assert (
+            looks_like_v3_capabilities(
+                {
+                    "adcp": {"major_versions": [3]},
+                    "supported_protocols": ["signals"],
+                    "account": {"require_operator_auth": True},
+                }
+            )
+            is True
+        )
+
+
+class TestLooksLikeV3CapabilitiesNegative:
+    """The heuristic rejects empty / non-dict / shape-mismatched inputs."""
+
+    def test_rejects_none(self):
+        assert looks_like_v3_capabilities(None) is False
+
+    def test_rejects_empty_dict(self):
+        assert looks_like_v3_capabilities({}) is False
+
+    def test_rejects_list(self):
+        assert looks_like_v3_capabilities([]) is False
+
+    def test_rejects_string(self):
+        assert looks_like_v3_capabilities("v3") is False
+
+    def test_rejects_number(self):
+        assert looks_like_v3_capabilities(42) is False
+
+    def test_rejects_object_with_only_unknown_fields(self):
+        assert looks_like_v3_capabilities({"foo": "bar", "baz": 1}) is False
+
+    def test_rejects_supported_protocols_when_not_an_array(self):
+        # Malformed — string instead of array. Don't promote to v3.
+        assert looks_like_v3_capabilities({"supported_protocols": "signals"}) is False
+
+    def test_rejects_v3_block_when_null(self):
+        assert looks_like_v3_capabilities({"media_buy": None}) is False
+
+    def test_rejects_adcp_when_null(self):
+        assert looks_like_v3_capabilities({"adcp": None}) is False
+
+    def test_rejects_adcp_when_array(self):
+        # Defensive: arrays are not plain objects.
+        assert looks_like_v3_capabilities({"adcp": []}) is False
+
+    def test_rejects_v3_block_when_array(self):
+        assert looks_like_v3_capabilities({"media_buy": []}) is False
+
+
+# ============================================================================
+# refresh_capabilities — wire-up
+# ============================================================================
+
+
+def _success_capabilities_dict() -> dict:
+    """A clean v3 capabilities dict that passes strict schema validation."""
+    return {
+        "adcp": {
+            "major_versions": [3],
+            "idempotency": {"supported": True, "replay_ttl_seconds": 86400},
+        },
+        "supported_protocols": ["media_buy"],
+    }
+
+
+def _broken_v3_capabilities_dict() -> dict:
+    """A response that is structurally v3-shaped but fails strict validation
+    — has the v3 envelope but a non-validating supported_protocols entry.
+    """
+    return {
+        "adcp": {
+            "major_versions": [3],
+            "idempotency": {"supported": True, "replay_ttl_seconds": 86400},
+        },
+        "supported_protocols": ["not-a-real-protocol"],
+        "account": {"require_operator_auth": True},
+    }
+
+
+def _v2_shaped_response_dict() -> dict:
+    """A response that has none of the v3 signals — heuristic returns False."""
+    return {"some_legacy_field": "value", "tools": []}
+
+
+class TestRefreshCapabilitiesV3Detection:
+    """Wire-up: refresh_capabilities surfaces v3 validation errors loudly."""
+
+    @pytest.mark.asyncio
+    async def test_clean_v3_response_parses_unchanged(self):
+        """Clean v3 capabilities → fetched and cached, no warnings."""
+        client = ADCPClient(_make_config())
+
+        raw = TaskResult(
+            status=TaskStatus.COMPLETED,
+            data=_success_capabilities_dict(),
+            success=True,
+        )
+        with patch.object(
+            client.adapter, "get_adcp_capabilities", new_callable=AsyncMock
+        ) as mock_adapter:
+            mock_adapter.return_value = raw
+
+            caps = await client.refresh_capabilities()
+
+        assert caps is not None
+        # Cached.
+        assert client.capabilities is caps
+        assert client.feature_resolver is not None
+
+    @pytest.mark.asyncio
+    async def test_broken_v3_response_raises_loud_v3_error(self):
+        """V3-shaped response with a validation bug → loud v3 error.
+
+        The error message must reference v3 explicitly so downstream users
+        don't waste time looking for a v2.5-schema-not-found cascade — the
+        underlying problem is a wire-shape bug in the agent.
+        """
+        client = ADCPClient(_make_config())
+        broken_dict = _broken_v3_capabilities_dict()
+
+        # Adapter returns the broken dict twice: once for the typed call (which
+        # fails validation downstream), once for the re-fetch that lets the
+        # heuristic inspect the raw shape.
+        raw = TaskResult(
+            status=TaskStatus.COMPLETED,
+            data=broken_dict,
+            success=True,
+        )
+        with patch.object(
+            client.adapter, "get_adcp_capabilities", new_callable=AsyncMock
+        ) as mock_adapter:
+            mock_adapter.return_value = raw
+
+            with pytest.raises(ADCPError) as exc_info:
+                await client.refresh_capabilities()
+
+        msg = str(exc_info.value)
+        # Must reference v3 explicitly — that's the whole point.
+        assert "v3" in msg
+        # Must NOT trigger the v2.5-schema-not-found cascade or claim v2.5.
+        assert "v2.5" not in msg
+        # Must reference schema validation so the fix-it path is obvious.
+        assert "validation" in msg.lower() or "schema" in msg.lower()
+
+    @pytest.mark.asyncio
+    async def test_genuine_v2_response_falls_through_to_transport_error(self):
+        """Genuine non-v3 response on validation failure → existing error path.
+
+        This is the "still downgrades correctly" lane: the heuristic returns
+        False, so the v3-loud branch is skipped and the existing
+        "Failed to fetch capabilities" error is raised. (The Python client
+        doesn't ship a v2-synthetic auto-fallback inside refresh_capabilities;
+        callers wanting one use ``build_synthetic_capabilities`` directly.)
+        """
+        client = ADCPClient(_make_config())
+        v2_dict = _v2_shaped_response_dict()
+
+        raw = TaskResult(
+            status=TaskStatus.COMPLETED,
+            data=v2_dict,
+            success=True,
+        )
+        with patch.object(
+            client.adapter, "get_adcp_capabilities", new_callable=AsyncMock
+        ) as mock_adapter:
+            mock_adapter.return_value = raw
+
+            with pytest.raises(ADCPError) as exc_info:
+                await client.refresh_capabilities()
+
+        msg = str(exc_info.value)
+        # Must NOT claim v3 — heuristic correctly identified this as not-v3.
+        assert "v3" not in msg
+        # Hits the original transport-style error.
+        assert "Failed to fetch capabilities" in msg
+
+    @pytest.mark.asyncio
+    async def test_transport_failure_with_no_data_still_raises(self):
+        """Original transport-failure path is preserved (no re-fetch attempt)."""
+        client = ADCPClient(_make_config())
+
+        failed = TaskResult(
+            status=TaskStatus.FAILED,
+            data=None,
+            success=False,
+            error="Connection refused",
+        )
+        with patch.object(
+            client.adapter, "get_adcp_capabilities", new_callable=AsyncMock
+        ) as mock_adapter:
+            mock_adapter.return_value = failed
+
+            with pytest.raises(ADCPError, match="Failed to fetch capabilities"):
+                await client.refresh_capabilities()
+
+        # Adapter is called exactly once — no second probe for raw shape on
+        # transport-style failures (no data ever arrived).
+        assert mock_adapter.call_count == 1