fix: bucket name error

backend/src/backend/auth.py

@@ -22,6 +22,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from backend.dependencies import database_session
 from common.auth.jwt import JwtError, issue_jwt, verify_jwt_token
+from common.auth.membership import resolve_is_system_admin
 from common.auth.passwords import verify_password
 from common.db.models import Roles, Users, WorkspaceMembers, Workspaces
 from common.ids import new_ulid
@@ -72,26 +73,6 @@ def _user_payload(
     }
 
 
-async def _resolve_is_system_admin(
-    session: AsyncSession,
-    user: Users,
-) -> bool:
-    """Return True iff the user holds a platform-scoped admin role.
-
-    The check is: ``Users.status == 'active'`` AND
-    ``Users.platform_role_id`` points to a ``Roles`` row whose
-    ``role_code == 'admin'``. Any other shape (no platform_role_id,
-    disabled user, wrong role code) returns False — the frontend reads
-    this to decide whether to show the system-admin entry point.
-    """
-    if user.status != "active" or user.platform_role_id is None:
-        return False
-    platform_role = await session.scalar(
-        select(Roles).where(Roles.role_id == user.platform_role_id)
-    )
-    return platform_role is not None and platform_role.role_code == "admin"
-
-
 def _workspace_payload(
     workspace: Workspaces,
     role: Roles,
@@ -187,7 +168,7 @@ async def login(
     token = issue_jwt(user.user_id, ttl_seconds=COOKIE_TTL_SECONDS)
     _set_session_cookie(request, response, token)
 
-    is_system_admin = await _resolve_is_system_admin(session, user)
+    is_system_admin = await resolve_is_system_admin(session, user)
 
     return {
         "request_id": new_ulid(),
@@ -266,7 +247,7 @@ async def me(
     if user_role_code is None and rows:
         user_role_code = rows[0][1].role_code
 
-    is_system_admin = await _resolve_is_system_admin(session, user)
+    is_system_admin = await resolve_is_system_admin(session, user)
 
     return {
         "request_id": new_ulid(),

backend/src/backend/dependencies.py

@@ -31,10 +31,15 @@ from dataclasses import dataclass
 from typing import AsyncIterator
 
 from fastapi import Depends, HTTPException, Query, Request, status
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from common.auth.jwt import JwtError, verify_jwt_token
-from common.auth.membership import MembershipError, load_active_membership
+from common.auth.membership import (
+    MembershipError,
+    load_active_membership,
+    resolve_is_system_admin,
+)
 from common.db import session_scope
 from common.db.models import Roles, Users, Workspaces
 from common.ids import new_ulid
@@ -49,10 +54,18 @@ class RequestContext:
     user: Users
     workspace: Workspaces
     role: Roles
+    # True when the requester holds the platform-scoped admin role (via
+    # Users.platform_role_id). Same flag exposed on ``/api/v1/auth/me``
+    # so the frontend can render the platform-admin entry point. System
+    # admins have full control over every workspace — including disabled
+    # ones — so ``request_context`` lets them through and this flag is
+    # the single signal handlers use to gate platform-only operations.
+    is_system_admin: bool = False
 
     @property
     def is_admin(self) -> bool:
-        return self.role.role_code == "admin"
+        """Workspace admin OR system admin (the latter is strictly stronger)."""
+        return self.is_system_admin or self.role.role_code == "admin"
 
 
 async def database_session(request: Request) -> AsyncIterator[AsyncSession]:
@@ -97,7 +110,7 @@ async def request_context(
     ),
     session: AsyncSession = Depends(database_session),
 ) -> RequestContext:
-    """Verify JWT and load the user's active membership for ``workspace_id``.
+    """Verify JWT and load the user's membership for ``workspace_id``.
 
     The ``request_id`` comes from the ``X-Request-ID`` header if
     present, else from a freshly minted ULID. Handlers receive the
@@ -105,8 +118,37 @@ async def request_context(
     header-based implementation, so the 56 ``Depends(request_context)``
     call sites in this repo stay working without changes — they just
     now pass ``?workspace_id=`` instead of the old headers.
+
+    System admins (Users.platform_role_id → admin role) bypass the
+    active-membership requirement: they can address disabled workspaces
+    because they own the platform. Non-admin users still need an
+    active ``WorkspaceMembers`` row in an active ``Workspaces`` row.
     """
     user = await current_user(request, session)
+    is_system_admin = await resolve_is_system_admin(session, user)
+
+    if is_system_admin:
+        # System admin: any workspace (active or disabled) is fine.
+        # Still 404 if the workspace_id is genuinely unknown — the
+        # ``?workspace_id=`` query param is part of the URL contract.
+        workspace = await session.get(Workspaces, workspace_id)
+        if workspace is None:
+            raise HTTPException(
+                status.HTTP_404_NOT_FOUND,
+                "workspace not found",
+            )
+        # Synthesize a role object so the rest of ``RequestContext``
+        # (and downstream ``is_admin`` checks) keep working without
+        # branching on whether a real WorkspaceMembers row exists.
+        role = await _load_admin_role(session)
+        if role is None:
+            # The seed migration creates this row, so this is a
+            # hard config error if it's missing.
+            raise HTTPException(
+                status.HTTP_500_INTERNAL_SERVER_ERROR,
+                "admin role not configured",
+            )
+    else:
         try:
             _user, workspace, role = await load_active_membership(
                 session, user.user_id, workspace_id,
@@ -116,10 +158,19 @@ async def request_context(
                 status.HTTP_403_FORBIDDEN,
                 "active workspace membership is required",
             ) from exc
+
     request_id = request.headers.get("X-Request-ID") or new_ulid()
     return RequestContext(
         request_id=request_id,
         user=user,
         workspace=workspace,
         role=role,
+        is_system_admin=is_system_admin,
+    )
+
+
+async def _load_admin_role(session: AsyncSession) -> Roles | None:
+    """Return the singleton ``role_code='admin'`` row (None if absent)."""
+    return await session.scalar(
+        select(Roles).where(Roles.role_code == "admin")
     )

backend/src/backend/main.py

@@ -10,7 +10,13 @@ from fastapi.routing import APIRoute
 from common.config import settings
 from common.db import create_database_engine, create_session_factory
 from common.service_app import create_service_app
-from common.storage import AsyncStorageBackend, PURPOSE_BUCKETS, build_storage_config, create_storage
+from common.storage import (
+    AsyncStorageBackend,
+    PURPOSE_BUCKETS,
+    actual_bucket_name,
+    build_storage_config,
+    create_storage,
+)
 from backend.admin import router as admin_router
 from backend.auth import router as auth_router
 from backend.platform import router as platform_router
@@ -31,10 +37,13 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
 
     # Storage API is part of the backend process. Platform routers call
     # the helpers in ``backend.services.storage`` directly (in-process),
-    # so no HTTP client is needed. ``build_storage_config`` picks s3 vs
+    # so no HTTP client is needed. The dict is keyed by the actual
-    # local based on ``settings.storage_backend``.
+    # bucket name (e.g. "versions"), matching ``UploadSessions.bucket_name``
+    # and ``StorageObjects.bucket_name`` so call sites can do
+    # ``object_stores[upload.bucket_name].put(...)`` directly.
     app.state.object_stores: dict[str, AsyncStorageBackend] = {
-        name: create_storage(build_storage_config(name)) for name in PURPOSE_BUCKETS
+        actual_bucket_name(purpose): create_storage(build_storage_config(purpose))
+        for purpose in PURPOSE_BUCKETS
     }
     app.state.default_bucket = settings.s3_workspace_bucket
     runtime_http_client = httpx.AsyncClient(

backend/src/backend/storage_api.py

@@ -23,6 +23,7 @@ from common.service_app import create_service_app
 from common.storage import (
     AsyncStorageBackend,
     PURPOSE_BUCKETS,
+    actual_bucket_name,
     build_storage_config,
     create_storage,
 )
@@ -62,20 +63,26 @@ def safe_file_name(value: str) -> str:
     return name
 
 
-# Map an upload's usage_type to the S3 bucket that should hold the
+# Map an upload's usage_type to its purpose (which then resolves to the
-# resulting object. ``usage_type`` is the only signal available at the
+# actual bucket / directory via ``actual_bucket_name``). Keeping the
-# storage edge (the request comes from either the public API or the
+# purpose as the intermediate value means s3 mode and local mode share
-# internal schedule worker), so we make the routing decision in one place
+# the same routing logic — only the final ``actual_bucket_name`` differs.
-# here and let every other layer — server-object create, multipart upload,
+USAGE_TYPE_TO_PURPOSE: dict[str, str] = {
-# direct put — inherit the mapping.
+    "working_copy": "workspace",
+    "public_script": "workspace",
+    "data_resource": "workspace",
+    "snapshot": "workspace",
+    "version_artifact": "version",
+    "run_log": "run_log",
+    "run_result": "run_log",
+}
+
+# Pre-resolved bucket map (for read-only callers like services/storage.py).
+# Re-resolved at module load; re-resolve via resolve_bucket() if the
+# workspace.artifact_bucket override matters.
 BUCKET_FOR_USAGE: dict[str, str] = {
-    "working_copy": settings.s3_workspace_bucket,
+    usage_type: actual_bucket_name(purpose)
-    "public_script": settings.s3_workspace_bucket,
+    for usage_type, purpose in USAGE_TYPE_TO_PURPOSE.items()
-    "data_resource": settings.s3_workspace_bucket,
-    "snapshot": settings.s3_workspace_bucket,
-    "version_artifact": settings.s3_version_bucket,
-    "run_log": settings.s3_run_log_bucket,
-    "run_result": settings.s3_run_log_bucket,
 }
 
 
@@ -92,7 +99,8 @@ def resolve_bucket(
     """
     if workspace.artifact_bucket:
         return workspace.artifact_bucket
-    return BUCKET_FOR_USAGE.get(usage_type, settings.s3_workspace_bucket)
+    purpose = USAGE_TYPE_TO_PURPOSE.get(usage_type, "workspace")
+    return actual_bucket_name(purpose)
 
 
 def storage_payload(item: StorageObjects) -> dict[str, Any]:
@@ -124,9 +132,12 @@ async def lifespan(app: Any) -> AsyncIterator[None]:
     app.state.session_factory = create_session_factory(engine)
     # Buckets are pre-provisioned by the deployment; the storage layer no
     # longer auto-creates them. ``build_storage_config`` picks s3 vs local
-    # based on ``settings.storage_backend``.
+    # based on ``settings.storage_backend``. The dict is keyed by the
+    # actual bucket name so ``object_stores[upload.bucket_name]``
+    # works without a reverse mapping.
     app.state.object_stores: dict[str, AsyncStorageBackend] = {
-        name: create_storage(build_storage_config(name)) for name in PURPOSE_BUCKETS
+        actual_bucket_name(purpose): create_storage(build_storage_config(purpose))
+        for purpose in PURPOSE_BUCKETS
     }
     app.state.default_bucket = settings.s3_workspace_bucket
     try:

common/src/common/auth/membership.py

@@ -22,6 +22,34 @@ class MembershipError(Exception):
     """Raised when the user is not an active member of the workspace."""
 
 
+async def resolve_is_system_admin(
+    session: AsyncSession,
+    user: Users,
+) -> bool:
+    """Return True iff the user holds a platform-scoped admin role.
+
+    The check is: ``Users.status == 'active'`` AND
+    ``Users.platform_role_id`` points to a ``Roles`` row whose
+    ``role_code == 'admin'``. Any other shape (no platform_role_id,
+    disabled user, wrong role code) returns False — the frontend reads
+    this to decide whether to show the system-admin entry point.
+
+    System admins own the platform: they can address disabled workspaces,
+    bypass per-workspace membership checks, etc. Anything that wants to
+    gate "platform-only" behavior (deleting a workspace, soft-deleting
+    a user globally, …) should consult this flag — it is the single
+    source of truth.
+    """
+    from sqlalchemy import select
+
+    if user.status != "active" or user.platform_role_id is None:
+        return False
+    platform_role = await session.scalar(
+        select(Roles).where(Roles.role_id == user.platform_role_id)
+    )
+    return platform_role is not None and platform_role.role_code == "admin"
+
+
 async def load_active_membership(
     session: AsyncSession,
     user_id: str,
@@ -66,4 +94,5 @@ async def load_active_membership(
 __all__ = [
     "MembershipError",
     "load_active_membership",
+    "resolve_is_system_admin",
 ]

common/src/common/storage/__init__.py

@@ -22,6 +22,7 @@ from .base import AsyncStorageBackend, ObjectMeta, StorageBackend
 from .factory import (
     PURPOSE_BUCKETS,
     RCLONE_REMOTE_NAME,
+    actual_bucket_name,
     build_storage_config,
     create_storage,
     rclone_remote_spec,
@@ -32,6 +33,7 @@ from .registry import register_backend, registered_backends
 __all__ = [
     "create_storage",
     "build_storage_config",
+    "actual_bucket_name",
     "workspaces_root",
     "rclone_remote_spec",
     "RCLONE_REMOTE_NAME",

common/src/common/storage/factory.py

@@ -62,6 +62,22 @@ def create_storage(config: Dict[str, Any]) -> AnyStorageBackend:
 PURPOSE_BUCKETS: tuple[str, ...] = ("workspace", "version", "run_log", "trash")
 
 
+def actual_bucket_name(purpose: str) -> str:
+    """把 purpose 名称解析成实际桶路径 / 名（runtime 数据会存在这个字符串里）。
+
+    - s3 模式:`settings.s3_<purpose>_bucket`(e.g. ``"versions"``)
+    - local 模式:`${local_storage_base_dir}/<purpose>`(e.g. ``"/data/version"``)
+
+    这是 ``app.state.object_stores`` 的 dict key——``UploadSessions.bucket_name``
+    和 ``StorageObjects.bucket_name`` 都存这个值,所以 dict 必须用这个串
+    做 key 才能在 ``object_stores[upload.bucket_name]`` 那里直接命中。
+    """
+    from common.config import settings  # 延迟 import 避免循环
+    if settings.storage_backend == "local":
+        return str(Path(settings.local_storage_base_dir) / purpose)
+    return getattr(settings, f"s3_{purpose}_bucket")
+
+
 def build_storage_config(bucket_name: str) -> Dict[str, Any]:
     """根据 ``settings.storage_backend`` 构造 ``create_storage()`` 的入参。