From ff700d8db4216a9a9181f853d5c1dd20ce2eee1a Mon Sep 17 00:00:00 2001 From: Winnie <3308978791@qq.com> Date: Mon, 17 Aug 2026 18:55:31 +0800 Subject: [PATCH] fix: stabilize cron scheduling and run history --- backend/src/backend/admin.py | 11 +++ backend/src/backend/auth.py | 29 ++++---- backend/src/backend/dependencies.py | 23 ++++++- backend/src/backend/jupyter.py | 13 ++++ backend/src/backend/main.py | 29 +++++--- backend/src/backend/platform.py | 36 +++++++++- backend/src/backend/resources.py | 15 +++++ backend/src/backend/runtime_client.py | 9 +++ backend/src/backend/schedule_runs.py | 16 +++++ backend/src/backend/schedules.py | 22 ++++++ backend/src/backend/scripts.py | 28 ++++++++ backend/src/backend/storage_api.py | 23 ++++--- common/src/common/scheduler/trigger.py | 8 +-- .../app/features/schedules/SchedulePage.tsx | 40 ++++++++--- .../schedules/state/schedulesStore.ts | 6 +- schedule/src/schedule/scheduler.py | 67 ++++++++++++++++--- 16 files changed, 320 insertions(+), 55 deletions(-) diff --git a/backend/src/backend/admin.py b/backend/src/backend/admin.py index 2813faa..73628ee 100644 --- a/backend/src/backend/admin.py +++ b/backend/src/backend/admin.py @@ -1,3 +1,9 @@ +"""旧版工作区级员工管理接口。 + +路由前缀为 ``/api/v1/admin``，依赖当前工作区的管理员权限。新的系统级用户、 +工作区和成员管理接口在 ``platform.py``；本模块主要保留给兼容旧前端调用。 +""" + from __future__ import annotations from typing import Any, Literal @@ -39,6 +45,7 @@ class EmployeeUpdate(BaseModel): def require_admin(context: RequestContext) -> None: + # 统一在路由入口处做角色判断，避免每个 CRUD 分支重复写权限代码。 if not context.is_admin: raise HTTPException(status.HTTP_403_FORBIDDEN, "仅管理员可以管理员工") @@ -80,6 +87,7 @@ async def member_row( return row +# 旧版接口：列出当前工作区内的员工及其角色。 @router.get("/employees") async def list_employees( context: RequestContext = Depends(request_context), @@ -107,6 +115,7 @@ async def list_employees( } +# 旧版接口：在当前工作区中创建员工及成员关系。 @router.post("/employees", status_code=status.HTTP_201_CREATED) async def create_employee( payload: EmployeeCreate, @@ -160,6 +169,7 @@ async def create_employee( } +# 旧版接口：更新员工显示信息、角色或状态。 @router.patch("/employees/{user_id}") async def update_employee( user_id: str, @@ -215,6 +225,7 @@ async def update_employee( } +# 旧版接口：移除当前工作区中的员工成员关系。 @router.delete("/employees/{user_id}") async def delete_employee( user_id: str, diff --git a/backend/src/backend/auth.py b/backend/src/backend/auth.py index a68e11b..d23ebab 100644 --- a/backend/src/backend/auth.py +++ b/backend/src/backend/auth.py @@ -1,4 +1,10 @@ -"""Cookie+JWT authentication endpoints. +"""登录与会话认证接口。 + +中文导读：浏览器调用登录接口后，后端把 JWT 放入 HttpOnly Cookie；之后前端 +通过 ``fetch(..., credentials='same-origin')`` 自动携带 Cookie。需要工作区的 +接口会继续由 ``request_context`` 校验 ``workspace_id`` 和成员权限。 + +Cookie+JWT authentication endpoints. The user-facing flow is: 1. POST /api/v1/auth/login — verify password, set HttpOnly cookie @@ -30,10 +36,8 @@ from backend.dependencies import database_session, load_user_permissions router = APIRouter(tags=["auth"]) -# Cookie config. ``secure=True`` requires HTTPS — the only safe -# assumption in production. Dev environments running on plain HTTP -# should reverse-proxy with TLS termination or set the env knob -# (future extension). +# Cookie 配置：生产环境走 HTTPS 时应设置 Secure；本地 HTTP 开发环境会根据 +# 实际请求协议决定是否设置，避免浏览器因 Secure Cookie 而丢弃登录状态。 COOKIE_NAME = "access_token" COOKIE_TTL_SECONDS = 24 * 60 * 60 COOKIE_SAMESITE = "lax" @@ -41,12 +45,8 @@ COOKIE_SAMESITE = "lax" def _set_session_cookie(request: Request, response: Response, token: str) -> None: forwarded_scheme = request.headers.get("x-forwarded-proto", request.url.scheme) - # Operators running behind a TLS-terminating proxy that strips - # X-Forwarded-Proto can opt into forcing the Secure flag via - # ``settings.cookie_force_secure`` — without that override a plain - # HTTP request (no forwarded scheme, no TLS upgrade visible to the - # app) would yield an insecure cookie and modern browsers would - # silently drop it on the HTTPS round trip. + # 反向代理通常用 X-Forwarded-Proto 告诉后端原始协议；若代理未传该头， + # 可通过配置强制启用 Secure，防止 HTTPS 场景下出现不安全 Cookie。 secure = forwarded_scheme == "https" or settings.cookie_force_secure response.set_cookie( key=COOKIE_NAME, @@ -95,6 +95,7 @@ def _workspace_payload( } +# 校验账号密码，设置登录 Cookie，并返回用户可进入的工作区列表。 @router.post("/api/v1/auth/login") async def login( request: Request, @@ -131,8 +132,8 @@ async def login( "invalid username or password", ) - # Pull all active memberships. The earliest join wins as default - # because there is no `is_default` column on `workspace_members`. + # 返回用户可进入的工作区列表；当前没有默认工作区字段，因此最早加入的 + # 工作区作为前端的初始选择。 rows = ( await session.execute( select(Workspaces, Roles, WorkspaceMembers.joined_at) @@ -196,6 +197,7 @@ async def login( } +# 清除浏览器 Cookie，使当前会话立即失效。 @router.post("/api/v1/auth/logout") async def logout(response: Response) -> dict[str, Any]: """Clear the session cookie. Idempotent.""" @@ -207,6 +209,7 @@ async def logout(response: Response) -> dict[str, Any]: } +# 返回当前登录用户、权限和可访问工作区，用于前端初始化登录态。 @router.get("/api/v1/auth/me") async def me( request: Request, diff --git a/backend/src/backend/dependencies.py b/backend/src/backend/dependencies.py index 5fc7103..5ea6575 100644 --- a/backend/src/backend/dependencies.py +++ b/backend/src/backend/dependencies.py @@ -1,4 +1,16 @@ -"""FastAPI dependencies for the public API. +"""公共 FastAPI 依赖。 + +中文导读： + +* ``database_session``：为一次请求提供数据库事务；成功提交、异常回滚。 +* ``current_user``：只验证登录 Cookie 并得到当前用户，不关心工作区。 +* ``request_context``：普通业务接口最常使用的依赖，同时验证用户、 + ``workspace_id``、工作区成员关系和角色权限。 + +路由函数把这些函数写进 ``Depends(...)`` 后，FastAPI 会先完成校验，再把 +结果作为参数传给路由函数；因此业务代码无需重复解析 Cookie 或查询成员关系。 + +FastAPI dependencies for the public API. Two distinct concerns live here: @@ -48,6 +60,12 @@ ACCESS_TOKEN_COOKIE = "access_token" @dataclass(frozen=True) class RequestContext: + """已完成认证和工作区授权后的请求上下文。 + + 路由依赖 ``request_context`` 后会得到该对象，用其中的用户、工作区和角色 + 执行业务权限判断，避免在每个接口中重复查询。 + """ + request_id: str user: Users workspace: Workspaces @@ -67,6 +85,7 @@ class RequestContext: async def database_session(request: Request) -> AsyncIterator[AsyncSession]: + # 一个请求对应一个事务范围，避免不同请求意外共用同一个 Session。 async with session_scope(request.app.state.session_factory) as session: yield session @@ -82,6 +101,7 @@ async def current_user( a workspace-scoped context, or use ``Depends(current_user)`` for workspace-agnostic endpoints (e.g. ``/api/v1/auth/me``). """ + # 登录接口写入 HttpOnly Cookie；后续浏览器请求会自动携带它。 token = request.cookies.get(ACCESS_TOKEN_COOKIE) if not token: raise HTTPException(status.HTTP_401_UNAUTHORIZED, "not authenticated") @@ -122,6 +142,7 @@ async def request_context( 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) diff --git a/backend/src/backend/jupyter.py b/backend/src/backend/jupyter.py index 97e01d7..74d01e1 100644 --- a/backend/src/backend/jupyter.py +++ b/backend/src/backend/jupyter.py @@ -1,3 +1,12 @@ +"""Jupyter 访问的鉴权桥接。 + +浏览器访问 ``/jupyter/{workspace_id}/...`` 时，Nginx 会先向本模块的 +``/api/v1/auth/jupyter`` 发起内部 auth_request。后端验证用户、工作区和文件 +锁，再把实际 Jupyter 地址与内部令牌写进响应头，由 Nginx 转发请求。 + +本模块只负责鉴权和路由选择；真正启动、管理 Jupyter 进程的是 ``runtime`` 服务。 +""" + import re from common.auth.jwt import JwtError, verify_jwt_token @@ -44,6 +53,8 @@ async def check_notebook_is_locked( different user, and is currently locked. The owner is always let through; a missing row is treated as "not owned yet" and allowed. """ + # 锁信息保存在数据库的 Scripts 记录中，而非前端内存；因此多浏览器/多用户 + # 访问同一 notebook 时也能得到一致结果。 statement = select(Scripts.owner_user_id, Scripts.is_locked).where( Scripts.workspace_id == workspace_id, Scripts.script_name == notebook_path, @@ -78,6 +89,8 @@ async def load_active_membership_or_403( ) from exc +# 供 Nginx auth_request 调用：验证访问 Jupyter 的身份、成员关系和文件锁， +# 再返回应转发到的 Jupyter 地址及内部令牌。 @router.get("/api/v1/auth/jupyter") async def verify_jupyter_access( request: Request, diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py index 5e71f63..f9df7ae 100644 --- a/backend/src/backend/main.py +++ b/backend/src/backend/main.py @@ -1,3 +1,14 @@ +"""后端服务总入口。 + +这里负责两件事： +1. 在应用启动时准备数据库、对象存储以及 Runtime 的 HTTP 客户端； +2. 将各业务模块的路由注册到同一个 FastAPI 应用中。 + +浏览器请求先经过 Nginx 的 ``/api/`` 反向代理，随后才会到达本文件创建的 +应用。具体接口实现分别在 ``auth.py``、``scripts.py``、``schedules.py`` 等 +模块中。 +""" + from __future__ import annotations import time @@ -38,15 +49,14 @@ configure_logging(settings.log_level) @asynccontextmanager async def lifespan(app: Any) -> AsyncIterator[None]: + # 生命周期内创建的对象挂在 app.state 上，路由通过 Depends 或 Request + # 取得它们；这样每个请求不会重复创建数据库连接或 HTTP 客户端。 engine = create_database_engine(settings.database_url) app.state.session_factory = create_session_factory(engine) - # 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. The dict is keyed by the actual - # 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. + # 存储接口与业务路由运行在同一个 backend 进程中，因此直接复用存储对象， + # 不需要再通过 HTTP 调用自己。字典键使用真实桶名，便于上传会话和存储 + # 对象记录直接定位对应的存储后端。 app.state.object_stores: dict[str, StorageBackend | AsyncStorageBackend] = { # noqa: F821 actual_bucket_name(purpose): create_storage(build_storage_config(purpose)) for purpose in PURPOSE_BUCKETS @@ -63,7 +73,7 @@ async def lifespan(app: Any) -> AsyncIterator[None]: }, ) app.state.runtime_client = RuntimeClient(runtime_http_client) - # Short timeout — refresh is best-effort and runs in a BackgroundTask. + # rclone 目录缓存刷新属于尽力而为的后台动作，不能拖慢用户保存文件的请求。 rclone_http_client = httpx.AsyncClient( base_url=settings.rclone_rc_url, timeout=httpx.Timeout(30.0), @@ -81,6 +91,7 @@ app = create_service_app( settings.service_name, lifespan=lifespan, ) +# 面向浏览器的公开 API：认证、脚本、资源、调度和系统管理。 app.include_router(auth_router) app.include_router(jupyter_router) app.include_router(resources_router) @@ -90,12 +101,14 @@ app.include_router(scripts_router) app.include_router(admin_router) app.include_router(platform_router) -# Reuse the proven storage endpoints without running another FastAPI service. +# 内部存储接口额外加上 /internal 前缀，供后端服务间调用，不作为普通前端 API。 app.include_router(storage_api_router, prefix="/internal") @app.middleware("http") async def access_log(request: Request, call_next): + # 每个 HTTP 请求都记录方法、路径、状态码和耗时；排查页面请求失败时， + # Docker Desktop 中 backend 容器的 Logs 就会显示这里生成的日志。 start = time.perf_counter() try: response = await call_next(request) diff --git a/backend/src/backend/platform.py b/backend/src/backend/platform.py index 92f176e..3af6959 100644 --- a/backend/src/backend/platform.py +++ b/backend/src/backend/platform.py @@ -1,4 +1,10 @@ -"""System-admin (platform-scope) endpoints for workspace & membership management. +"""系统级管理接口。 + +中文导读：本模块管理全平台的用户、工作区、成员关系和角色权限。它使用 +``system_admin_context`` 进行平台管理员校验，因此不要求请求者先加入某个具体 +工作区；普通工作区内的业务接口则使用 ``request_context``。 + +System-admin (platform-scope) endpoints for workspace & membership management. All routes under ``/api/v1/platform/*`` are gated by :func:`system_admin_context`, which requires the requester to hold a @@ -102,6 +108,7 @@ MEMBER_STATUS_VALUES = ("active", "disabled", "locked") # --------------------------------------------------------------------------- +# 创建工作区时前端提交的请求体；禁止未声明字段。 class WorkspaceCreate(BaseModel): model_config = ConfigDict(extra="forbid") @@ -111,6 +118,7 @@ class WorkspaceCreate(BaseModel): description: str | None = Field(default=None, max_length=1000) +# 编辑工作区时允许修改的字段；禁用操作必须走删除接口而不是直接传状态。 class WorkspaceUpdate(BaseModel): model_config = ConfigDict(extra="forbid") @@ -121,6 +129,7 @@ class WorkspaceUpdate(BaseModel): status: Literal["active", "archived"] | None = None +# 将已存在用户加入工作区的请求体；角色继承用户的平台角色。 class MemberCreate(BaseModel): """Add a user to a workspace. Role is inherited from the user's platform role (Users.platform_role_id) — not set here.""" @@ -130,6 +139,7 @@ class MemberCreate(BaseModel): user_id: str = Field(min_length=26, max_length=26) +# 更新成员在该工作区中的可用状态，不直接在这里修改平台角色。 class MemberUpdate(BaseModel): """Update a workspace membership's status. Role cannot be changed via this endpoint — workspace role is always inherited from the @@ -141,6 +151,7 @@ class MemberUpdate(BaseModel): member_status: Literal["active", "disabled", "locked"] | None = None +# 新建平台用户的请求体；创建用户不等同于把用户加入某个工作区。 class PlatformEmployeeCreate(BaseModel): model_config = ConfigDict(extra="forbid") @@ -151,6 +162,7 @@ class PlatformEmployeeCreate(BaseModel): role_code: Literal["admin", "developer"] | None = None +# 修改平台用户资料、状态或平台角色的请求体。 class PlatformEmployeeUpdate(BaseModel): model_config = ConfigDict(extra="forbid") @@ -160,6 +172,7 @@ class PlatformEmployeeUpdate(BaseModel): role_code: Literal["admin", "developer"] | None = None +# 用完整权限集合替换某个平台角色菜单权限的请求体。 class RolePermissionsPatch(BaseModel): """Replace a platform role's permission set wholesale. @@ -181,6 +194,8 @@ class RolePermissionsPatch(BaseModel): @dataclass(frozen=True) class SystemAdminContext: + """通过系统管理员校验后的上下文，只包含当前用户和请求追踪 ID。""" + """Resolved identity for a system-admin request. Carries the request id, the authenticated user row, and the resolved @@ -197,6 +212,7 @@ async def system_admin_context( request: Request, session: AsyncSession = Depends(database_session), ) -> SystemAdminContext: + """验证当前用户是否为平台管理员，供 /api/v1/platform 下的路由依赖。""" """Resolve the requester as a system admin. Steps: @@ -368,6 +384,7 @@ def _envelope(request_id: str, data: Any, meta: dict[str, Any] | None = None) -> # --------------------------------------------------------------------------- +# 列出整个平台的非删除用户；不局限于某一个工作区。 @router.get("/employees") async def list_platform_employees( context: SystemAdminContext = Depends(system_admin_context), @@ -389,6 +406,7 @@ async def list_platform_employees( ) +# 创建平台用户；后续可再通过成员接口把该用户加入工作区。 @router.post("/employees", status_code=status.HTTP_201_CREATED) async def create_platform_employee( payload: PlatformEmployeeCreate, @@ -429,6 +447,7 @@ async def create_platform_employee( ) +# 更新平台用户资料、账号状态或平台角色，同时保护最少管理员等约束。 @router.patch("/employees/{user_id}") async def update_platform_employee( user_id: str, @@ -577,6 +596,7 @@ async def update_platform_employee( ) +# 软删除平台用户，并级联标记其工作区成员关系为删除。 @router.delete("/employees/{user_id}") async def delete_platform_employee( user_id: str, @@ -629,6 +649,7 @@ async def delete_platform_employee( # --------------------------------------------------------------------------- +# 列出平台中全部未删除工作区。 @router.get("/workspaces") async def list_workspaces( context: SystemAdminContext = Depends(system_admin_context), @@ -656,6 +677,7 @@ async def list_workspaces( ) +# 创建工作区，并将当前系统管理员初始化为该工作区管理员。 @router.post("/workspaces", status_code=status.HTTP_201_CREATED) async def create_workspace( payload: WorkspaceCreate, @@ -702,6 +724,7 @@ async def create_workspace( return _envelope(context.request_id, workspace_payload(workspace)) +# 读取单个工作区详情，包含已归档或禁用状态。 @router.get("/workspaces/{workspace_id}") async def get_workspace( workspace_id: str, @@ -713,6 +736,7 @@ async def get_workspace( return _envelope(context.request_id, workspace_payload(workspace)) +# 更新工作区可编辑属性，例如名称、配额、描述和归档状态。 @router.patch("/workspaces/{workspace_id}") async def update_workspace( workspace_id: str, @@ -740,6 +764,7 @@ async def update_workspace( return _envelope(context.request_id, workspace_payload(workspace)) +# 软删除/禁用工作区，并级联处理其活动成员关系。 @router.delete("/workspaces/{workspace_id}") async def delete_workspace( workspace_id: str, @@ -780,6 +805,7 @@ async def delete_workspace( # --------------------------------------------------------------------------- +# 列出一个工作区的活动成员与成员状态。 @router.get("/workspaces/{workspace_id}/members") async def list_members( workspace_id: str, @@ -811,6 +837,7 @@ async def list_members( ) +# 将已有平台用户加入指定工作区。 @router.post( "/workspaces/{workspace_id}/members", status_code=status.HTTP_201_CREATED, @@ -894,6 +921,7 @@ async def add_member( return _envelope(context.request_id, member_payload(user, role, membership)) +# 更新成员状态，例如禁用或锁定；同时保证工作区不会失去最后一个管理员。 @router.patch("/workspaces/{workspace_id}/members/{user_id}") async def update_member( workspace_id: str, @@ -952,6 +980,7 @@ async def update_member( return _envelope(context.request_id, member_payload(user, role, membership)) +# 移除某个工作区成员，并保护最后一名管理员及当前操作者的安全约束。 @router.delete("/workspaces/{workspace_id}/members/{user_id}") async def remove_member( workspace_id: str, @@ -1062,6 +1091,7 @@ def _role_payload(role: Roles, permission_codes: list[str]) -> dict[str, Any]: } +# 列出平台角色及其拥有的菜单权限代码。 @router.get("/roles") async def list_platform_roles( context: SystemAdminContext = Depends(system_admin_context), @@ -1086,6 +1116,7 @@ async def list_platform_roles( ) +# 获取一个角色当前配置的权限代码集合。 @router.get("/roles/{role_code}/permissions") async def get_role_permissions( role_code: str, @@ -1100,6 +1131,7 @@ async def get_role_permissions( ) +# 以请求中的完整集合更新角色权限，并保留管理员角色的必要系统权限。 @router.patch("/roles/{role_code}/permissions") async def patch_role_permissions( role_code: str, @@ -1242,4 +1274,4 @@ __all__ = [ "SystemAdminContext", "router", "system_admin_context", -] \ No newline at end of file +] diff --git a/backend/src/backend/resources.py b/backend/src/backend/resources.py index bed6630..bd4baf1 100644 --- a/backend/src/backend/resources.py +++ b/backend/src/backend/resources.py @@ -1,3 +1,10 @@ +"""工作区数据资源 API。 + +资源上传分为三步：创建上传会话 → 写入文件字节 → 绑定为可见的数据资源。 +这种拆分使前端可以分别处理元数据、文件传输和最终展示；资源文件本身由存储层 +保存，数据库只保存资源与存储对象的关联关系。 +""" + from __future__ import annotations import os @@ -134,6 +141,7 @@ def can_view(resource: DataResources, context: RequestContext) -> bool: ) +# 根据当前脚本位置计算资源的相对路径，便于 Notebook 中用相对路径读取文件。 @router.post("/{resource_id}/jupyter-relative-path") async def resource_jupyter_relative_path( resource_id: str, @@ -163,6 +171,7 @@ async def resource_jupyter_relative_path( } +# 上传第 1 步：创建上传会话，登记文件名、大小、类型等预期元数据。 @router.post("/uploads", status_code=status.HTTP_201_CREATED) async def create_resource_upload( payload: CreateResourceUploadRequest, @@ -193,6 +202,7 @@ async def create_resource_upload( return {"request_id": context.request_id, "data": data, "meta": {}} +# 上传第 2 步：将浏览器传来的二进制文件写入已创建的上传会话。 @router.put("/uploads/{upload_id}") async def upload_resource_bytes( upload_id: str, @@ -216,6 +226,7 @@ async def upload_resource_bytes( } +# 上传第 3 步：把已完成的上传会话绑定为工作区可见的数据资源。 @router.post("/uploads/{upload_id}/bind") async def bind_resource( upload_id: str, @@ -349,6 +360,7 @@ async def bind_resource( } +# 列出当前工作区可见的数据资源，可按可见性或关键字筛选。 @router.get("") async def list_resources( context: RequestContext = Depends(request_context), @@ -428,6 +440,7 @@ async def get_visible_resource( return row +# 查询单个数据资源的元数据与其关联文件信息。 @router.get("/{resource_id}") async def get_resource( resource_id: str, @@ -446,6 +459,7 @@ async def get_resource( } +# 为资源文件生成带时效的下载链接。 @router.post("/{resource_id}/download-url") async def resource_download_url( resource_id: str, @@ -467,6 +481,7 @@ async def resource_download_url( return {"request_id": context.request_id, "data": data["data"], "meta": {}} +# 软删除数据资源及其关联对象，遵循存储层的回收站策略。 @router.delete("/{resource_id}") async def delete_resource( resource_id: str, diff --git a/backend/src/backend/runtime_client.py b/backend/src/backend/runtime_client.py index c2cd6f1..1ca859b 100644 --- a/backend/src/backend/runtime_client.py +++ b/backend/src/backend/runtime_client.py @@ -1,3 +1,10 @@ +"""Backend 调用 Runtime 服务的轻量 HTTP 客户端。 + +后端不直接管理 Jupyter 进程；涉及工作区运行时状态、编辑会话或访问票据时， +会通过本客户端请求 Docker 内网中的 ``runtime`` 服务。网络错误会统一包装成 +``RuntimeClientError``，让路由返回可识别的 503 错误。 +""" + from __future__ import annotations from dataclasses import dataclass @@ -34,6 +41,8 @@ class RuntimeClient: path: str, payload: dict[str, Any], ) -> dict[str, Any]: + # 将 httpx 的连接/响应异常转换为项目统一的业务异常，调用方无需关心 + # 底层 HTTP 客户端的具体异常类型。 try: response = await self.client.request(method, path, json=payload) except httpx.RequestError as exc: diff --git a/backend/src/backend/schedule_runs.py b/backend/src/backend/schedule_runs.py index c40e578..43ff14f 100644 --- a/backend/src/backend/schedule_runs.py +++ b/backend/src/backend/schedule_runs.py @@ -1,3 +1,10 @@ +"""调度运行的触发与查询 API。 + +手动运行接口只负责在 MySQL 中创建 ``ScheduleRuns`` 和 Outbox 事件；真正执行 +任务的是 ``schedule`` 容器，它轮询 Outbox 后运行 DAG 节点。本模块还提供运行 +历史、节点状态、日志和结果文件的查询入口。 +""" + from __future__ import annotations from datetime import UTC, datetime @@ -51,6 +58,7 @@ RunStatus = Literal[ ] +# 手动触发调度时允许前端附带的运行原因；严格拒绝未定义字段。 class RunScheduleRequest(StrictModel): reason: str = Field(default="manual_run", min_length=1, max_length=255) @@ -64,6 +72,8 @@ def _iso(value: datetime | None) -> str | None: def _http_error_from_trigger(exc: TriggerError) -> HTTPException: + # 调度领域异常统一转换为 HTTP 状态码，前端可据此区分“找不到任务”、 + # “DAG 无效”和“请求参数不合法”等情况。 if isinstance(exc, ScheduleNotFound): return HTTPException(status.HTTP_404_NOT_FOUND, str(exc)) if isinstance(exc, InvalidDag): @@ -226,6 +236,7 @@ async def _artifact_bytes( ) +# 立即触发一次调度：写入运行记录和 Outbox，由 schedule 容器异步接手执行。 @router.post( "/api/v1/schedules/{schedule_id}/run", status_code=status.HTTP_202_ACCEPTED, @@ -276,6 +287,7 @@ async def run_schedule_now( } +# 按调度或状态筛选运行历史，供前端运行记录列表展示。 @router.get("/api/v1/schedule-runs") async def list_schedule_runs( schedule_id: str | None = Query(default=None), @@ -305,6 +317,7 @@ async def list_schedule_runs( } +# 查询一次运行的详情，包括每个节点的执行状态。 @router.get("/api/v1/schedule-runs/{run_id}") async def get_schedule_run( run_id: str, @@ -319,6 +332,7 @@ async def get_schedule_run( } +# 返回某个节点运行关联的日志/结果产物元数据及可访问地址。 @router.get( "/api/v1/schedule-runs/{run_id}/node-runs/{node_run_id}/artifacts" ) @@ -361,6 +375,7 @@ async def get_schedule_node_run_artifacts( } +# 读取节点运行日志正文，通常由前端日志面板按需调用。 @router.get( "/api/v1/schedule-runs/{run_id}/node-runs/{node_run_id}/logs" ) @@ -387,6 +402,7 @@ async def read_schedule_node_run_logs( ) +# 为节点运行结果生成下载响应或重定向地址。 @router.get( "/api/v1/schedule-runs/{run_id}/node-runs/{node_run_id}/result" ) diff --git a/backend/src/backend/schedules.py b/backend/src/backend/schedules.py index 5976343..d2cc110 100644 --- a/backend/src/backend/schedules.py +++ b/backend/src/backend/schedules.py @@ -1,3 +1,10 @@ +"""DAG 调度定义 API。 + +这里管理调度任务本身：名称、Cron、节点、边和 DAG 校验；实际的定时轮询与节点 +执行由独立的 ``schedule`` 容器完成。``workflow_version`` 用于乐观并发控制： +前端修改画布时必须携带当前版本，避免两个人的编辑互相覆盖。 +""" + from __future__ import annotations import heapq @@ -70,6 +77,7 @@ def cron_preview( count: int, base_time: datetime | None = None, ) -> dict[str, Any]: + # 该接口只计算并预览下几次触发时间，不会创建或修改任何调度任务。 if len(expression.split()) != 5: raise HTTPException( status.HTTP_422_UNPROCESSABLE_ENTITY, @@ -464,6 +472,7 @@ async def _require_valid_when_enabled( ) +# 根据 Cron 表达式预览未来触发时间，不会保存或执行任务。 @router.post("/api/v1/cron/preview") async def preview_cron( payload: CronPreviewRequest, @@ -481,6 +490,7 @@ async def preview_cron( } +# 列出调度产生的可展示版本/产物，供前端结果面板使用。 @router.get("/api/v1/schedule-artifacts") async def list_schedule_artifacts( limit: int = Query(default=100, ge=1, le=500), @@ -530,6 +540,7 @@ async def list_schedule_artifacts( } +# 列出当前工作区的调度定义及其节点、边数量等摘要信息。 @router.get("/api/v1/schedules") async def list_schedules( context: RequestContext = Depends(request_context), @@ -591,6 +602,7 @@ async def list_schedules( } +# 创建新的 DAG 调度定义；初始状态不包含节点和边。 @router.post( "/api/v1/schedules", status_code=status.HTTP_201_CREATED, @@ -643,6 +655,7 @@ async def create_schedule( } +# 获取一个调度的完整画布数据，包括节点、边和当前工作流版本。 @router.get("/api/v1/schedules/{schedule_id}") async def get_schedule( schedule_id: str, @@ -657,6 +670,7 @@ async def get_schedule( } +# 更新调度基本属性，如名称、Cron、时区、是否启用和并发策略。 @router.put("/api/v1/schedules/{schedule_id}") @router.patch("/api/v1/schedules/{schedule_id}") async def update_schedule( @@ -721,6 +735,7 @@ async def update_schedule( } +# 删除调度定义；请求携带 workflow_version 以避免误删他人刚修改的画布。 @router.delete("/api/v1/schedules/{schedule_id}") async def delete_schedule( schedule_id: str, @@ -752,6 +767,7 @@ async def delete_schedule( } +# 向调度画布新增一个执行节点，并关联已发布的脚本版本。 @router.post( "/api/v1/schedules/{schedule_id}/nodes", status_code=status.HTTP_201_CREATED, @@ -807,6 +823,7 @@ async def create_schedule_node( } +# 更新节点名称、执行参数、超时、重试和画布坐标等配置。 @router.put("/api/v1/schedules/{schedule_id}/nodes/{node_id}") async def update_schedule_node( schedule_id: str, @@ -862,6 +879,7 @@ async def update_schedule_node( } +# 从调度画布删除节点，并同步清理关联边。 @router.delete("/api/v1/schedules/{schedule_id}/nodes/{node_id}") async def delete_schedule_node( schedule_id: str, @@ -915,6 +933,7 @@ async def delete_schedule_node( } +# 在两个节点之间新增依赖边，表示目标节点必须等待源节点完成。 @router.post( "/api/v1/schedules/{schedule_id}/edges", status_code=status.HTTP_201_CREATED, @@ -991,6 +1010,7 @@ async def create_schedule_edge( } +# 修改一条依赖边的条件表达式或其他可编辑字段。 @router.put("/api/v1/schedules/{schedule_id}/edges/{edge_id}") async def update_schedule_edge( schedule_id: str, @@ -1025,6 +1045,7 @@ async def update_schedule_edge( } +# 删除节点之间的依赖关系，不会删除节点本身。 @router.delete("/api/v1/schedules/{schedule_id}/edges/{edge_id}") async def delete_schedule_edge( schedule_id: str, @@ -1059,6 +1080,7 @@ async def delete_schedule_edge( } +# 校验画布是否为可执行 DAG，例如是否存在环、孤立节点或无效版本。 @router.post("/api/v1/schedules/{schedule_id}/validate") async def validate_schedule( schedule_id: str, diff --git a/backend/src/backend/scripts.py b/backend/src/backend/scripts.py index 918281d..8b18722 100644 --- a/backend/src/backend/scripts.py +++ b/backend/src/backend/scripts.py @@ -1,3 +1,13 @@ +"""工作区脚本与 Notebook 的公开 API。 + +脚本元数据（名称、归属、锁、版本关系）保存在 MySQL；文件正文和版本产物保存到 +对象存储或本地 ``data/``。本模块负责把两者保持一致，并提供目录、上传、编辑锁、 +版本发布和下载地址等接口。 + +前端的 ``frontend/app/services/api.ts`` 通过 ``/api/v1/scripts`` 和 +``/api/v1/workspace-directories`` 调用这里的路由。 +""" + import base64 import hashlib import json @@ -555,6 +565,7 @@ async def create_script_record( return script, storage_object +# 新建空的 Python 脚本或 Notebook：同时创建数据库元数据和初始文件内容。 @router.post("/api/v1/scripts", status_code=status.HTTP_201_CREATED) async def create_script( payload: CreateScriptRequest, @@ -587,6 +598,7 @@ async def create_script( } +# 上传现有脚本文件：校验文件名/类型后写入存储，并建立 Scripts 记录。 @router.post( "/api/v1/scripts/upload", status_code=status.HTTP_201_CREATED, @@ -647,6 +659,7 @@ async def upload_script( } +# 返回旧版一次性完整目录树，保留给兼容旧前端；新页面通常按目录懒加载。 @router.get("/api/v1/workspace-tree") async def get_workspace_tree( context: RequestContext = Depends(request_context), @@ -718,6 +731,7 @@ async def get_workspace_tree( } +# 查询某个目录下的直接子目录，供前端按需展开工作区树。 @router.get("/api/v1/workspace-directories") async def list_workspace_directories( parent_path: str = Query(default=""), @@ -790,6 +804,7 @@ async def list_workspace_directories( } +# 在工作区内创建逻辑目录；目录信息由脚本相对路径推导，不对应容器本地文件夹。 @router.post( "/api/v1/workspace-directories", status_code=status.HTTP_201_CREATED, @@ -946,6 +961,7 @@ async def create_workspace_directory( } +# 删除逻辑目录及其下属脚本记录；实际文件按存储层的软删除规则处理。 @router.delete("/api/v1/workspace-directories") async def delete_workspace_directory( request: Request, @@ -1028,6 +1044,7 @@ async def delete_workspace_directory( } +# 列出当前工作区中用户有权查看的脚本与 Notebook 元数据。 @router.get("/api/v1/scripts") async def list_scripts( context: RequestContext = Depends(request_context), @@ -1057,6 +1074,7 @@ async def list_scripts( } +# 读取脚本正文或 Notebook JSON；编辑器打开文件时调用此接口。 @router.get("/api/v1/scripts/{script_id}/content") async def get_script_content( script_id: str, @@ -1107,6 +1125,7 @@ async def get_script_content( } +# 查询单个脚本的元数据，例如类型、路径、锁状态和拥有者。 @router.get("/api/v1/scripts/{script_id}") async def get_script( script_id: str, @@ -1125,6 +1144,7 @@ async def get_script( } +# 保存编辑器提交的新内容；会校验工作区权限和文件编辑锁。 @router.put("/api/v1/scripts/{script_id}") async def update_script( script_id: str, @@ -1190,6 +1210,7 @@ async def update_script( } +# 修改脚本锁定状态，避免其他用户同时编辑同一份文件。 @router.patch("/api/v1/scripts/{script_id}/lock") async def set_script_lock( script_id: str, @@ -1232,6 +1253,7 @@ async def set_script_lock( } +# 软删除脚本；元数据标记删除，历史版本可按规则继续保留。 @router.delete("/api/v1/scripts/{script_id}") async def delete_script( script_id: str, @@ -1277,6 +1299,7 @@ async def delete_script( } +# 将当前脚本内容发布为不可变版本，供调度节点和回溯下载使用。 @router.post( "/api/v1/scripts/{script_id}/versions", status_code=status.HTTP_201_CREATED, @@ -1398,6 +1421,7 @@ async def publish_version( } +# 列出某脚本已经发布的历史版本。 @router.get("/api/v1/scripts/{script_id}/versions") async def list_versions( script_id: str, @@ -1419,6 +1443,7 @@ async def list_versions( } +# 读取脚本最近一次发布的版本；未发布时返回空结果。 @router.get("/api/v1/scripts/{script_id}/latest-version") async def latest_version( script_id: str, @@ -1470,6 +1495,7 @@ async def latest_version( } +# 查询单个发布版本的元数据和关联脚本信息。 @router.get("/api/v1/versions/{versions_id}") async def get_version( versions_id: str, @@ -1486,6 +1512,7 @@ async def get_version( } +# 隐藏/删除一个发布版本；是否保留实际产物由存储删除策略决定。 @router.delete("/api/v1/versions/{versions_id}") async def delete_version( versions_id: str, @@ -1533,6 +1560,7 @@ async def delete_version( } +# 为某个版本产物生成带时效的下载地址，而非把大文件直接经 API 返回。 @router.post("/api/v1/versions/{versions_id}/download-url") async def version_download_url( versions_id: str, diff --git a/backend/src/backend/storage_api.py b/backend/src/backend/storage_api.py index 5227952..41c6d8b 100644 --- a/backend/src/backend/storage_api.py +++ b/backend/src/backend/storage_api.py @@ -1,3 +1,10 @@ +"""后端内部对象存储接口与通用辅助函数。 + +这些路由由 ``main.py`` 额外挂载到 ``/internal``，用于上传会话、对象记录和 +下载地址等内部协作。业务路由通常直接复用本模块/``services.storage`` 的函数， +而不是让浏览器直接调用这些内部接口。 +""" + from __future__ import annotations import hashlib @@ -85,9 +92,8 @@ def safe_file_name(value: str) -> str: return name -# 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. +# 预先解析的“用途 → 桶名”映射，供只读调用方快速使用。若工作区配置了专属 +# artifact_bucket，则必须调用 resolve_bucket()，让工作区级覆盖规则生效。 BUCKET_FOR_USAGE: dict[str, str] = { usage_type: actual_bucket_name(purpose) for usage_type, purpose in USAGE_TYPE_TO_PURPOSE.items() @@ -134,12 +140,9 @@ def storage_payload(item: StorageObjects) -> dict[str, Any]: } -# Internal routes are mounted from main.py via -# ``include_router(router, prefix="/internal")``. The engine, session -# factory, and object_stores live on the main app's lifespan — this -# module only owns the route definitions and the storage-helper -# utilities (``storage_payload``, ``resolve_bucket``, ``BUCKET_FOR_USAGE`` -# …) consumed by ``backend.services.storage``. +# 内部路由由 main.py 以 /internal 前缀挂载。数据库引擎、Session 工厂和对象 +# 存储实例均在应用生命周期中创建；本模块只定义路由和供 services.storage 复用的 +# 存储辅助函数（如 storage_payload、resolve_bucket、BUCKET_FOR_USAGE）。 router = APIRouter(tags=["internal-storage"]) @@ -470,6 +473,7 @@ async def restore_object( } +# 管理动作：永久清理超过保留期限或指定的回收站对象。 @router.post("/v1/admin/trash/purge") async def purge_trash_object( payload: dict[str, Any], @@ -520,6 +524,7 @@ async def purge_trash_object( return {"data": {"storage_object_id": storage_object_id, "purged": True}} +# 检查内部存储后端是否可用，供健康检查和排障使用。 @router.get("/health/storage") async def internal_health() -> dict[str, str]: return {"status": "ready", "service": "storage-api"} diff --git a/common/src/common/scheduler/trigger.py b/common/src/common/scheduler/trigger.py index f2eb132..ce14826 100644 --- a/common/src/common/scheduler/trigger.py +++ b/common/src/common/scheduler/trigger.py @@ -35,10 +35,10 @@ from common.db.models import ( from common.eventing import add_outbox_event, schedule_event_type, utcnow from common.ids import new_ulid -# A stable user_id used for cron-triggered runs. The corresponding -# ``Users`` row is seeded by the auth-bootstrap migration so any audit -# query joining on ``ScheduleRuns.triggered_by`` still resolves. -SYSTEM_CRON_USER_ID = "01HZZZZZZZZZZZZZZZZZZZZZZCR" +# Cron 运行需要一个可审计的 ``triggered_by``。这里使用基线迁移已经创建的 +# 管理员用户；该值必须是 26 个字符，才能写入 ``ScheduleRuns.triggered_by`` +# 的 ``CHAR(26)`` 字段。 +SYSTEM_CRON_USER_ID = "00000000000000000000000001" TriggerType = Literal["manual", "cron", "api"] diff --git a/frontend/app/features/schedules/SchedulePage.tsx b/frontend/app/features/schedules/SchedulePage.tsx index 2a53868..ec823a4 100644 --- a/frontend/app/features/schedules/SchedulePage.tsx +++ b/frontend/app/features/schedules/SchedulePage.tsx @@ -141,19 +141,38 @@ export default function SchedulePage({ setArtifacts([]); setSchedule(null); setRuns([]); - setRunsLoading(true); + setRunsLoading(false); void useSchedulesStore.getState().loadInitial(); }, [workspaceId, api, setSchedules, setArtifacts, setSchedule]); + // 进入调度页或切换调度方案后，主动加载该方案已经存在的运行记录。 + // 原先只有「手动运行」和运行中的轮询会更新列表，因此从其他页面返回时会 + // 一直停留在加载状态，直到产生新的运行记录。 + useEffect(() => { + const scheduleId = schedule?.schedule_id; + if (!scheduleId || !api) { + setRuns([]); + setRunsLoading(false); + return; + } + void useSchedulesStore.getState().refreshRuns(scheduleId, true); + }, [schedule?.schedule_id, api, setRuns, setRunsLoading]); + useEffect(() => { const scheduleId = schedule?.schedule_id; if (!scheduleId || !api) return; - if (!runs.some((item) => item.run_status === "queued" || item.run_status === "running")) return; + const hasActiveRun = runs.some( + (item) => item.run_status === "queued" || item.run_status === "running", + ); + const isEnabledCron = schedule.trigger_type === "cron" && schedule.enabled; + // Cron 会由后端在未来某个整分钟创建新记录。即使当前没有运行中的 + // 记录，也要持续刷新，才能让新一轮运行自动出现在右侧列表中。 + if (!hasActiveRun && !isEnabledCron) return; const timer = window.setInterval(() => { void useSchedulesStore.getState().refreshRuns(scheduleId); - }, 1500); + }, hasActiveRun ? 1500 : 3000); return () => window.clearInterval(timer); - }, [schedule?.schedule_id, runs, api]); + }, [schedule?.schedule_id, schedule?.trigger_type, schedule?.enabled, runs, api]); // Form submit wrapper for create-schedule modal — keeps the FormEvent flow out of JSX. const handleCreateScheduleSubmit = (event: FormEvent): void => { @@ -292,10 +311,15 @@ export default function SchedulePage({ }} onDrop={onCanvasDrop} onClick={(event) => { - if (event.target === event.currentTarget) { - setSelectedNodeId(null); - setSelectedEdgeId(null); - } + // 画布实际内容位于 surface/SVG 子元素中，不能只比较 currentTarget； + // 点击节点或连线时保留选择，点击其余空白位置则返回调度方案基本信息。 + const target = event.target as Element; + if ( + target.closest(".schedule-node") || + target.matches(".schedule-edge-line, .schedule-edge-hit") + ) return; + setSelectedNodeId(null); + setSelectedEdgeId(null); }} >
((set, get) => { } catch (error) { if (showLoading) await handleError(error, "运行记录加载失败"); } finally { - if (showLoading) set({ runsLoading: false }); + // 如果用户已切换到另一个调度方案，不能让旧请求结束时覆盖新方案的 + // 加载状态；新方案会由自己的请求负责关闭 loading。 + if (showLoading && get().schedule?.schedule_id === scheduleId) { + set({ runsLoading: false }); + } } }, diff --git a/schedule/src/schedule/scheduler.py b/schedule/src/schedule/scheduler.py index 5d03941..5de0167 100644 --- a/schedule/src/schedule/scheduler.py +++ b/schedule/src/schedule/scheduler.py @@ -16,7 +16,7 @@ from __future__ import annotations import asyncio from collections.abc import Awaitable, Callable -from datetime import UTC +from datetime import UTC, datetime, timedelta from zoneinfo import ZoneInfo from apscheduler.schedulers.asyncio import AsyncIOScheduler @@ -66,6 +66,12 @@ class CronScheduler: self._on_trigger = on_trigger _ACTIVE_TRIGGER = on_trigger self._sync_task: asyncio.Task[None] | None = None + # 仅在调度配置实际变化时才重置 APScheduler job。若每 5 秒都 + # reschedule，一旦恰好落在整分钟之后，就可能把本分钟的触发跳过。 + self._job_signatures: dict[str, tuple[str, str, int]] = {} + # APScheduler 的定时唤醒异常时，由 5 秒同步循环兜底。键保存的是 + # 已由兜底路径处理的“本地整分钟”，避免同一分钟重复提交。 + self._fallback_dispatched_minutes: dict[str, datetime] = {} def start(self) -> None: """Start APScheduler and spawn the periodic sync loop.""" @@ -120,8 +126,9 @@ class CronScheduler: - adds jobs for enabled cron schedules present in MySQL - removes jobs whose schedule has been disabled / soft-deleted - - updates ``Schedules.next_run_at`` from the next APScheduler tick + - updates ``Schedules.next_run_at`` from the Cron expression itself """ + due_schedule_ids: list[str] = [] async with session_scope(self.session_factory) as session: schedules = list( ( @@ -142,14 +149,16 @@ class CronScheduler: job_id = f"schedule:{item.schedule_id}" active_job_ids.add(job_id) expression = (item.cron_expression or "").strip() + max_instances = max(1, item.max_concurrency) + signature = (expression, item.timezone, max_instances) trigger = CronTrigger.from_crontab( expression, timezone=ZoneInfo(item.timezone), ) - if self.scheduler.get_job(job_id) is not None: - self.scheduler.reschedule_job(job_id, trigger=trigger) - updated_count += 1 - else: + now = datetime.now(ZoneInfo(item.timezone)) + minute = now.replace(second=0, microsecond=0) + job_changed = False + if self.scheduler.get_job(job_id) is None: self.scheduler.add_job( dispatch_persisted_cron, trigger=trigger, @@ -157,12 +166,45 @@ class CronScheduler: id=job_id, replace_existing=True, coalesce=True, - max_instances=max(1, item.max_concurrency), + max_instances=max_instances, misfire_grace_time=60, ) added_count += 1 - job = self.scheduler.get_job(job_id) - item.next_run_at = naive_utc(job.next_run_time) + job_changed = True + elif self._job_signatures.get(job_id) != signature: + # 服务重启后的首次同步也会走这里，确保持久化 job 与 + # 数据库当前配置一致；之后配置不变时保留原定时点。 + self.scheduler.reschedule_job(job_id, trigger=trigger) + self.scheduler.modify_job( + job_id, + max_instances=max_instances, + ) + updated_count += 1 + job_changed = True + self._job_signatures[job_id] = signature + # 以 CronTrigger 本身计算下次执行时间，不依赖 APScheduler 的 + # 内部唤醒状态；页面展示的「下次执行」也因此保持准确。 + item.next_run_at = naive_utc( + trigger.get_next_fire_time(None, now) + ) + + # 首次观察或刚修改表达式时，从下一个整分钟才开始兜底，符合 + # Cron 的常规语义，避免用户在本分钟中途保存后立刻多跑一次。 + if job_changed or job_id not in self._fallback_dispatched_minutes: + self._fallback_dispatched_minutes[job_id] = minute + + # 正常情况下 APScheduler 会在整分钟回调。实测其偶发漏唤醒时， + # 这里每 5 秒检查一次当前分钟是否命中表达式，并补发一次。 + due_at = trigger.get_next_fire_time( + minute - timedelta(minutes=1), + minute, + ) + if ( + due_at == minute + and self._fallback_dispatched_minutes.get(job_id) != minute + ): + self._fallback_dispatched_minutes[job_id] = minute + due_schedule_ids.append(item.schedule_id) removed_count = 0 for job in self.scheduler.get_jobs(): if ( @@ -170,6 +212,8 @@ class CronScheduler: and job.id not in active_job_ids ): self.scheduler.remove_job(job.id) + self._job_signatures.pop(job.id, None) + self._fallback_dispatched_minutes.pop(job.id, None) removed_count += 1 if added_count or updated_count or removed_count: logger.info( @@ -178,6 +222,11 @@ class CronScheduler: updated_count, removed_count, ) + # 在数据库同步事务提交后再创建运行记录，避免两个会话同时读取调度方案 + # 时发生不必要的锁等待。重复回调由运行记录的幂等键自动去重。 + for schedule_id in due_schedule_ids: + logger.debug("cron fallback dispatch: schedule={}", schedule_id[-12:]) + await self._on_trigger(schedule_id) __all__ = ["CronScheduler"]