diff --git a/api/db/services/knowledgebase_service.py b/api/db/services/knowledgebase_service.py
index a164287fa4..d6bb9e1db1 100644
--- a/api/db/services/knowledgebase_service.py
+++ b/api/db/services/knowledgebase_service.py
@@ -48,6 +48,25 @@ class KnowledgebaseService(CommonService):
     """
     model = Knowledgebase
 
+    @classmethod
+    def _visibility_and_status_filter(cls, joined_tenant_ids, user_id):
+        """
+        Build a Peewee filter expression representing knowledgebase visibility
+        for a given user, combined with a valid-status constraint.
+
+        Visibility rules:
+        - Team KBs (`permission == TenantPermission.TEAM`) owned by any tenant in `joined_tenant_ids`
+        - KBs owned by the current user (`tenant_id == user_id`)
+        Always constrained to `StatusEnum.VALID`.
+        """
+        return (
+            (
+                (cls.model.tenant_id.in_(joined_tenant_ids) & (cls.model.permission == TenantPermission.TEAM.value))
+                | (cls.model.tenant_id == user_id)
+            )
+            & (cls.model.status == StatusEnum.VALID.value)
+        )
+
     @classmethod
     @DB.connection_context()
     def accessible4deletion(cls, kb_id, user_id):
@@ -169,18 +188,12 @@ class KnowledgebaseService(CommonService):
         ]
         if keywords:
             kbs = cls.model.select(*fields).join(User, on=(cls.model.tenant_id == User.id)).where(
-                ((cls.model.tenant_id.in_(joined_tenant_ids) & (cls.model.permission ==
-                                                                TenantPermission.TEAM.value)) | (
-                    cls.model.tenant_id == user_id))
-                & (cls.model.status == StatusEnum.VALID.value),
-                (fn.LOWER(cls.model.name).contains(keywords.lower()))
+                cls._visibility_and_status_filter(joined_tenant_ids, user_id),
+                fn.LOWER(cls.model.name).contains(keywords.lower()),
             )
         else:
             kbs = cls.model.select(*fields).join(User, on=(cls.model.tenant_id == User.id)).where(
-                ((cls.model.tenant_id.in_(joined_tenant_ids) & (cls.model.permission ==
-                                                                TenantPermission.TEAM.value)) | (
-                    cls.model.tenant_id == user_id))
-                & (cls.model.status == StatusEnum.VALID.value)
+                cls._visibility_and_status_filter(joined_tenant_ids, user_id),
             )
         if parser_id:
             kbs = kbs.where(cls.model.parser_id == parser_id)
@@ -213,11 +226,7 @@ class KnowledgebaseService(CommonService):
             cls.model.update_date
         ]
         # find team kb and owned kb
-        kbs = cls.model.select(*fields).where(
-            (cls.model.tenant_id.in_(tenant_ids) & (cls.model.permission ==TenantPermission.TEAM.value)) | (
-                cls.model.tenant_id == user_id
-            )
-        )
+        kbs = cls.model.select(*fields).where(cls._visibility_and_status_filter(tenant_ids, user_id))
         # sort by create_time asc
         kbs.order_by(cls.model.create_time.asc())
         # maybe cause slow query by deep paginate, optimize later.
@@ -459,12 +468,7 @@ class KnowledgebaseService(CommonService):
         if parser_id:
             kbs = kbs.where(cls.model.parser_id == parser_id)
 
-        kbs = kbs.where(
-            ((cls.model.tenant_id.in_(joined_tenant_ids) & (cls.model.permission ==
-                                                            TenantPermission.TEAM.value)) | (
-                cls.model.tenant_id == user_id))
-            & (cls.model.status == StatusEnum.VALID.value)
-        )
+        kbs = kbs.where(cls._visibility_and_status_filter(joined_tenant_ids, user_id))
 
         if desc:
             kbs = kbs.order_by(cls.model.getter_by(orderby).desc())