fix(tenant): guard against sanitization collisions with production tenant

The request-level tenant guard compared the raw user string exactly
(user == PRODUCTION_TENANT), but all local namespaces (Qdrant
collections, Redis keys) are derived through sanitize_user_id(), which
lowercases and strips/maps punctuation. Case or punctuation variants
("JPMSchweitzer", "jpmschweitzer.", " jpmschweitzer") therefore passed
the guard yet resolved to the production namespaces, letting a dev
instance on the shared services read/write production tenant data.

- context.py: compare sanitize_user_id(user) against the sanitized
  production tenant; expose the guard as public apply_tenant_guard()
- config.py: startup refusal validator uses the same sanitized
  comparison, so a colliding DEFAULT_USER refuses startup loudly
  instead of relying on the allowlist fallback
- tests: variant matrix at both config and request-context level,
  plus a non-colliding passthrough case

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/core/config.py

@@ -224,10 +224,19 @@ class Config(BaseSettings):
         This is the hard stop of the tenant isolation guard: a dev/test
         instance must never be able to read or write the production
         tenant's data in the shared services.
+
+        The comparison is on the sanitized form: namespaces are derived
+        through sanitize_user_id(), so variants like "JPMSchweitzer" or
+        "jpmschweitzer." collide with the production namespaces and are
+        refused just as loudly.
         """
+        from src.core.multi_tenancy import sanitize_user_id
+
         if (
             self.ENVIRONMENT != Environment.PRODUCTION
-            and self.DEFAULT_USER == PRODUCTION_TENANT
+            and self.DEFAULT_USER is not None
+            and sanitize_user_id(self.DEFAULT_USER)
+            == sanitize_user_id(PRODUCTION_TENANT)
         ):
             raise ValueError(
                 f"Refusing to start: ENVIRONMENT={self.ENVIRONMENT.value} is "

src/core/context.py

@@ -41,20 +41,28 @@ current_conversation: ContextVar[str | None] = ContextVar(
 )
 
 
-def _apply_tenant_guard(user: str) -> str:
+def apply_tenant_guard(user: str) -> str:
     """
     Enforce tenant isolation at request-context resolution.
 
     In non-production environments the production tenant must never be
     the effective user - a request that explicitly asks for it is forced
     to the reserved test tenant instead (with a loud log line).
+
+    Comparison happens on the *sanitized* form of the user: every local
+    namespace (Qdrant collection, Redis key) is derived through
+    sanitize_user_id(), so any raw variant that collides with the
+    production tenant after sanitization ("JPMSchweitzer",
+    "jpmschweitzer.", " jpmschweitzer", ...) would otherwise resolve to
+    the production namespaces. Those variants are forced too.
     """
     # Import here to avoid circular dependency
     from src.core.config import PRODUCTION_TENANT, TEST_TENANT, Environment, config
+    from src.core.multi_tenancy import sanitize_user_id
 
     if (
         config.ENVIRONMENT != Environment.PRODUCTION
-        and user == PRODUCTION_TENANT
+        and sanitize_user_id(user) == sanitize_user_id(PRODUCTION_TENANT)
     ):
         from src.core.logging_config import get_logger
 
@@ -84,7 +92,7 @@ def get_user() -> str:
     user = current_user.get()
     if user == _USER_NOT_SET:
         return get_default_user()
-    return _apply_tenant_guard(user)
+    return apply_tenant_guard(user)
 
 
 def get_conversation_id() -> str | None:

tests/core/test_tenant_guard.py

@@ -63,6 +63,26 @@ class TestEffectiveDefaultUserMatrix:
         assert "Refusing to start" in str(exc_info.value)
         assert PRODUCTION_TENANT in str(exc_info.value)
 
+    @pytest.mark.parametrize(
+        "variant",
+        [
+            "JPMSchweitzer",
+            "JPMSCHWEITZER",
+            "jpmschweitzer.",
+            " jpmschweitzer",
+            "jpmschweitzer ",
+            "_jpmschweitzer_",
+            "jpmschweitzer!",
+        ],
+    )
+    def test_dev_with_production_tenant_variant_refuses_startup(self, variant):
+        """Sanitization collisions with the production tenant are refused too."""
+        with pytest.raises(ValidationError, match="Refusing to start"):
+            make_config(
+                ENVIRONMENT=Environment.DEVELOPMENT,
+                DEFAULT_USER=variant,
+            )
+
     # --- testing ---
 
     def test_testing_without_default_user_forces_test_tenant(self):
@@ -125,6 +145,48 @@ class TestRequestContextGuard:
         with RequestContext(user=PRODUCTION_TENANT):
             assert get_user() == TEST_TENANT
 
+    @pytest.mark.parametrize(
+        "variant",
+        [
+            "JPMSchweitzer",
+            "JPMSCHWEITZER",
+            "jpmschweitzer.",
+            " jpmschweitzer",
+            "jpmschweitzer ",
+            "_jpmschweitzer_",
+            "jpmschweitzer!",
+        ],
+    )
+    @pytest.mark.parametrize(
+        "environment", [Environment.DEVELOPMENT, Environment.TESTING]
+    )
+    def test_production_tenant_sanitization_variants_are_forced(
+        self, monkeypatch, environment, variant
+    ):
+        """
+        Any raw user that sanitizes to the production tenant would resolve
+        to the production namespaces (memories_jpmschweitzer,
+        session:jpmschweitzer:*) - the guard must force it to the test
+        tenant in non-production environments.
+        """
+        from src.core.multi_tenancy import get_memory_collection_name
+
+        self._patch_environment(monkeypatch, environment)
+        with RequestContext(user=variant):
+            effective = get_user()
+            assert effective == TEST_TENANT
+            assert (
+                get_memory_collection_name(effective)
+                != get_memory_collection_name(PRODUCTION_TENANT)
+            )
+
+    def test_dev_non_colliding_user_is_not_forced(self, monkeypatch):
+        """A user that sanitizes to a different namespace passes through."""
+        self._patch_environment(monkeypatch, Environment.DEVELOPMENT)
+        with RequestContext(user="jpm.schweitzer"):
+            # sanitizes to jpm_schweitzer != jpmschweitzer
+            assert get_user() == "jpm.schweitzer"
+
     def test_dev_explicit_other_user_passes_through(self, monkeypatch):
         self._patch_environment(monkeypatch, Environment.DEVELOPMENT)
         with RequestContext(user="testuser"):