feat(validation): add ADCP_VALIDATION_MODE env var for TS-SDK parity

src/adcp/client.py

@@ -377,11 +377,18 @@ def __init__(
                 JSON schemas. Defaults (matching the TS port): requests
                 in ``warn`` mode (drift logged but not blocked — partial
                 payloads in error-path tests still work) and responses
-                in ``strict`` mode (agent drift fails the task). The
-                response mode flips to ``warn`` when any of ``ADCP_ENV``
-                / ``PYTHON_ENV`` / ``ENV`` / ``ENVIRONMENT`` is set to
-                ``production`` / ``prod``. Storyboards and compliance
-                runners that want hard-stop enforcement everywhere pass
+                in ``strict`` mode (agent drift fails the task).
+                Override both sides at once with the ``ADCP_VALIDATION_MODE``
+                env var (``strict`` / ``warn`` / ``off``); per-side
+                defaults still apply for whichever side an explicit
+                ``ValidationHookConfig`` field overrides.
+                Legacy: the response mode alone flips to ``warn`` when
+                ``ADCP_ENV`` is set to ``production`` / ``prod``.
+                Only ``ADCP_ENV`` and ``ADCP_VALIDATION_MODE`` are
+                consulted — generic ``ENV`` / ``ENVIRONMENT`` / ``PYTHON_ENV``
+                are ignored to avoid collisions with unrelated tooling.
+                Storyboards and compliance runners that want hard-stop
+                enforcement everywhere pass
                 ``validation=ValidationHookConfig(requests="strict",
                 responses="strict")``; high-throughput callers can set
                 either side to ``"off"`` to skip the validator entirely

src/adcp/validation/client_hooks.py

@@ -8,7 +8,7 @@
 import os
 from dataclasses import dataclass
 from datetime import datetime, timezone
-from typing import Any, Literal, TypedDict
+from typing import Any, Literal, TypedDict, cast
 
 from adcp.validation.schema_errors import build_validation_error
 from adcp.validation.schema_validator import (
@@ -22,6 +22,8 @@
 
 ValidationMode = Literal["strict", "warn", "off"]
 
+_VALID_MODES: frozenset[str] = frozenset({"strict", "warn", "off"})
+
 
 @dataclass(frozen=True)
 class ValidationHookConfig:
@@ -59,6 +61,28 @@ class DebugLogEntry(TypedDict, total=False):
     issues: list[dict[str, Any]]
 
 
+def _read_validation_mode_env() -> ValidationMode | None:
+    """Read ``ADCP_VALIDATION_MODE``; raise ``ValueError`` on invalid values; ``None`` if unset.
+
+    Read at call time (not import time) so tests that ``patch.dict`` the
+    environment work without a module-level reset hook.
+
+    Raises ``ValueError`` immediately at client-construction time (when
+    ``resolve_validation_modes`` is called) so misconfigured deployments
+    fail loudly rather than silently using the wrong mode.
+    """
+    val = os.environ.get("ADCP_VALIDATION_MODE")
+    if val is None:
+        return None
+    normalized = val.lower()
+    if normalized not in _VALID_MODES:
+        raise ValueError(
+            f"ADCP_VALIDATION_MODE={val!r} is not valid. "
+            "Accepted values: strict, warn, off."
+        )
+    return cast(ValidationMode, normalized)
+
+
 def _default_response_mode() -> ValidationMode:
     """Response default: ``strict`` unless ``ADCP_ENV`` declares production.
 
@@ -74,11 +98,21 @@ def _default_response_mode() -> ValidationMode:
 def resolve_validation_modes(
     config: ValidationHookConfig | None = None,
 ) -> tuple[ValidationMode, ValidationMode]:
-    """Return the effective ``(requests, responses)`` modes."""
-    req: ValidationMode = (config.requests if config is not None else None) or "warn"
+    """Return the effective ``(requests, responses)`` modes.
+
+    Precedence (highest to lowest):
+    1. Explicit config field (``ValidationHookConfig.requests`` / ``.responses``)
+    2. ``ADCP_VALIDATION_MODE`` env var — applies to both sides when set
+    3. ``ADCP_ENV`` legacy fallback — response side only (``production`` → ``warn``)
+    4. Hard defaults: requests ``"warn"``, responses ``"strict"``
+    """
+    env_mode = _read_validation_mode_env()
+    # The `or` chain is safe here: none of the three valid mode strings
+    # ("strict", "warn", "off") are falsy, so `or` acts as null-coalescing.
+    req: ValidationMode = (config.requests if config is not None else None) or env_mode or "warn"
     resp: ValidationMode = (
         config.responses if config is not None else None
-    ) or _default_response_mode()
+    ) or env_mode or _default_response_mode()
     return req, resp
 
 

tests/test_schema_validation_client.py

@@ -15,6 +15,7 @@
 
 from adcp import ADCPClient, ValidationHookConfig
 from adcp.types import AgentConfig, Protocol
+from adcp.validation.client_hooks import _read_validation_mode_env, resolve_validation_modes
 
 
 def _agent_config() -> AgentConfig:
@@ -116,3 +117,56 @@ async def test_off_mode_skips_validator(self) -> None:
 
         assert result.success is True
         assert result.data == {"products": "oops"}
+
+
+class TestAdcpValidationModeEnvVar:
+    """Tests for ADCP_VALIDATION_MODE env var (issue #385)."""
+
+    @pytest.mark.parametrize("mode", ["strict", "warn", "off"])
+    def test_env_var_applies_to_both_sides(self, mode: str) -> None:
+        with patch.dict(os.environ, {"ADCP_VALIDATION_MODE": mode}, clear=True):
+            req, resp = resolve_validation_modes()
+        assert req == mode
+        assert resp == mode
+
+    def test_env_var_case_insensitive(self) -> None:
+        with patch.dict(os.environ, {"ADCP_VALIDATION_MODE": "STRICT"}, clear=True):
+            req, resp = resolve_validation_modes()
+        assert req == "strict"
+        assert resp == "strict"
+
+    def test_invalid_env_var_raises_value_error(self) -> None:
+        with patch.dict(os.environ, {"ADCP_VALIDATION_MODE": "verbose"}, clear=True):
+            with pytest.raises(ValueError, match="ADCP_VALIDATION_MODE"):
+                _read_validation_mode_env()
+
+    def test_explicit_config_field_wins_over_env_var(self) -> None:
+        with patch.dict(os.environ, {"ADCP_VALIDATION_MODE": "off"}, clear=True):
+            req, resp = resolve_validation_modes(
+                ValidationHookConfig(requests="strict", responses="warn")
+            )
+        assert req == "strict"
+        assert resp == "warn"
+
+    def test_adcp_validation_mode_wins_over_adcp_env(self) -> None:
+        # ADCP_VALIDATION_MODE=strict should override the ADCP_ENV→warn fallback
+        # on the response side.
+        with patch.dict(
+            os.environ,
+            {"ADCP_VALIDATION_MODE": "strict", "ADCP_ENV": "production"},
+            clear=True,
+        ):
+            _, resp = resolve_validation_modes()
+        assert resp == "strict"
+
+    def test_unset_env_var_keeps_defaults(self) -> None:
+        with patch.dict(os.environ, {}, clear=True):
+            req, resp = resolve_validation_modes()
+        assert req == "warn"
+        assert resp == "strict"
+
+    def test_adcp_validation_mode_env_var_on_client(self) -> None:
+        with patch.dict(os.environ, {"ADCP_VALIDATION_MODE": "off"}, clear=True):
+            client = ADCPClient(_agent_config())
+        assert client.adapter.request_validation_mode == "off"
+        assert client.adapter.response_validation_mode == "off"