Fix private skill space visibility

frontend/server/skills/consts.py

@@ -42,6 +42,8 @@ def managed_description(self) -> str:
 SYSTEM_SKILL_SPACES = (SHARE_SPACE, REVIEW_SPACE)
 RESERVED_SKILL_SPACE_NAMES = frozenset(space.name for space in SYSTEM_SKILL_SPACES)
 SKILL_SPACE_DISPLAY_NAME_TAG = "display_name"
+SKILL_VISIBILITY_TAG = "veadk:visibility"
+SKILL_VISIBILITY_SHARED = "shared"
 REVIEW_SOURCE_SPACE_TAG = "studio:review-source-space"
 REVIEW_SOURCE_SKILL_TAG = "studio:review-source-skill"
 REVIEW_SOURCE_VERSION_TAG = "studio:review-source-version"

frontend/server/skills/repository.py

@@ -30,7 +30,12 @@
 from uuid import uuid4
 
 from .archive import SkillArchive
-from .consts import SHARED_SOURCE_VERSION_TAG, SKILL_SPACE_DISPLAY_NAME_TAG
+from .consts import (
+    SHARED_SOURCE_VERSION_TAG,
+    SKILL_SPACE_DISPLAY_NAME_TAG,
+    SKILL_VISIBILITY_SHARED,
+    SKILL_VISIBILITY_TAG,
+)
 from .space_names import skill_space_display_name
 
 if TYPE_CHECKING:
@@ -46,6 +51,50 @@ class SkillSpaceListResult:
     degraded: bool = False
 
 
+def _tags(value: Any) -> dict[str, str]:
+    return {
+        str(getattr(tag, "key", "") or ""): str(getattr(tag, "value", "") or "")
+        for tag in (getattr(value, "tags", None) or [])
+    }
+
+
+def skill_space_visible_to_author(
+    space: Any, *, author: str, is_admin: bool = False
+) -> bool:
+    from .system_spaces import is_shared_space
+
+    if is_admin:
+        return True
+    return is_shared_space(space) or _tags(space).get("author") == author
+
+
+def require_space_read(
+    client: Any,
+    skills_types: Any,
+    *,
+    space_id: str,
+    author: str,
+    is_admin: bool = False,
+) -> None:
+    from .system_spaces import is_review_space, is_shared_space
+
+    if is_admin:
+        return
+    space = client.get_skill_space(skills_types.GetSkillSpaceRequest(Id=space_id))
+    if is_shared_space(space):
+        return
+    if is_review_space(space):
+        raise SkillRepositoryError(
+            "SKILL_REVIEW_FORBIDDEN", "仅管理员可以查看审核申请", status_code=403
+        )
+    if _tags(space).get("author") != author:
+        raise SkillRepositoryError(
+            "SKILL_SPACE_READ_FORBIDDEN",
+            "只能查看自己创建的 Skill 空间",
+            status_code=403,
+        )
+
+
 def _is_missing_skill_relation(error: BaseException) -> bool:
     expected = "ResourceNotFound.skill"
     for name in ("code", "error_code", "Code"):
@@ -65,18 +114,24 @@ def list_skill_space_items(
 ) -> SkillSpaceListResult:
     """List authoritative relations, recovering readable names only on one 404."""
 
-    try:
-        response = client.list_skills_by_skill_space(
+    def load_space() -> Any:
+        return client.get_skill_space(skills_types.GetSkillSpaceRequest(Id=space_id))
+
+    def list_relations(request_page: int, request_page_size: int) -> Any:
+        return client.list_skills_by_skill_space(
             skills_types.ListSkillsBySkillSpaceRequest(
                 SkillSpaceId=space_id,
-                PageNumber=page,
-                PageSize=page_size,
+                PageNumber=request_page,
+                PageSize=request_page_size,
             )
         )
+
+    try:
+        response = list_relations(page, page_size)
     except Exception as relation_error:
         if not _is_missing_skill_relation(relation_error):
             raise
-        space = client.get_skill_space(skills_types.GetSkillSpaceRequest(Id=space_id))
+        space = load_space()
         space_name = str(getattr(space, "name", "") or "").strip()
         if not space_name:
             raise relation_error
@@ -126,19 +181,18 @@ def list_skill_space_items(
         )
 
     raw_items = list(getattr(response, "items", None) or [])
+    space = None
     metadata: dict[str, dict[str, str]] = {}
     if include_display_metadata and raw_items:
         from .system_spaces import is_shared_space
 
-        space = client.get_skill_space(skills_types.GetSkillSpaceRequest(Id=space_id))
+        space = space or load_space()
         if is_shared_space(space):
             for item in raw_items:
                 skill_id = str(getattr(item, "skill_id", "") or "")
                 if skill_id and skill_id not in metadata:
                     skill = client.get_skill(skills_types.GetSkillRequest(Id=skill_id))
-                    tags = {
-                        tag.key: tag.value for tag in getattr(skill, "tags", None) or []
-                    }
+                    tags = _tags(skill)
                     metadata[skill_id] = {
                         "author": tags.get("author", ""),
                         "sourceVersion": tags.get(SHARED_SOURCE_VERSION_TAG, ""),
@@ -155,11 +209,9 @@ def list_skill_space_items(
             }
             for item in raw_items
         ),
-        total_count=(
-            int(response.total_count)
-            if getattr(response, "total_count", None) is not None
-            else len(raw_items)
-        ),
+        total_count=int(response.total_count)
+        if getattr(response, "total_count", None) is not None
+        else len(raw_items),
     )
 
 
