diff --git a/reflexio/cli/commands/services.py b/reflexio/cli/commands/services.py
index 9e1c0a78f..232ab8c15 100644
--- a/reflexio/cli/commands/services.py
+++ b/reflexio/cli/commands/services.py
@@ -14,6 +14,7 @@
 from reflexio.cli import run_services as run_mod
 from reflexio.cli import stop_services as stop_mod
 from reflexio.cli.bootstrap_config import _VALID_STORAGE_BACKENDS
+from reflexio.server.env_utils import env_truthy
 
 _logger = logging.getLogger(__name__)
 
@@ -228,9 +229,8 @@ def start(
     # (e.g. REFLEXIO_STORAGE=supabase) are visible to the resolution chain.
     load_reflexio_env()
 
-    if (
-        not os.environ.get("REFLEXIO_EMBEDDING_PROVIDER")
-        and os.environ.get("CLAUDE_SMART_USE_LOCAL_EMBEDDING") == "1"
+    if not os.environ.get("REFLEXIO_EMBEDDING_PROVIDER") and env_truthy(
+        os.environ.get("CLAUDE_SMART_USE_LOCAL_EMBEDDING", "")
     ):
         os.environ["REFLEXIO_EMBEDDING_PROVIDER"] = "local_service"
 
diff --git a/reflexio/cli/commands/setup_cmd.py b/reflexio/cli/commands/setup_cmd.py
index 0f4d5fe9e..e8b025084 100644
--- a/reflexio/cli/commands/setup_cmd.py
+++ b/reflexio/cli/commands/setup_cmd.py
@@ -17,6 +17,8 @@
 
 import typer
 
+from reflexio.server.env_utils import env_truthy
+
 app = typer.Typer(
     help=(
         "Configure Reflexio: run 'init' for plain CLI setup, use 'openclaw' "
@@ -188,7 +190,7 @@ def _is_non_interactive() -> bool:
     - stdin is not a TTY (``nohup``, container, pipe-fed shell)
     - ``REFLEXIO_NONINTERACTIVE=1`` in the environment (explicit opt-out)
     """
-    if os.environ.get("REFLEXIO_NONINTERACTIVE") == "1":
+    if env_truthy(os.environ.get("REFLEXIO_NONINTERACTIVE", "")):
         return True
     return not sys.stdin.isatty()
 
diff --git a/reflexio/integrations/openclaw/plugin/src/openclaw_smart/internal_call.py b/reflexio/integrations/openclaw/plugin/src/openclaw_smart/internal_call.py
index ac8af8bbb..584a182ab 100644
--- a/reflexio/integrations/openclaw/plugin/src/openclaw_smart/internal_call.py
+++ b/reflexio/integrations/openclaw/plugin/src/openclaw_smart/internal_call.py
@@ -23,6 +23,8 @@
 from pathlib import Path
 from typing import Any
 
+from reflexio.server.env_utils import env_truthy
+
 INTERNAL_ENV = "OPENCLAW_SMART_INTERNAL"
 
 # Plugin layout (in-repo / editable):
@@ -59,7 +61,7 @@ def is_internal_invocation(payload: dict[str, Any]) -> bool:
             reflexio repository. False otherwise, including when ``cwd`` is
             missing or unresolvable.
     """
-    if os.environ.get(INTERNAL_ENV) == "1":
+    if env_truthy(os.environ.get(INTERNAL_ENV, "")):
         return True
     cwd = payload.get("cwd") or payload.get("workspaceDir")
     if not isinstance(cwd, str) or not cwd:
diff --git a/reflexio/server/__init__.py b/reflexio/server/__init__.py
index 0e2b8c04b..f7613b6fb 100644
--- a/reflexio/server/__init__.py
+++ b/reflexio/server/__init__.py
@@ -14,6 +14,7 @@
 
 from reflexio.cli.env_loader import load_reflexio_env
 from reflexio.cli.paths import reflexio_home
+from reflexio.server.env_utils import env_truthy
 
 # Load environment variables using shared discovery logic
 load_reflexio_env()
@@ -131,7 +132,7 @@ def format(self, record: logging.LogRecord) -> str:
 def _truthy_env(name: str) -> bool:
     """Return whether an environment variable is explicitly truthy."""
     raw = os.environ.get(name, "").strip().lower()
-    return raw in ("true", "yes", "1", "on")
+    return env_truthy(raw)
 
 
 def _is_production_environment() -> bool:
diff --git a/reflexio/server/api_endpoints/account_api.py b/reflexio/server/api_endpoints/account_api.py
index df240d792..09ea7c883 100644
--- a/reflexio/server/api_endpoints/account_api.py
+++ b/reflexio/server/api_endpoints/account_api.py
@@ -15,7 +15,6 @@
 from __future__ import annotations
 
 import logging
-import os
 from typing import Any, cast
 
 from reflexio.lib._storage_labels import describe_storage
@@ -24,6 +23,7 @@
     WhoamiResponse,
 )
 from reflexio.server.cache.reflexio_cache import get_reflexio
+from reflexio.server.env_utils import env_bool
 
 logger = logging.getLogger(__name__)
 
@@ -87,7 +87,7 @@ def my_config_allowed() -> bool:
     FastAPI endpoint wrapper in ``reflexio.server.api`` and does not
     flow through this helper.
     """
-    return os.environ.get(_ALLOW_MY_CONFIG_ENV_VAR, "").lower() in {"1", "true", "yes"}
+    return env_bool(_ALLOW_MY_CONFIG_ENV_VAR, default=False)
 
 
 def my_config(org_id: str) -> MyConfigResponse:
diff --git a/reflexio/server/env_utils.py b/reflexio/server/env_utils.py
index 099bf2a32..ab5d3dfc5 100644
--- a/reflexio/server/env_utils.py
+++ b/reflexio/server/env_utils.py
@@ -60,6 +60,77 @@ def env_required_literal(
     return value
 
 
+_TRUE = "true"
+_FALSE = "false"
+
+
+class EnvBoolError(ValueError):
+    """A boolean environment variable carried a value that is not true/false."""
+
+
+def env_bool(
+    name: str,
+    *,
+    default: bool,
+    env: Mapping[str, str] | None = None,
+) -> bool:
+    """Parse a boolean environment variable, refusing anything ambiguous.
+
+    Accepts ``true`` / ``false`` only, case-insensitively and after stripping.
+    Unset or blank resolves to ``default``, matching :func:`env_str`'s
+    "set to blank behaves like unset" rule. **Every other value raises.**
+
+    Raising is the point, and it is what this codebase previously lacked. The
+    predecessor (``env_truthy``) was ``value.lower() in {...}``, so anything
+    unrecognised returned ``False`` -- meaning ``REFLEXIO_REQUIRE_DATA_DB=ture``
+    silently DISABLED a guard rather than failing. A typo that switches a safety
+    knob off without a word is the same silent-green class as a CI lane that
+    never runs.
+
+    Erroring on unrecognised input is the universal convention: Go's
+    ``strconv.ParseBool`` states "Any other value returns an error", and Pydantic
+    raises ``bool_parsing``. Neither defaults.
+
+    The accepted SET is deliberately narrower than either, which both admit
+    ``1``/``0``. ``.env.template`` documents ~28 variables as
+    ``options: true | false``, so accepting more would leave that documentation a
+    half-truth -- and the direction of travel is narrowing, not widening: YAML
+    1.2 removed ``yes``/``no``/``on``/``off`` as booleans outright after ``NO``
+    (Norway) silently parsed as false.
+
+    Args:
+        name (str): Environment variable name, used in the error message.
+        default (bool): Value returned when the variable is unset or blank.
+        env (Mapping[str, str] | None): Mapping to read; defaults to os.environ.
+
+    Returns:
+        bool: The parsed value, or ``default`` when unset or blank.
+
+    Raises:
+        EnvBoolError: The variable is set to anything other than true/false.
+    """
+    raw = (env if env is not None else os.environ).get(name)
+    if raw is None or not raw.strip():
+        return default
+    value = raw.strip().lower()
+    if value == _TRUE:
+        return True
+    if value == _FALSE:
+        return False
+    raise EnvBoolError(
+        f"{name} must be {_TRUE!r} or {_FALSE!r} (got {raw.strip()!r}). "
+        f"Values such as '1', '0', 'yes' and 'on' are deliberately NOT accepted: "
+        f"they were parsed inconsistently across this codebase, and an "
+        f"unrecognised value used to read as false without any error."
+    )
+
+
 def env_truthy(value: str) -> bool:
-    """Return whether a string environment value is truthy."""
+    """Return whether a string environment value is truthy.
+
+    Deprecated in favour of :func:`env_bool`, which takes the variable NAME and
+    can therefore name it when refusing a bad value. Retained because it is a
+    public export of this package; it keeps the historical permissive set so an
+    external caller's behaviour does not change silently.
+    """
     return value.strip().lower() in {"1", "true", "yes", "on"}
diff --git a/reflexio/server/llm/providers/claude_code_provider.py b/reflexio/server/llm/providers/claude_code_provider.py
index dd6356d2f..56482b9d1 100644
--- a/reflexio/server/llm/providers/claude_code_provider.py
+++ b/reflexio/server/llm/providers/claude_code_provider.py
@@ -45,6 +45,7 @@
 )
 from pydantic import BaseModel
 
+from reflexio.server.env_utils import env_truthy
 from reflexio.server.llm.providers.claude_code_stream_parser import (
     ParseResult,
     classify_stall,
@@ -80,7 +81,6 @@ def _is_windows() -> bool:
 _WINDOWS_ARGV_SYSTEM_PROMPT_LIMIT = 3_000
 _WINDOWS_CLI_SUFFIXES = (".cmd", ".exe", ".bat")
 
-_TRUTHY_ENV_VALUES = {"1", "true", "yes"}
 _UNSUPPORTED_PARAMS_WARNED: set[str] = set()
 _IMAGE_WARNED = False
 _MULTITURN_WARNED = False
@@ -103,7 +103,7 @@ def _env_enabled() -> bool:
         bool: True if the opt-in env var is set, False otherwise.
     """
     raw = os.environ.get(ENV_ENABLE)
-    return bool(raw) and raw.lower() in _TRUTHY_ENV_VALUES
+    return bool(raw) and env_truthy(raw)
 
 
 def _host() -> str:
diff --git a/reflexio/server/llm/providers/embedding_service_provider.py b/reflexio/server/llm/providers/embedding_service_provider.py
index cfd7a112c..56e122d69 100644
--- a/reflexio/server/llm/providers/embedding_service_provider.py
+++ b/reflexio/server/llm/providers/embedding_service_provider.py
@@ -20,6 +20,7 @@
 
 import httpx
 
+from reflexio.server.env_utils import env_truthy
 from reflexio.server.tracing import profile_step
 
 _LOGGER = logging.getLogger(__name__)
@@ -333,7 +334,7 @@ def embedding_provider_mode(model: str | None = None) -> EmbeddingProviderMode:
     if model and not _uses_embedding_service(model):
         return "cloud"
 
-    if os.environ.get(_ENV_CLAUDE_SMART_LOCAL) == "1":
+    if env_truthy(os.environ.get(_ENV_CLAUDE_SMART_LOCAL, "")):
         return "local_service"
 
     if os.environ.get(_ENV_SERVICE_URL, "").strip():
diff --git a/reflexio/server/llm/providers/local_embedding_provider.py b/reflexio/server/llm/providers/local_embedding_provider.py
index 4860f764e..6058ae79e 100644
--- a/reflexio/server/llm/providers/local_embedding_provider.py
+++ b/reflexio/server/llm/providers/local_embedding_provider.py
@@ -28,6 +28,8 @@
 from pathlib import Path
 from typing import Any, BinaryIO, cast
 
+from reflexio.server.env_utils import env_truthy
+
 try:
     import fcntl
 except ImportError:  # pragma: no cover - Windows only
@@ -505,7 +507,7 @@ def is_local_embedder_available() -> bool:
         bool: True when ``CLAUDE_SMART_USE_LOCAL_EMBEDDING=1``
             AND the ONNX dependencies are importable.
     """
-    if os.environ.get(_ENV_ENABLE) != "1":
+    if not env_truthy(os.environ.get(_ENV_ENABLE, "")):
         return False
     return are_local_embedding_dependencies_available()
 
diff --git a/reflexio/server/llm/providers/nomic_embedding_provider.py b/reflexio/server/llm/providers/nomic_embedding_provider.py
index 4fcab25eb..9e4449bbb 100644
--- a/reflexio/server/llm/providers/nomic_embedding_provider.py
+++ b/reflexio/server/llm/providers/nomic_embedding_provider.py
@@ -34,6 +34,7 @@
 import threading
 from typing import Any
 
+from reflexio.server.env_utils import env_truthy
 from reflexio.server.llm.llm_utils import positive_int_env
 
 _LOGGER = logging.getLogger(__name__)
@@ -286,7 +287,7 @@ def register_if_enabled() -> bool:
     global _REGISTERED
     if _REGISTERED:
         return True
-    if os.environ.get(_ENV_ENABLE) != "1":
+    if not env_truthy(os.environ.get(_ENV_ENABLE, "")):
         return False
     provider = os.environ.get(_ENV_PROVIDER, "").strip().lower()
     if provider in {"local_service", "internal_service", "off"}:
@@ -296,7 +297,7 @@ def register_if_enabled() -> bool:
             provider,
         )
         return False
-    if not provider and os.environ.get(_ENV_DAEMON) != "1":
+    if not provider and not env_truthy(os.environ.get(_ENV_DAEMON, "")):
         _LOGGER.info(
             "Nomic in-process prewarm skipped; %s=1 now defaults to the "
             "shared embedding service.",
diff --git a/reflexio/server/llm/providers/openclaw_provider.py b/reflexio/server/llm/providers/openclaw_provider.py
index 7f74c3034..a525dd991 100644
--- a/reflexio/server/llm/providers/openclaw_provider.py
+++ b/reflexio/server/llm/providers/openclaw_provider.py
@@ -25,6 +25,8 @@
 from litellm.llms.custom_llm import CustomLLM
 from litellm.types.utils import Choices, Message, ModelResponse, Usage
 
+from reflexio.server.env_utils import env_truthy
+
 _LOGGER = logging.getLogger(__name__)
 
 PROVIDER_KEY = "openclaw"
@@ -34,7 +36,6 @@
 ENV_TIMEOUT = "OPENCLAW_CLI_TIMEOUT"
 _DEFAULT_TIMEOUT_SECONDS = 180
 
-_TRUTHY = {"1", "true", "yes"}
 
 # Module-level state reset by tests via the _reset_module_state fixture.
 _REGISTERED: bool = False
@@ -51,7 +52,7 @@ def _env_enabled() -> bool:
     Returns:
         bool: True if the opt-in env var is set to a truthy value, else False.
     """
-    return os.environ.get(ENV_ENABLE, "").lower() in _TRUTHY
+    return env_truthy(os.environ.get(ENV_ENABLE, ""))
 
 
 def _resolve_cli_path() -> str | None:
diff --git a/reflexio/server/services/base_generation/_should_run.py b/reflexio/server/services/base_generation/_should_run.py
index 995542474..b55a58aca 100644
--- a/reflexio/server/services/base_generation/_should_run.py
+++ b/reflexio/server/services/base_generation/_should_run.py
@@ -18,12 +18,12 @@
 """
 
 import logging
-import os
 import time
 from typing import TYPE_CHECKING, Any, Generic, TypeVar
 
 from reflexio.models.api_schema.internal_schema import RequestInteractionDataModel
 from reflexio.server.api_endpoints.request_context import RequestContext
+from reflexio.server.env_utils import env_bool
 from reflexio.server.llm.litellm_client import LiteLLMClient
 from reflexio.server.services.extractor_config_utils import get_extractor_name
 from reflexio.server.services.extractor_interaction_utils import (
@@ -101,7 +101,7 @@ def _should_run_before_extraction(self, extractor_config: TExtractorConfig) -> b
             return True
 
         # Skip for mock mode
-        if os.getenv("MOCK_LLM_RESPONSE", "").lower() == "true":
+        if env_bool("MOCK_LLM_RESPONSE", default=False):
             return True
 
         # `force_extraction=True` is the caller's explicit "no gates" signal —
diff --git a/reflexio/server/services/braintrust/_cron.py b/reflexio/server/services/braintrust/_cron.py
index 3dc158860..0a20f8daf 100644
--- a/reflexio/server/services/braintrust/_cron.py
+++ b/reflexio/server/services/braintrust/_cron.py
@@ -27,6 +27,7 @@
 from collections.abc import Callable
 from dataclasses import dataclass, field
 
+from reflexio.server.env_utils import env_bool
 from reflexio.server.services.braintrust.client import (
     DEFAULT_BASE_URL,
     BraintrustClient,
@@ -41,7 +42,7 @@
 
 def _interval_seconds() -> int:
     """Pick the recurring interval based on `IS_TEST_ENV`."""
-    if os.environ.get("IS_TEST_ENV", "").strip().lower() == "true":
+    if env_bool("IS_TEST_ENV", default=False):
         return _TEST_INTERVAL_SECONDS
     return _DEFAULT_INTERVAL_SECONDS
 
diff --git a/reflexio/server/services/playbook/components/aggregator.py b/reflexio/server/services/playbook/components/aggregator.py
index 0b53055cd..31239ddd2 100644
--- a/reflexio/server/services/playbook/components/aggregator.py
+++ b/reflexio/server/services/playbook/components/aggregator.py
@@ -2,7 +2,6 @@
 
 import logging
 import math
-import os
 import time
 import uuid
 from collections.abc import Callable, Sequence
@@ -25,6 +24,7 @@
     PlaybookAggregatorConfig,
 )
 from reflexio.server.api_endpoints.request_context import RequestContext
+from reflexio.server.env_utils import env_bool
 from reflexio.server.error_reporting import capture_anomaly, error_tags
 from reflexio.server.llm._litellm_types import ModelProvenance
 from reflexio.server.llm.litellm_client import LiteLLMClient
@@ -1771,9 +1771,8 @@ def run(self, playbook_aggregator_request: PlaybookAggregatorRequest) -> dict:
                         # other caller still aborts on a missing embedding: a
                         # centroid-less cluster row would silently break the
                         # incremental re-aggregation this table exists to feed.
-                        if (
-                            not saved_fb.embedding
-                            and os.getenv("MOCK_LLM_RESPONSE", "").lower() != "true"
+                        if not saved_fb.embedding and not env_bool(
+                            "MOCK_LLM_RESPONSE", default=False
                         ):
                             raise RuntimeError(
                                 "rerun agent playbook has no centroid embedding"
@@ -2051,7 +2050,7 @@ def get_clusters(
             model_name=self.storage.embedding_model_name,
         )
         # Mock mode: cluster by trigger
-        if os.getenv("MOCK_LLM_RESPONSE", "").lower() == "true":
+        if env_bool("MOCK_LLM_RESPONSE", default=False):
             logger.info("Mock mode: clustering by trigger")
             return aggregator_clustering.cluster_by_trigger_mock(
                 user_playbooks, min_cluster_size
@@ -2327,7 +2326,7 @@ def _generate_playbook_from_cluster_outcome(
         if not cluster_playbooks:
             return AggregationGenerationOutcome("retryable_failure", [])
 
-        if os.getenv("MOCK_LLM_RESPONSE", "").lower() == "true":
+        if env_bool("MOCK_LLM_RESPONSE", default=False):
             # Extract structured fields directly from cluster
             triggers = [fb.trigger for fb in cluster_playbooks if fb.trigger]
 
diff --git a/reflexio/server/services/playbook/components/consolidator.py b/reflexio/server/services/playbook/components/consolidator.py
index acde259f2..3a3a7f3ce 100644
--- a/reflexio/server/services/playbook/components/consolidator.py
+++ b/reflexio/server/services/playbook/components/consolidator.py
@@ -4,7 +4,6 @@
 """
 
 import logging
-import os
 from datetime import UTC, datetime
 from typing import Annotated, Literal
 
@@ -21,6 +20,7 @@
     normalize_provider_value,
 )
 from reflexio.server.api_endpoints.request_context import RequestContext
+from reflexio.server.env_utils import env_bool
 from reflexio.server.error_reporting import error_tags
 from reflexio.server.llm._litellm_types import ModelProvenance
 from reflexio.server.llm.litellm_client import (
@@ -966,7 +966,7 @@ def deduplicate(
             raise TypeError("agent_version is required")
 
         # Check if mock mode is enabled
-        if os.getenv("MOCK_LLM_RESPONSE", "").lower() == "true":
+        if env_bool("MOCK_LLM_RESPONSE", default=False):
             logger.info("Mock mode: skipping consolidation")
             all_playbooks: list[UserPlaybook] = []
             for result in results:
diff --git a/reflexio/server/services/playbook/components/extractor.py b/reflexio/server/services/playbook/components/extractor.py
index 6d29539ad..b72d64ce4 100644
--- a/reflexio/server/services/playbook/components/extractor.py
+++ b/reflexio/server/services/playbook/components/extractor.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import logging
-import os
 from collections import Counter
 from typing import TYPE_CHECKING
 
@@ -9,6 +8,7 @@
 from reflexio.models.api_schema.service_schemas import UserPlaybook
 from reflexio.models.config_schema import PlaybookConfig
 from reflexio.server.api_endpoints.request_context import RequestContext
+from reflexio.server.env_utils import env_bool
 from reflexio.server.llm.litellm_client import LiteLLMClient
 from reflexio.server.llm.model_defaults import ModelRole, resolve_model_name
 from reflexio.server.llm.token_accounting import RunTokenTotals, sum_trace_tokens
@@ -273,7 +273,7 @@ def extract_playbook_entries(
         )
 
         # Check if mock mode is enabled
-        if os.getenv("MOCK_LLM_RESPONSE", "").lower() == "true":
+        if env_bool("MOCK_LLM_RESPONSE", default=False):
             logger.info("Mock mode: generating mock playbook entry")
             mock_response = self._generate_mock_playbook_list(
                 request_interaction_data_models, prompt_context.evidence_sources
diff --git a/reflexio/server/services/playbook/service.py b/reflexio/server/services/playbook/service.py
index 361a592ea..1c0540ba1 100644
--- a/reflexio/server/services/playbook/service.py
+++ b/reflexio/server/services/playbook/service.py
@@ -1,11 +1,12 @@
 from __future__ import annotations
 
 import logging
-import os
 import uuid
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
+from reflexio.server.env_utils import env_bool
+
 if TYPE_CHECKING:
     from reflexio.server.api_endpoints.request_context import RequestContext
     from reflexio.server.llm.litellm_client import LiteLLMClient
@@ -449,7 +450,7 @@ def _resolve_write_plan(
             and reviewer.is_enabled()
             and playbook_config is not None
             and self.service_config is not None
-            and os.getenv("MOCK_LLM_RESPONSE", "").lower() != "true"
+            and not env_bool("MOCK_LLM_RESPONSE", default=False)
         ):
             review_interactions = self._review_interaction_window(all_playbooks)
             if not review_interactions:
diff --git a/reflexio/server/services/profile/components/consolidator.py b/reflexio/server/services/profile/components/consolidator.py
index 864d7d324..2f51ef7f8 100644
--- a/reflexio/server/services/profile/components/consolidator.py
+++ b/reflexio/server/services/profile/components/consolidator.py
@@ -4,7 +4,6 @@
 """
 
 import logging
-import os
 from collections import Counter
 from datetime import UTC, datetime
 
@@ -15,6 +14,7 @@
 from reflexio.models.profile_id import new_profile_id
 from reflexio.models.structured_output import StrictStructuredOutput
 from reflexio.server.api_endpoints.request_context import RequestContext
+from reflexio.server.env_utils import env_bool
 from reflexio.server.error_reporting import capture_anomaly
 from reflexio.server.llm._litellm_types import ModelProvenance
 from reflexio.server.llm.litellm_client import (
@@ -521,7 +521,7 @@ def deduplicate(
         self.consolidated_output_indices = set()
 
         # Check if mock mode is enabled
-        if os.getenv("MOCK_LLM_RESPONSE", "").lower() == "true":
+        if env_bool("MOCK_LLM_RESPONSE", default=False):
             logger.info("Mock mode: skipping deduplication")
             return new_profiles, [], []
 
diff --git a/reflexio/server/services/profile/components/extractor.py b/reflexio/server/services/profile/components/extractor.py
index cb59b9962..18b1861f7 100644
--- a/reflexio/server/services/profile/components/extractor.py
+++ b/reflexio/server/services/profile/components/extractor.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import logging
-import os
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING
 
@@ -12,6 +11,7 @@
 from reflexio.models.config_schema import ProfileExtractorConfig
 from reflexio.models.profile_id import new_profile_id
 from reflexio.server.api_endpoints.request_context import RequestContext
+from reflexio.server.env_utils import env_bool
 from reflexio.server.llm.litellm_client import LiteLLMClient
 from reflexio.server.llm.token_accounting import RunTokenTotals, sum_trace_tokens
 from reflexio.server.services.deferred_learning_plan import ExtractorBookmarkAdvance
@@ -341,8 +341,7 @@ def _generate_raw_updates_from_sessions(
             list[dict]: List of profile dicts with content, time_to_live, and optional metadata
         """
         # Check if mock mode is enabled
-        mock_env_for_raw = os.getenv("MOCK_LLM_RESPONSE", "")
-        if mock_env_for_raw.lower() == "true":
+        if env_bool("MOCK_LLM_RESPONSE", default=False):
             return self._generate_mock_profiles(
                 request_interaction_data_models=request_interaction_data_models,
             )
diff --git a/reflexio/server/services/tagging/service.py b/reflexio/server/services/tagging/service.py
index 7dd3f91ca..be06b908d 100644
--- a/reflexio/server/services/tagging/service.py
+++ b/reflexio/server/services/tagging/service.py
@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 import logging
-import os
 
 from pydantic import ConfigDict, Field
 
@@ -13,6 +12,7 @@
 )
 from reflexio.models.structured_output import StrictStructuredOutput
 from reflexio.server.api_endpoints.request_context import RequestContext
+from reflexio.server.env_utils import env_bool
 from reflexio.server.llm.litellm_client import LiteLLMClient
 from reflexio.server.llm.model_defaults import ModelRole, resolve_model_name
 from reflexio.server.services.service_utils import log_llm_messages, log_model_response
@@ -198,7 +198,7 @@ def _tag_agent_playbooks(
     def _generate_tags(
         self, *, tagging_definition_prompt: str, content: str
     ) -> list[str]:
-        if os.getenv("MOCK_LLM_RESPONSE", "").lower() == "true":
+        if env_bool("MOCK_LLM_RESPONSE", default=False):
             return ["example_tag"]
 
         prompt = self.request_context.prompt_manager.render_prompt(
diff --git a/reflexio/server/services/tagging/tagging_scheduler.py b/reflexio/server/services/tagging/tagging_scheduler.py
index 2c5c2c5c9..7b0258ef5 100644
--- a/reflexio/server/services/tagging/tagging_scheduler.py
+++ b/reflexio/server/services/tagging/tagging_scheduler.py
@@ -18,7 +18,6 @@
 import heapq
 import itertools
 import logging
-import os
 import threading
 import time
 from collections.abc import Callable
@@ -26,6 +25,7 @@
 
 from reflexio.server.api_endpoints.request_context import RequestContext
 from reflexio.server.callback_executor import drain_callbacks, submit_callback
+from reflexio.server.env_utils import env_bool
 from reflexio.server.error_reporting import capture_anomaly
 from reflexio.server.llm.litellm_client import LiteLLMClient
 from reflexio.server.services.tagging.service import TaggingService
@@ -42,7 +42,7 @@
 # to debounce a burst of publishes into one pass, short enough that tags appear
 # promptly. Kept as a patch point for tests.
 TAGGING_DELAY_SECONDS = 15
-IS_TEST_ENV = os.environ.get("IS_TEST_ENV", "false").strip().lower() == "true"
+IS_TEST_ENV = env_bool("IS_TEST_ENV", default=False)
 _EFFECTIVE_DELAY_SECONDS = 1 if IS_TEST_ENV else TAGGING_DELAY_SECONDS
 
 # (org_id, project_id, user_id, agent_version)
diff --git a/reflexio/test_support/llm_mock.py b/reflexio/test_support/llm_mock.py
index 381e63d56..60c16a396 100644
--- a/reflexio/test_support/llm_mock.py
+++ b/reflexio/test_support/llm_mock.py
@@ -41,6 +41,7 @@ def test_something_e2e():
 from unittest.mock import MagicMock, NonCallableMock, patch
 
 from reflexio.models.structured_output import find_schema_keyword as _find_schema_key
+from reflexio.server.env_utils import env_bool
 from reflexio.test_support.llm_model_registry import get_model_registry
 
 _LEARNING_REF_PATTERN = re.compile(r'"learning_ref":\s*"([^"]+)"')
@@ -324,7 +325,7 @@ def assert_litellm_unpatched() -> None:
             "tests/e2e_tests/ and must carry the requires_credentials marker; "
             "the e2e conftest lifts the session patch for those."
         )
-    if os.getenv("MOCK_LLM_RESPONSE", "").lower() == "true":
+    if env_bool("MOCK_LLM_RESPONSE", default=False):
         raise AssertionError(
             "MOCK_LLM_RESPONSE=true, so service code takes its canned branch "
             "without calling litellm at all -- this live-provider test would "
diff --git a/tests/server/api_endpoints/test_account_api.py b/tests/server/api_endpoints/test_account_api.py
index 1a96082e3..bc2d748df 100644
--- a/tests/server/api_endpoints/test_account_api.py
+++ b/tests/server/api_endpoints/test_account_api.py
@@ -132,3 +132,38 @@ def test_exception_returns_generic_message_no_leakage(self):
         assert body["success"] is False
         assert body["message"] == "Failed to load storage configuration"
         assert "boom-secret-db-url" not in response.text
+
+
+def test_my_config_gate_accepts_only_true(monkeypatch) -> None:
+    """The credential-export gate must not be opened by an ambiguous spelling.
+
+    ``my_config`` is the "download my creds" endpoint: on OS/self-host its only
+    guard is this variable, and the response carries the caller's storage
+    configuration. So the gate is parsed with ``env_bool`` -- narrower than the
+    ``{"1", "true", "yes"}`` allowlist it replaced, and far narrower than the
+    permissive ``env_truthy`` (which accepts ``"on"``) briefly used here.
+
+    A wrong value raises rather than silently opening or closing the gate: for a
+    credential path, refusing to guess is the only safe behaviour.
+    """
+    from reflexio.server.api_endpoints.account_api import (
+        _ALLOW_MY_CONFIG_ENV_VAR,
+        my_config_allowed,
+    )
+    from reflexio.server.env_utils import EnvBoolError
+
+    monkeypatch.setenv(_ALLOW_MY_CONFIG_ENV_VAR, "true")
+    assert my_config_allowed() is True
+
+    monkeypatch.setenv(_ALLOW_MY_CONFIG_ENV_VAR, "false")
+    assert my_config_allowed() is False
+
+    monkeypatch.delenv(_ALLOW_MY_CONFIG_ENV_VAR, raising=False)
+    assert my_config_allowed() is False, "unset must not open the gate"
+
+    # "on" is the specific spelling that a permissive parser would have let
+    # through; "1" and "yes" were accepted by the older allowlist.
+    for ambiguous in ("on", "1", "yes", "y"):
+        monkeypatch.setenv(_ALLOW_MY_CONFIG_ENV_VAR, ambiguous)
+        with pytest.raises(EnvBoolError):
+            my_config_allowed()
diff --git a/tests/server/test_env_utils.py b/tests/server/test_env_utils.py
index 7e4d59d4c..d2fb95788 100644
--- a/tests/server/test_env_utils.py
+++ b/tests/server/test_env_utils.py
@@ -3,6 +3,8 @@
 import pytest
 
 from reflexio.server.env_utils import (
+    EnvBoolError,
+    env_bool,
     env_get,
     env_required,
     env_required_literal,
@@ -67,3 +69,60 @@ def test_env_truthy_true_values(value: str) -> None:
 @pytest.mark.parametrize("value", ["", "0", "false", "off", "no"])
 def test_env_truthy_false_values(value: str) -> None:
     assert env_truthy(value) is False
+
+
+def test_env_bool_accepts_true_and_false_case_insensitively() -> None:
+    for raw in ("true", "TRUE", "True", "  true  "):
+        assert env_bool("KEY", default=False, env={"KEY": raw}) is True
+    for raw in ("false", "FALSE", "False", "  false  "):
+        assert env_bool("KEY", default=True, env={"KEY": raw}) is False
+
+
+def test_env_bool_treats_unset_and_blank_as_the_default() -> None:
+    """Same invariant env_str carries: a `KEY=` line reads as unset."""
+    assert env_bool("KEY", default=True, env={}) is True
+    assert env_bool("KEY", default=False, env={}) is False
+    assert env_bool("KEY", default=True, env={"KEY": ""}) is True
+    assert env_bool("KEY", default=True, env={"KEY": "   "}) is True
+
+
+@pytest.mark.parametrize("raw", ["1", "0", "yes", "no", "on", "off", "y", "n"])
+def test_env_bool_refuses_the_values_that_were_parsed_inconsistently(raw: str) -> None:
+    """These are the spellings that meant different things in different modules.
+
+    `on` was truthy under env_truthy and falsy under the `{1,true,yes}` set; `y`
+    worked in exactly one module. Accepting them here would preserve the
+    ambiguity this helper exists to remove.
+    """
+    with pytest.raises(EnvBoolError):
+        env_bool("KEY", default=False, env={"KEY": raw})
+
+
+def test_env_bool_refuses_a_typo_rather_than_silently_returning_false() -> None:
+    """The defect that motivated this helper.
+
+    `env_truthy("ture")` is False -- indistinguishable from a deliberate
+    `false`, so a mistyped safety knob switches itself off in silence. A wrong
+    value must be loud.
+    """
+    with pytest.raises(EnvBoolError) as excinfo:
+        env_bool(
+            "REFLEXIO_REQUIRE_DATA_DB",
+            default=False,
+            env={"REFLEXIO_REQUIRE_DATA_DB": "ture"},
+        )
+
+    message = str(excinfo.value)
+    assert "REFLEXIO_REQUIRE_DATA_DB" in message, "the error must name the variable"
+    assert "ture" in message, "the error must quote the offending value"
+
+
+def test_env_truthy_keeps_its_permissive_set_for_external_callers() -> None:
+    """Deliberate: it is a public export, so narrowing it would break callers.
+
+    Undocumented dev/CI knobs still route through it -- `REFLEXIO_REQUIRE_DOCKER=1`
+    is set in ci-fast.yml and must keep working.
+    """
+    assert env_truthy("1") is True
+    assert env_truthy("on") is True
+    assert env_truthy("ture") is False