fix(cloud): scope public login capability discovery

src/langbot/pkg/api/http/service/user.py

@@ -238,12 +238,20 @@ class UserService:
         return account is not None
 
     async def get_login_capabilities(self) -> dict[str, bool]:
-        """Derive enabled public login methods from all active Accounts."""
+        """Derive enabled public login methods in an explicit discovery scope."""
         password_count = sqlalchemy.func.count().filter(user.User.password.is_not(None), user.User.password != '')
         space_count = sqlalchemy.func.count().filter(user.User.space_account_uuid.is_not(None))
-        result = await self.ap.persistence_mgr.execute_async(
-            sqlalchemy.select(password_count, space_count).where(user.User.status == user.AccountStatus.ACTIVE.value)
+        statement = sqlalchemy.select(password_count, space_count).where(
+            user.User.status == user.AccountStatus.ACTIVE.value
         )
+        digest = hashlib.sha256(f'login-capabilities:{self._jwt_identity()[1]}'.encode('utf-8')).hexdigest()
+        current_session = getattr(self.ap.persistence_mgr, 'current_session', lambda: None)
+        identity_uow = getattr(self.ap.persistence_mgr, 'identity_discovery_uow', None)
+        if current_session() is None and callable(identity_uow):
+            async with identity_uow(digest) as discovery:
+                result = await discovery.session.execute(statement)
+        else:
+            result = await self.ap.persistence_mgr.execute_async(statement)
         password_accounts, space_accounts = result.one()
         return {
             'password_login_enabled': bool(password_accounts),

tests/unit_tests/api/service/test_user_service.py

@@ -210,6 +210,41 @@ class TestUserServiceIsInitialized:
         assert result is False
 
 
+class TestUserServiceGetLoginCapabilities:
+    """Tests for public login capability discovery."""
+
+    async def test_uses_explicit_identity_discovery_scope(self):
+        discovery_result = Mock()
+        discovery_result.one = Mock(return_value=(1, 2))
+        discovery_session = SimpleNamespace(execute=AsyncMock(return_value=discovery_result))
+
+        class DiscoveryContext:
+            async def __aenter__(self):
+                return SimpleNamespace(session=discovery_session)
+
+            async def __aexit__(self, exc_type, exc, tb):
+                return False
+
+        ap = SimpleNamespace()
+        ap.persistence_mgr = SimpleNamespace(
+            current_session=Mock(return_value=None),
+            identity_discovery_uow=Mock(return_value=DiscoveryContext()),
+            execute_async=AsyncMock(side_effect=AssertionError('unscoped persistence access')),
+        )
+        ap.workspace_service = SimpleNamespace(instance_uuid='instance-a')
+        service = UserService(ap)
+
+        result = await service.get_login_capabilities()
+
+        assert result == {
+            'password_login_enabled': True,
+            'space_login_enabled': True,
+        }
+        ap.persistence_mgr.identity_discovery_uow.assert_called_once()
+        discovery_session.execute.assert_awaited_once()
+        ap.persistence_mgr.execute_async.assert_not_awaited()
+
+
 class TestUserServiceGetUserByEmail:
     """Tests for get_user_by_email method."""