feat(client): mark local refusals as undispatched and classify every error's recovery

src/adcp/__init__.py

@@ -272,6 +272,8 @@ def _resolve_version() -> str:
         "AccountPaymentRequiredError",
         "ConfigurationError",
         "extract_adcp_error_info",
+        "is_pre_dispatch_error",
+        "is_pre_dispatch_error",
         "IdempotencyConflictError",
         "IdempotencyExpiredError",
         "IdempotencyScopeError",
@@ -853,6 +855,7 @@ def _resolve_version() -> str:
         "uses_deprecated_assets_field",
     ),
     "adcp.validation": (
+        "RequestValidationError",
         "SchemaValidationError",
         "UnknownFieldPolicy",
         "ValidationError",
@@ -1652,6 +1655,7 @@ def get_adcp_version() -> str:
     "RegistryErrorDetails",
     "RegistryValidationIssue",
     # Validation utilities
+    "RequestValidationError",
     "SchemaValidationError",
     "UnknownFieldPolicy",
     "ValidationError",
@@ -1902,6 +1906,7 @@ def get_adcp_version() -> str:
         RegistryErrorDetails,
         RegistryValidationIssue,
         extract_adcp_error_info,
+        is_pre_dispatch_error,
     )
     from adcp.feed_mirror import (
         EventHandler,
@@ -2576,6 +2581,7 @@ def get_adcp_version() -> str:
         uses_deprecated_assets_field,
     )
     from adcp.validation import (
+        RequestValidationError,
         SchemaValidationError,
         UnknownFieldPolicy,
         ValidationError,

src/adcp/capabilities.py

@@ -304,6 +304,7 @@ def require(
             declared_features=declared,
             agent_id=agent_id,
             agent_uri=agent_uri,
+            dispatched=False,  # a check against capabilities already held
         )
 
     def get_declared_features(self) -> list[str]:
@@ -377,7 +378,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 '{method_name}' method"
                 )
                 continue
 

src/adcp/client.py

@@ -1082,6 +1082,9 @@ async def _prepare_versioned_task(self, tool: str, params: dict[str, Any]) -> Co
         try:
             return await self._convert_outgoing_request(tool, params)
         except ADCPFeatureUnsupportedError as exc:
+            # Version negotiation and request conversion run before any of
+            # the task's request is sent.
+            exc.dispatched = False
             message = str(exc)
             details: dict[str, Any] = {"wire_response_received": False}
             for name in ("pointer", "source_version", "target_version", "reason"):
@@ -1093,6 +1096,7 @@ async def _prepare_versioned_task(self, tool: str, params: dict[str, Any]) -> Co
                 error=message,
                 adcp_error={"code": "UNSUPPORTED_FEATURE", "message": message, "details": details},
                 idempotency_key=params.get("idempotency_key"),
+                dispatched=False,
                 metadata={
                     CONVERSION_METADATA_KEY: {
                         "wire_version": details.get("target_version"),

src/adcp/compat/reporting_version.py

@@ -168,5 +168,6 @@ def assert_reporting_request_supported(tool_name: str, params: Any) -> None:
         [
             f"{issue.detail} at {issue.field} requires AdCP 3.2-beta.6 or newer "
             f"(target is {target_version}); no request was sent"
-        ]
+        ],
+        dispatched=False,
     )

src/adcp/compat/versions.py

@@ -189,7 +189,9 @@ def negotiate_version(supported: list[str], *, maximum: str) -> str:
         and is_adcp_version_at_least(maximum, _release(version))
     ]
     if not common:
-        raise ADCPFeatureUnsupportedError(["No common AdCP version with this seller"])
+        raise ADCPFeatureUnsupportedError(
+            ["No common AdCP version with this seller"], dispatched=False
+        )
     selected = common[0]
     for candidate in common[1:]:
         candidate_release, selected_release = _release(candidate), _release(selected)
@@ -271,7 +273,9 @@ def _score(schema: dict[str, Any], value: Any, root: dict[str, Any]) -> int:
                     else (
                         "string"
                         if isinstance(value, str)
-                        else "number" if isinstance(value, (int, float)) else "object"
+                        else "number"
+                        if isinstance(value, (int, float))
+                        else "object"
                     )
                 )
             )

src/adcp/exceptions.py

@@ -129,8 +129,45 @@ def extract_adcp_error_info(err: Any) -> AdcpErrorInfo:
     )
 
 