@@ -285,6 +337,24 @@ def require_review_read(
             self._client_factory(region), space_id, skill_id=skill_id, is_admin=is_admin
         )
 
+    def require_space_read(
+        self,
+        *,
+        region: str,
+        space_id: str,
+        author: str,
+        is_admin: bool,
+    ) -> None:
+        from agentkit.sdk.skills import types as skills_types
+
+        require_space_read(
+            self._client_factory(region),
+            skills_types,
+            space_id=space_id,
+            author=author,
+            is_admin=is_admin,
+        )
+
     def ensure_shared_space(self, *, region: str) -> dict[str, object]:
         from .consts import SHARE_SPACE
 
@@ -317,27 +387,61 @@ def list_spaces(
     ) -> dict[str, object]:
         from agentkit.sdk.skills import types as skills_types
 
+        from .system_spaces import is_review_space, is_shared_space
+
         tag_filters = None
         if author:
             tag_filters = [
                 skills_types.TagFilterForSkill(Key="author", Values=[author])
             ]
-        response = self._client_factory(region).list_skill_spaces(
-            skills_types.ListSkillSpacesRequest(
-                PageNumber=page,
-                PageSize=page_size,
-                ProjectName=project_name,
-                TagFilters=tag_filters,
+
+        def list_page(request_page: int, request_page_size: int) -> Any:
+            return self._client_factory(region).list_skill_spaces(
+                skills_types.ListSkillSpacesRequest(
+                    PageNumber=request_page,
+                    PageSize=request_page_size,
+                    ProjectName=project_name,
+                    TagFilters=tag_filters,
+                )
             )
-        )
-        from .system_spaces import is_review_space
+
+        def visible(item: Any) -> bool:
+            if is_review_space(item) or is_shared_space(item):
+                return False
+            return author is None or _tags(item).get("author") == author
+
+        if author:
+            visible_items: list[Any] = []
+            scanned = 0
+            request_page = 1
+            request_page_size = 100
+            while True:
+                response = list_page(request_page, request_page_size)
+                items = list(response.items or [])
+                scanned += len(items)
+                visible_items.extend(item for item in items if visible(item))
+                total = response.total_count
+                if len(items) < request_page_size or (
+                    total is not None and scanned >= total
+                ):
+                    break
+                request_page += 1
+            start = (page - 1) * page_size
+            page_items = visible_items[start : start + page_size]
+            return {
+                "items": [self._space_item(item, region) for item in page_items],
+                "scannedCount": len(page_items),
+                "totalCount": len(visible_items),
+                "page": page,
+                "pageSize": page_size,
+            }
+
+        response = list_page(page, page_size)
 
         items = list(response.items or [])
         return {
             "items": [
-                self._space_item(item, region)
-                for item in items
-                if not is_review_space(item)
+                self._space_item(item, region) for item in items if visible(item)
             ],
             "scannedCount": len(items),
             "totalCount": response.total_count
@@ -694,7 +798,11 @@ def publish_archive(
                 ProjectName=project_name,
                 Tags=[skills_types.TagForSkill(Key="author", Value=author)]
                 + (
-                    [skills_types.TagForSkill(Key="veadk:visibility", Value="shared")]
+                    [
+                        skills_types.TagForSkill(
+                            Key=SKILL_VISIBILITY_TAG, Value=SKILL_VISIBILITY_SHARED
+                        )
+                    ]
                     if shared
                     else []
                 ),
@@ -798,4 +906,6 @@ def _space_item(value: Any, region: str) -> dict[str, object]:
     "SkillRepositoryError",
     "SkillSpaceListResult",
     "list_skill_space_items",
+    "require_space_read",
+    "skill_space_visible_to_author",
 ]

frontend/server/skills/reviews.py

@@ -45,6 +45,8 @@
     REVIEWER_OWNER_TAG,
     SHARED_REVIEW_TAG,
     SHARED_SOURCE_VERSION_TAG,
+    SKILL_VISIBILITY_SHARED,
+    SKILL_VISIBILITY_TAG,
     SCORE_STATUS_TAG,
     SCORE_TOTAL_TAG,
     SCORE_TIME_TAG,
@@ -371,6 +373,7 @@ def decide(
                         "author": application["author"],
                         SHARED_REVIEW_TAG: application_id,
                         SHARED_SOURCE_VERSION_TAG: application["version"],
+                        SKILL_VISIBILITY_TAG: SKILL_VISIBILITY_SHARED,
                     }
                     shared_id, shared_version = self._copy_archive(
                         client,
@@ -417,6 +420,11 @@ def decide(
                         "共享副本信息不完整，请重试",
                         status_code=502,
                     )
+                update_skill_tags(
+                    client,
+                    shared_id,
+                    {SKILL_VISIBILITY_TAG: SKILL_VISIBILITY_SHARED},
+                )
                 update_skill_tags(
                     client,
                     application_id,

frontend/server/skills/service.py

@@ -168,10 +168,10 @@ def skill_files(
         skill_space_name: str | None = None,
         skill_name: str | None = None,
     ) -> dict[str, object]:
-        self._repository.require_review_read(
+        self._repository.require_space_read(
             region=region,
             space_id=space_id,
-            skill_id=skill_id,
+            author=identity.author,
             is_admin=identity.is_admin,
         )
         return self._repository.skill_files(
@@ -194,10 +194,10 @@ def skill_archive(
         skill_space_name: str | None = None,
         skill_name: str | None = None,
     ) -> tuple[bytes, str]:
-        self._repository.require_review_read(
+        self._repository.require_space_read(
             region=region,
             space_id=space_id,
-            skill_id=skill_id,
+            author=identity.author,
             is_admin=identity.is_admin,
         )
         return self._repository.skill_archive(

frontend/server/skills/versions.py

@@ -101,12 +101,14 @@ def _source(
                 status_code=404,
             )
         personal = not is_shared_space(space) and not is_review_space(space)
-        author = _tags(skill).get("author") or _tags(space).get("author")
-        if personal and author and author != identity.author and not identity.is_admin:
+        space_owner = _tags(space).get("author", "")
+        if personal and space_owner != identity.author and not identity.is_admin:
             raise SkillRepositoryError(
-                "SKILL_VERSION_FORBIDDEN", "只能查看自己创建的技能版本", status_code=403
+                "SKILL_VERSION_FORBIDDEN",
+                "只能查看自己创建的技能空间中的版本",
+                status_code=403,
             )
-        can_update = personal and (identity.is_admin or author == identity.author)
+        can_update = personal and (identity.is_admin or space_owner == identity.author)
         return space, skill, relations, can_update
 
     def list(

tests/frontend/server/skills/test_reviews.py

@@ -29,6 +29,8 @@
     REVIEW_STATUS_TAG,
     REVIEW_SOURCE_SKILL_TAG,
     SHARED_SOURCE_VERSION_TAG,
+    SKILL_VISIBILITY_SHARED,
+    SKILL_VISIBILITY_TAG,
 )
 from frontend.server.skills.models import SkillIdentity
 from frontend.server.skills.repository import (
@@ -349,6 +351,7 @@ def test_approval_publishes_exact_snapshot_and_retains_audit_on_retry(setup):
     }
     assert shared_tags["author"] == "alice"
     assert shared_tags[SHARED_SOURCE_VERSION_TAG] == "v3"
+    assert shared_tags[SKILL_VISIBILITY_TAG] == SKILL_VISIBILITY_SHARED
     assert REVIEW_SOURCE_SKILL_TAG not in shared_tags
     require_review_read(cloud, "shared", skill_id=approved["sharedSkillId"])
     assert decide(repository, pending["id"], actor="another-admin") == approved
@@ -397,10 +400,23 @@ def test_failed_final_tag_write_resumes_without_duplicate_public_copy(setup):
         decide(repository, pending["id"], "returned", "cannot return while publishing")
     assert error.value.code == "SKILL_REVIEW_PUBLISHING"
     cloud.fail_tag_call = 0
+    shared_id = cloud.relations["shared"][0].skill_id
+    cloud.skills[shared_id].tags = [
+        tag for tag in cloud.skills[shared_id].tags if tag.key != SKILL_VISIBILITY_TAG
+    ]
+    cloud.tag_calls.clear()
     approved = decide(repository, pending["id"], actor="second-admin")
     assert approved["reviewedBy"] == "admin"
     assert approved["status"] == "approved"
     assert len(cloud.relations["shared"]) == 1
+    shared_tags = {tag.key: tag.value for tag in cloud.skills[shared_id].tags}
+    assert shared_tags[SKILL_VISIBILITY_TAG] == SKILL_VISIBILITY_SHARED
+    assert any(
+        call[1]["ResourceIds"] == [shared_id]
+        and call[1]["Tags"]
+        == [{"Key": SKILL_VISIBILITY_TAG, "Value": SKILL_VISIBILITY_SHARED}]
+        for call in cloud.tag_calls
+    )
 
 
 def test_failed_initial_tags_never_publish_and_failed_publication_can_retry(setup):