diff --git a/docs/install_and_use.md b/docs/install_and_use.md
index d091a2ac..4dde8745 100644
--- a/docs/install_and_use.md
+++ b/docs/install_and_use.md
@@ -42,3 +42,8 @@ print(validators.email("someone@example.com"))
         raise ValidationError(func, _func_args_as_dict(func, *args, **kwargs))
     validators.utils.ValidationError: ValidationError(func=visa, args={'value': 'bad_visa_number'})
     ```
+
+The environment option raises at the outermost validation call. Nested validators
+return validation errors so a composite validator can try alternatives, such as
+checking an IP address after a domain check fails. Passing `r_ve=True` explicitly
+still raises even inside another validator.
diff --git a/src/validators/utils.py b/src/validators/utils.py
index 77937f68..8a7a38bb 100644
--- a/src/validators/utils.py
+++ b/src/validators/utils.py
@@ -1,13 +1,45 @@
 """Utils."""
 
 # standard
+from contextvars import ContextVar
+from dataclasses import dataclass
 from functools import wraps
 from inspect import getfullargspec
 from itertools import chain
 from os import environ
+from sys import modules
+from threading import get_ident
 from typing import Any, Callable, Dict
 
 
+@dataclass
+class _ValidationScope:
+    """Track the lifetime, thread and task of an outermost validation call."""
+
+    thread_id: int
+    task_id: int | None
+    active: bool = True
+
+
+_validation_scope: ContextVar[_ValidationScope | None] = ContextVar(
+    "_validation_scope", default=None
+)
+
+
+def _current_task_id() -> int | None:
+    """Identify an active task without loading asyncio for synchronous callers."""
+    if "asyncio" not in modules:
+        return None
+
+    from asyncio import current_task
+
+    try:
+        task = current_task()
+    except RuntimeError:
+        return None
+    return id(task) if task is not None else None
+
+
 class ValidationError(Exception):
     """Exception class when validation failure occurs."""
 
@@ -48,6 +80,9 @@ def validator(func: Callable[..., Any]):
     Whenever the given `func` returns `False` this
     decorator returns `ValidationError` object.
 
+    The environment option raises at the outermost validation call so nested
+    validators can try alternatives. An explicit `r_ve=True` always raises.
+
     Examples:
         >>> @validator
         ... def even(value):
@@ -73,8 +108,21 @@ def validator(func: Callable[..., Any]):
     @wraps(func)
     def wrapper(*args: Any, **kwargs: Any):
         raise_validation_error = bool(kwargs.pop("r_ve", False))
-        if environ.get("RAISE_VALIDATION_ERROR", "False") == "True":
-            raise_validation_error = True
+        scope = _validation_scope.get()
+        thread_id = get_ident()
+        task_id = _current_task_id()
+        if (
+            scope is None
+            or not scope.active
+            or scope.thread_id != thread_id
+            or scope.task_id != task_id
+        ):
+            if environ.get("RAISE_VALIDATION_ERROR", "False") == "True":
+                raise_validation_error = True
+            scope = _ValidationScope(thread_id, task_id)
+            token = _validation_scope.set(scope)
+        else:
+            token = None
 
         try:
             if raise_validation_error:
@@ -95,5 +143,9 @@ def wrapper(*args: Any, **kwargs: Any):
                 ) from exp
             else:
                 return ValidationError(func, _func_args_as_dict(func, *args, **kwargs), str(exp))
+        finally:
+            if token is not None:
+                scope.active = False
+                _validation_scope.reset(token)
 
     return wrapper
diff --git a/tests/test_validation_failure.py b/tests/test_validation_failure.py
index df308410..0a182554 100644
--- a/tests/test_validation_failure.py
+++ b/tests/test_validation_failure.py
@@ -1,13 +1,19 @@
 """Test validation Failure."""
 
 # standard
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+from contextvars import Context, copy_context
+import sys
+from threading import Event
 from typing import Any, Callable
 
 # third-party
 import pytest
 
 # local
-from validators import ValidationError, between, slug, uuid
+from validators import ValidationError, between, hostname, slug, url, uuid
+from validators.utils import validator
 
 
 @pytest.mark.parametrize("function,value", [(slug, "invalid_"), (uuid, "not-a-uuid")])
@@ -41,6 +47,176 @@ def test_true_raise_validation_error_flag(
         function(value, r_ve=True)
 
 
+@pytest.mark.parametrize("global_raise", [False, True])
+@pytest.mark.parametrize(
+    "function,value",
+    [
+        (hostname, "127.0.0.1"),
+        (hostname, "::1"),
+        (hostname, "127.0.0.1:8080"),
+        (hostname, "[::1]:8080"),
+        (url, "http://10.0.0.1"),
+        (url, "http://[::1]"),
+    ],
+)
+def test_raise_mode_accepts_valid_composite_values(
+    function: Callable[..., Any],
+    value: str,
+    global_raise: bool,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Both raise modes allow failed alternatives inside a valid composite validator."""
+    monkeypatch.setenv("RAISE_VALIDATION_ERROR", str(global_raise))
+    assert function(value, r_ve=not global_raise) is True
+
+
+@pytest.mark.parametrize("global_raise", [False, True])
+@pytest.mark.parametrize(
+    "function,value", [(hostname, "not a hostname"), (url, "http://not a hostname")]
+)
+def test_raise_mode_reports_invalid_composite_at_public_boundary(
+    function: Callable[..., Any],
+    value: str,
+    global_raise: bool,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Invalid composite inputs raise an error for the validator the caller invoked."""
+    monkeypatch.setenv("RAISE_VALIDATION_ERROR", str(global_raise))
+    with pytest.raises(ValidationError) as exc_info:
+        function(value, r_ve=not global_raise)
+    assert exc_info.value.func.__name__ == function.__name__
+    assert exc_info.value.__dict__["value"] == value
+
+
+def test_nested_explicit_raise_flag_is_respected(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A nested caller can still explicitly request an exception."""
+    monkeypatch.setenv("RAISE_VALIDATION_ERROR", "True")
+
+    @validator
+    def explicit_nested_raise(value: str):
+        return slug(value, r_ve=True)
+
+    with pytest.raises(ValidationError) as exc_info:
+        explicit_nested_raise("invalid_")
+    assert exc_info.value.func.__name__ == "slug"
+
+
+@pytest.mark.parametrize("global_raise", [False, True])
+def test_raise_mode_is_independent_between_threads(
+    global_raise: bool, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A validation in another thread must not suppress this caller's raise mode."""
+    monkeypatch.setenv("RAISE_VALIDATION_ERROR", str(global_raise))
+    entered = Event()
+    release = Event()
+
+    @validator
+    def waiting_validation():
+        entered.set()
+        assert release.wait(timeout=5)
+        return True
+
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        pending = executor.submit(waiting_validation, r_ve=not global_raise)
+        try:
+            assert entered.wait(timeout=5)
+            with pytest.raises(ValidationError):
+                slug("invalid_", r_ve=not global_raise)
+        finally:
+            release.set()
+        assert pending.result(timeout=5) is True
+
+
+def test_copied_context_is_independent_in_another_thread(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Copying an active context to another thread must not suppress its raise mode."""
+    monkeypatch.setenv("RAISE_VALIDATION_ERROR", "True")
+
+    def validate_on_worker() -> bool:
+        try:
+            slug("invalid_")
+        except ValidationError:
+            return True
+        return False
+
+    @validator
+    def validation_with_worker():
+        context = copy_context()
+        with ThreadPoolExecutor(max_workers=1) as executor:
+            pending = executor.submit(context.run, validate_on_worker)
+            assert pending.result(timeout=5)
+        return True
+
+    assert validation_with_worker() is True
+
+
+@pytest.mark.parametrize("eager", [False, True])
+def test_inherited_task_context_does_not_suppress_independent_validation(
+    eager: bool,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An inherited task context must not suppress independent synchronous validation."""
+    if eager and sys.version_info < (3, 12):
+        pytest.skip("Eager task factories require Python 3.12 or newer")
+    monkeypatch.setenv("RAISE_VALIDATION_ERROR", "True")
+
+    async def validate_later() -> None:
+        with pytest.raises(ValidationError):
+            slug("invalid_")
+
+    async def run_validation() -> None:
+        if eager and sys.version_info >= (3, 12):
+            asyncio.get_running_loop().set_task_factory(asyncio.eager_task_factory)
+        tasks: list[asyncio.Task[None]] = []
+
+        @validator
+        def schedule_validation():
+            assert hostname("127.0.0.1") is True
+            assert isinstance(slug("invalid_"), ValidationError)
+            tasks.append(asyncio.create_task(validate_later()))
+            assert hostname("::1") is True
+            return True
+
+        assert schedule_validation() is True
+        await tasks[0]
+
+    asyncio.run(run_validation())
+
+
+@pytest.mark.parametrize("outcome", ["valid", "invalid", "parse_error", "unexpected_error"])
+def test_global_raise_mode_is_restored_after_validation(
+    outcome: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Returning or raising from a validator must not suppress a later independent call."""
+    monkeypatch.setenv("RAISE_VALIDATION_ERROR", "True")
+    contexts: list[Context] = []
+
+    @validator
+    def capture_context():
+        contexts.append(copy_context())
+        return True
+
+    @validator
+    def sample_validation():
+        assert capture_context() is True
+        assert isinstance(contexts[0].run(slug, "invalid_"), ValidationError)
+        if outcome == "unexpected_error":
+            raise RuntimeError("unexpected")
+        if outcome == "parse_error":
+            raise ValueError("parse error")
+        return outcome == "valid"
+
+    if outcome == "valid":
+        assert sample_validation() is True
+    else:
+        expected_error = RuntimeError if outcome == "unexpected_error" else ValidationError
+        with pytest.raises(expected_error):
+            sample_validation()
+    with pytest.raises(ValidationError):
+        slug("invalid_")
+    with pytest.raises(ValidationError):
+        contexts[0].run(slug, "invalid_")
+
+
 failed_obj_repr = "ValidationError(func=between"