+def is_pre_dispatch_error(error: object) -> bool:
+    """Whether the SDK refused this request locally, before sending any of it.
+
+    Such errors carry ``dispatched = False``: no bytes of the task's request
+    left this process, so the agent cannot have acted on it, and the failure
+    says nothing about the agent's health. The SDK sets the marker only on
+    refusals it decides by itself: request schema validation, the version and
+    feature gates, and refusals that follow already-known capabilities.
+
+    Failures that involve contacting the agent are never marked, even when the
+    task itself was not sent: endpoint discovery, OAuth token fetches and the
+    ``get_adcp_capabilities`` fetch. A refused connection, a 5xx or a 429 there
+    is an agent failure.
+
+    This checks the attribute, not the class, so it also accepts a
+    :class:`~adcp.types.core.TaskResult` carrying ``dispatched=False`` and an
+    error from another copy of the SDK.
+    """
+    return getattr(error, "dispatched", None) is False
+
+
 class ADCPError(Exception):
-    """Base exception for all AdCP client errors."""
+    """Base exception for all AdCP client errors.
+
+    Attributes:
+        dispatched: ``False`` when the SDK refused the request locally and no
+            bytes of the task's request left this process (see
+            :func:`is_pre_dispatch_error`). ``None`` otherwise: the request
+            may have been sent.
+        recovery: How to recover, in the AdCP ``error.recovery`` vocabulary:
+            ``"transient"`` (retry later), ``"correctable"`` (fix the request
+            or configuration, then retry) or ``"terminal"`` (needs human
+            action). ``None`` when the SDK cannot classify the failure.
+            :class:`ADCPTimeoutError` keeps its own ``recovery``: the
+            :class:`~adcp.task_options.TaskRecoveryMetadata` for a safe retry.
+    """
+
+    dispatched: Literal[False] | None = None
+    recovery: RecoveryLiteral | None = None
 
     def __init__(
         self,
@@ -164,6 +201,8 @@ def is_retryable(self) -> bool:
 class ADCPConnectionError(ADCPError):
     """Connection to agent failed."""
 
+    recovery: RecoveryLiteral | None = "transient"
+
     def __init__(self, message: str, agent_id: str | None = None, agent_uri: str | None = None):
         """Initialize connection error."""
         suggestion = (
@@ -194,6 +233,8 @@ class ADCPAuthenticationError(ADCPError):
     for credential recovery.
     """
 
+    recovery: RecoveryLiteral | None = "correctable"
+
     def __init__(
         self,
         message: str,
@@ -233,7 +274,10 @@ def __init__(
         self.timeout = timeout
         self.task_name = task_name
         self.operation_id = operation_id
-        self.recovery = recovery
+        # The safe-retry identity, not the base class's classification: a
+        # timeout is always retryable (see ``is_retryable``), and
+        # ``recovery.outcome_unknown`` says the mutation may have applied.
+        self.recovery = recovery  # type: ignore[assignment]
         suggestion = (
             f"The request took longer than {timeout}s." if timeout else "The request timed out."
         )
@@ -479,14 +523,24 @@ def __init__(
 
 
 class ADCPFeatureUnsupportedError(ADCPError):
-    """Seller does not support one or more required features."""
+    """Seller does not support one or more required features.
+
+    ``dispatched`` is ``False`` where the SDK refused the task from
+    already-known capabilities or its own version and feature gates, before
+    sending any of it. It is ``None`` where the refusal follows a response,
+    for example a response that cannot be converted to the caller's version.
+    """
+
+    recovery: RecoveryLiteral | None = "correctable"
 
     def __init__(
         self,
         unsupported_features: list[str],
         declared_features: list[str] | None = None,
         agent_id: str | None = None,
         agent_uri: str | None = None,
+        *,
+        dispatched: Literal[False] | None = None,
     ):
         """Initialize feature unsupported error.
 
@@ -495,7 +549,11 @@ def __init__(
             declared_features: Features the seller does declare.
             agent_id: Optional agent ID for context.
             agent_uri: Optional agent URI for context.
+            dispatched: Pass ``False`` only where no bytes of the task's
+                request left the process. See :func:`is_pre_dispatch_error`.
         """
+        if dispatched is False:
+            self.dispatched = False
         self.unsupported_features = unsupported_features
         self.declared_features = declared_features or []
 
@@ -639,6 +697,11 @@ def __init__(
         self.error_codes = [
             code for err in errors if isinstance((code := _access(err, "code")), str) and code
         ]
+        recoveries = self._effective_recoveries()
+        # The most demanding entry decides: terminal over correctable over transient.
+        self.recovery = next(
+            (r for r in ("terminal", "correctable", "transient") if r in recoveries), None
+        )
 
         message = f"{operation} failed"
         if errors:
@@ -703,6 +766,11 @@ def is_retryable(self) -> bool:
         `correctable` entry blocks (caller must fix the request first — a
         retry as-is would re-trigger the same error).
         """
+        effective = self._effective_recoveries()
+        return bool(effective) and all(r == "transient" for r in effective)
+
+    def _effective_recoveries(self) -> list[RecoveryLiteral]:
+        """Each entry's recovery: the wire value, else the code table, else transient."""
         from adcp.server.helpers import CORRECTABLE_CODES, TERMINAL_CODES, TRANSIENT_CODES
 
         effective: list[RecoveryLiteral] = []
@@ -725,10 +793,7 @@ def is_retryable(self) -> bool:
                 # against a producer that ships a new code before the SDK
                 # learns about it.
                 effective.append("transient")
-
-        if not effective:
-            return False
-        return not any(r in ("terminal", "correctable") for r in effective)
+        return effective
 
 
 class IdempotencyConflictError(ADCPTaskError):
@@ -827,9 +892,13 @@ class IdempotencyUnsupportedError(ADCPError):
     declares ``supported=False``, or declares ``supported=True`` without a
     ``replay_ttl_seconds`` window. Per AdCP spec, clients MUST NOT assume a
     default — a seller that does not positively declare support cannot be
-    safely retried.
+    safely retried. The mutating task is never sent (``dispatched`` is
+    ``False``).
     """
 
+    dispatched: Literal[False] | None = False
+    recovery: RecoveryLiteral | None = "correctable"
+
     def __init__(
         self,
         agent_id: str | None = None,

src/adcp/protocols/a2a.py

@@ -476,6 +476,7 @@ async def _call_a2a_tool_wire(
                 error=str(exc),
                 success=False,
                 idempotency_key=idempotency_key,
+                dispatched=False,
             )
 
         if tool_name != "get_adcp_capabilities" and self.signing_capability_check:

src/adcp/protocols/mcp.py

@@ -934,8 +934,7 @@ def _log_cleanup_error(self, exc: BaseException, context: str) -> None:
             and ("cancel scope" in exc_str or "async context" in exc_str)
         ) or (
             # HTTP errors during cleanup (if httpx is available)
-            HTTPX_AVAILABLE
-            and isinstance(exc, _ALL_HTTP_STATUS_ERROR_TYPES)
+            HTTPX_AVAILABLE and isinstance(exc, _ALL_HTTP_STATUS_ERROR_TYPES)
         )
 
         if is_known_cleanup_error:
@@ -1231,6 +1230,7 @@ async def _call_mcp_tool_wire(self, tool_name: str, params: dict[str, Any]) -> T
                     error=str(exc),
                     success=False,
                     idempotency_key=idempotency_key,
+                    dispatched=False,
                 )
 
             # Streamable HTTP sends from a long-lived writer task whose

src/adcp/types/core.py

@@ -265,6 +265,11 @@ class TaskResult(BaseModel, Generic[T]):
     # Agents that emit side effects on success (notifications, memory writes,
     # downstream tool calls) must check this flag and suppress duplicates.
     replayed: bool = False
+    # False when the SDK refused the task locally and no bytes of its request
+    # left the process (request schema validation, version and feature gates):
+    # the failure says nothing about the agent's health. None otherwise. See
+    # ``adcp.exceptions.is_pre_dispatch_error``.
+    dispatched: Literal[False] | None = None
 
 
 class ActivityType(str, Enum):

src/adcp/validation/__init__.py

@@ -55,6 +55,7 @@
     list_validator_keys,
 )
 from adcp.validation.schema_validator import (
+    RequestValidationError,
     SchemaValidationError,
     ValidationIssue,
     ValidationOutcome,
@@ -74,6 +75,7 @@
     # Schema core
     "Direction",
     "ResponseVariant",
+    "RequestValidationError",
     "SchemaValidationError",
     "ValidationIssue",
     "ValidationOutcome",

src/adcp/validation/schema_errors.py

@@ -7,6 +7,7 @@
 from typing import Any
 
 from adcp.validation.schema_validator import (
+    RequestValidationError,
     SchemaValidationError,
     ValidationIssue,
     _issue_to_wire,
@@ -43,8 +44,11 @@ def build_validation_error(
     """Build a :class:`SchemaValidationError` carrying every failure.
 
     Strict-mode client hooks raise this so callers can inspect the full
-    pointer list via ``.issues`` and the ``details`` dict.
+    pointer list via ``.issues`` and the ``details`` dict. A request-side
+    failure is a :class:`RequestValidationError` (``dispatched`` is ``False``).
     """
+    if side == "request":
+        return RequestValidationError(tool, issues)
     return SchemaValidationError(tool, side, issues)