Merge branch 'master' of http://192.168.0.200/knowledge-base-bj/deerflow-server into dhk-1

offline-backend-20260512/backend/app/gateway/app.py

@@ -378,7 +378,7 @@ def _release_singleton_lock(handle) -> None:
             pass
 
 
-async def _curator_scheduler_loop() -> None:
+async def _curator_scheduler_loop() -> None:
     """常驻后台任务:每隔 _CURATOR_CHECK_INTERVAL_SECONDS 检查一次需不需要进化。
 
     多 worker 部署下用文件锁做 leader 选举 —— 只有抢到锁的那个 worker 进程真正
@@ -404,11 +404,248 @@ async def _curator_scheduler_loop() -> None:
             except Exception:
                 logger.exception("Curator scheduler tick failed (non-fatal)")
             await asyncio.sleep(_CURATOR_CHECK_INTERVAL_SECONDS)
-    finally:
-        _release_singleton_lock(lock_handle)
-
-
-def _warn_if_multi_worker() -> None:
+    finally:
+        _release_singleton_lock(lock_handle)
+
+
+def _env_bool(name: str, default: bool) -> bool:
+    raw = os.getenv(name)
+    if raw is None:
+        return default
+    return raw.strip().lower() not in {"0", "false", "no", "off"}
+
+
+def _env_int(name: str, default: int, *, alias: str | None = None, minimum: int = 1) -> int:
+    raw = os.getenv(name)
+    if raw is None and alias:
+        raw = os.getenv(alias)
+    if raw is None:
+        raw = str(default)
+    try:
+        value = int(raw)
+    except ValueError:
+        return default
+    return max(minimum, value)
+
+
+def _env_float(name: str, default: float, *, alias: str | None = None, minimum: float = 0.0) -> float:
+    raw = os.getenv(name)
+    if raw is None and alias:
+        raw = os.getenv(alias)
+    if raw is None:
+        raw = str(default)
+    try:
+        value = float(raw)
+    except ValueError:
+        return default
+    return max(minimum, value)
+
+
+_LEADERBOARD_SNAPSHOT_BACKGROUND_ENABLED = _env_bool("ADMIN_LEADERBOARD_BACKGROUND_ENABLED", True)
+_LEADERBOARD_SNAPSHOT_PREWARM_DAYS = _env_int(
+    "ADMIN_LEADERBOARD_BACKGROUND_PREWARM_DAYS",
+    30,
+)
+_LEADERBOARD_SNAPSHOT_BATCH_SIZE = _env_int(
+    "ADMIN_LEADERBOARD_BACKGROUND_BATCH_SIZE",
+    1,
+)
+_LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS = _env_float(
+    "ADMIN_LEADERBOARD_BACKGROUND_INTERVAL_SECONDS",
+    300.0,
+)
+_LEADERBOARD_SNAPSHOT_SLEEP_SECONDS = _env_float(
+    "ADMIN_LEADERBOARD_BACKGROUND_SLEEP_SECONDS",
+    2.0,
+)
+_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS = _env_int(
+    "ADMIN_LEADERBOARD_BACKGROUND_RUNNING_TTL_SECONDS",
+    1800,
+)
+_LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS = _env_float(
+    "ADMIN_LEADERBOARD_BACKGROUND_STARTUP_DELAY_SECONDS",
+    60.0,
+)
+
+
+def _recent_leaderboard_stat_dates(days: int) -> list[str]:
+    from datetime import datetime, timedelta
+
+    from deerflow.persistence.types import BEIJING_TZ
+
+    today = datetime.now(BEIJING_TZ).date()
+    start = today - timedelta(days=max(1, days) - 1)
+    return [(start + timedelta(days=i)).isoformat() for i in range((today - start).days + 1)]
+
+
+async def _leaderboard_snapshot_due_dates(
+    session_factory,
+    *,
+    settings_hash: str,
+    limit: int,
+    running_ttl_seconds: int,
+) -> list[str]:
+    from datetime import UTC, datetime, timedelta
+
+    from sqlalchemy import and_, or_, select
+
+    from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
+
+    cutoff = datetime.now(UTC) - timedelta(seconds=running_ttl_seconds)
+    async with session_factory() as session:
+        rows = (
+            await session.execute(
+                select(AdminLeaderboardDailyStatRow.stat_date)
+                .where(
+                    AdminLeaderboardDailyStatRow.settings_hash == settings_hash,
+                    or_(
+                        AdminLeaderboardDailyStatRow.status.in_(["queued", "error"]),
+                        and_(
+                            AdminLeaderboardDailyStatRow.status == "running",
+                            AdminLeaderboardDailyStatRow.updated_at < cutoff,
+                        ),
+                    ),
+                )
+                .order_by(AdminLeaderboardDailyStatRow.stat_date.desc())
+                .limit(limit)
+            )
+        ).all()
+    return [str(row[0]) for row in rows]
+
+
+async def _claim_leaderboard_snapshot(
+    session_factory,
+    *,
+    stat_date: str,
+    settings_hash: str,
+    running_ttl_seconds: int,
+) -> bool:
+    from datetime import UTC, datetime
+
+    from app.gateway.routers.admin_users import _snapshot_is_fresh
+    from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
+    from deerflow.persistence.types import BEIJING_TZ
+
+    async with session_factory() as session:
+        row = await session.get(
+            AdminLeaderboardDailyStatRow,
+            {"stat_date": stat_date, "settings_hash": settings_hash},
+        )
+        if row is None or _snapshot_is_fresh(row, stat_date):
+            return False
+        if row.status == "running":
+            updated_at = row.updated_at
+            if isinstance(updated_at, datetime):
+                if updated_at.tzinfo is None:
+                    updated_at = updated_at.replace(tzinfo=BEIJING_TZ)
+                age = (datetime.now(BEIJING_TZ) - updated_at.astimezone(BEIJING_TZ)).total_seconds()
+                if age < running_ttl_seconds:
+                    return False
+        row.status = "running"
+        row.message = "排行榜后台统计任务已认领。"
+        row.error = None
+        row.updated_at = datetime.now(UTC)
+        await session.commit()
+    return True
+
+
+async def _run_leaderboard_snapshot_tick() -> tuple[int, int]:
+    from app.gateway.routers.admin_users import (
+        _daily_job_key,
+        _queue_daily_leaderboard_snapshots,
+        _refresh_daily_leaderboard_snapshot,
+        _resolve_leaderboard_snapshot_scope,
+    )
+    from deerflow.config.system_settings import load_system_settings
+    from deerflow.persistence.engine import get_session_factory
+
+    session_factory = get_session_factory()
+    if session_factory is None:
+        return 0, 0
+
+    settings = load_system_settings().leaderboard
+    excluded, include_scheduled, include_admins, include_failed, settings_hash = _resolve_leaderboard_snapshot_scope(settings)
+    stat_dates = _recent_leaderboard_stat_dates(_LEADERBOARD_SNAPSHOT_PREWARM_DAYS)
+    queued_dates = await _queue_daily_leaderboard_snapshots(
+        session_factory,
+        stat_dates,
+        settings_hash,
+        message="排行榜后台统计任务定期预热/补算队列。",
+    )
+
+    due_dates = await _leaderboard_snapshot_due_dates(
+        session_factory,
+        settings_hash=settings_hash,
+        limit=_LEADERBOARD_SNAPSHOT_BATCH_SIZE,
+        running_ttl_seconds=_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS,
+    )
+    processed = 0
+    for stat_date in due_dates:
+        claimed = await _claim_leaderboard_snapshot(
+            session_factory,
+            stat_date=stat_date,
+            settings_hash=settings_hash,
+            running_ttl_seconds=_LEADERBOARD_SNAPSHOT_RUNNING_TTL_SECONDS,
+        )
+        if not claimed:
+            continue
+        logger.info("Leaderboard snapshot scheduler: building date=%s hash=%s", stat_date, settings_hash[:10])
+        await _refresh_daily_leaderboard_snapshot(
+            _daily_job_key(stat_date, settings_hash),
+            session_factory=session_factory,
+            settings=settings,
+            stat_date=stat_date,
+            settings_hash=settings_hash,
+            excluded=excluded,
+            include_scheduled=include_scheduled,
+            include_admins=include_admins,
+            include_failed=include_failed,
+        )
+        processed += 1
+        if _LEADERBOARD_SNAPSHOT_SLEEP_SECONDS > 0 and processed < len(due_dates):
+            await asyncio.sleep(_LEADERBOARD_SNAPSHOT_SLEEP_SECONDS)
+    return processed, len(queued_dates)
+
+
+async def _leaderboard_snapshot_scheduler_loop() -> None:
+    """Background task inside the existing gateway process for daily snapshots."""
+    if not _LEADERBOARD_SNAPSHOT_BACKGROUND_ENABLED:
+        logger.info("Leaderboard snapshot scheduler disabled by ADMIN_LEADERBOARD_BACKGROUND_ENABLED")
+        return
+
+    from deerflow.config.runtime_paths import runtime_home
+
+    lock_handle = _acquire_singleton_lock(runtime_home() / ".leaderboard_snapshot_scheduler.lock")
+    if lock_handle is None:
+        logger.info("Leaderboard snapshot scheduler: another gateway worker holds the lock; this worker skips")
+        return
+
+    logger.info(
+        "Leaderboard snapshot scheduler started: prewarm_days=%d batch_size=%d interval=%.1fs",
+        _LEADERBOARD_SNAPSHOT_PREWARM_DAYS,
+        _LEADERBOARD_SNAPSHOT_BATCH_SIZE,
+        _LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS,
+    )
+    try:
+        if _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS > 0:
+            await asyncio.sleep(_LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS)
+        while True:
+            try:
+                processed, queued = await _run_leaderboard_snapshot_tick()
+                if processed or queued:
+                    logger.info(
+                        "Leaderboard snapshot scheduler tick: queued=%d processed=%d",
+                        queued,
+                        processed,
+                    )
+            except Exception:
+                logger.exception("Leaderboard snapshot scheduler tick failed (non-fatal)")
+            await asyncio.sleep(_LEADERBOARD_SNAPSHOT_INTERVAL_SECONDS)
+    finally:
+        _release_singleton_lock(lock_handle)
+
+
+def _warn_if_multi_worker() -> None:
     """检测 uvicorn / gunicorn 多 worker 部署并打 ERROR 日志。
 
     主 agent / canvas / AI 写作流式响应在 Gateway 进程内通过 ``StreamBridge``
@@ -562,16 +799,21 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         # 默认每天凌晨 3 点扫一次,删 retention_days(默认 7 天)以前的会话,联动
         # 清理 MySQL 业务表 + SQLite checkpointer 里的 thread 状态。多 worker
         # 下用文件锁选 leader,只有抢到锁的 worker 真正跑清理。
-        try:
-            from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop
-
-            app.state.ai_writing_cleanup_task = asyncio.create_task(cleanup_scheduler_loop(app))
-        except Exception:
-            logger.exception("AI writing cleanup scheduler failed to start (non-fatal)")
-
-        # Start IM channel service if any channels are configured
-        try:
-            from app.channels.service import start_channel_service
+        try:
+            from app.gateway.ai_writing_cleanup import cleanup_scheduler_loop
+
+            app.state.ai_writing_cleanup_task = asyncio.create_task(cleanup_scheduler_loop(app))
+        except Exception:
+            logger.exception("AI writing cleanup scheduler failed to start (non-fatal)")
+
+        try:
+            app.state.leaderboard_snapshot_scheduler_task = asyncio.create_task(_leaderboard_snapshot_scheduler_loop())
+        except Exception:
+            logger.exception("Leaderboard snapshot scheduler failed to start (non-fatal)")
+
+        # Start IM channel service if any channels are configured
+        try:
+            from app.channels.service import start_channel_service
 
             channel_service = await start_channel_service(app.state.config)
             logger.info("Channel service started: %s", channel_service.get_status())
@@ -615,7 +857,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
             except asyncio.CancelledError:
                 pass
             except Exception:
-                logger.exception("AI writing cleanup scheduler shutdown error")
+                logger.exception("AI writing cleanup scheduler shutdown error")
+
+        leaderboard_snapshot_task = getattr(app.state, "leaderboard_snapshot_scheduler_task", None)
+        if leaderboard_snapshot_task is not None:
+            leaderboard_snapshot_task.cancel()
+            try:
+                await leaderboard_snapshot_task
+            except asyncio.CancelledError:
+                pass
+            except Exception:
+                logger.exception("Leaderboard snapshot scheduler shutdown error")
 
         # 阶段 7 删冗余：AI 写作 graph 不再自管 checkpointer / in-flight task；
         # thread 状态由 LangGraph Server 主 checkpointer 接管，lifespan 出口由

offline-backend-20260512/backend/app/gateway/routers/tags.py

@@ -1,9 +1,9 @@
 """Tag management API.
 
 Tags are split into three types (``agent`` / ``skill`` / ``scheduled_task``).
-A tag may only be applied to a resource of its own type. Tag CRUD and tag
-assignment are admin-only; listing tags and reading a resource's tags is open
-to everyone (the frontend needs them to render chips and search filters).
+A tag may only be applied to a resource of its own type. Tag CRUD, listing,
+and assignment are available to ordinary users so teams can organize resources
+without asking an administrator to tag every item.
 """
 
 from __future__ import annotations
@@ -11,10 +11,10 @@ from __future__ import annotations
 import logging
 import uuid
 
-from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from fastapi import APIRouter, Depends, HTTPException, Query
 from pydantic import BaseModel, Field
 
-from app.gateway.deps import get_optional_user_from_request, get_tag_store
+from app.gateway.deps import get_tag_store
 from deerflow.persistence.tags import TAG_TYPES, TagStore
 
 logger = logging.getLogger(__name__)
@@ -22,18 +22,6 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/tags", tags=["tags"])
 
 
-# --------------------------------------------------------------------------- auth
-
-
-async def _require_admin(request: Request) -> None:
-    """Reject non-admins. Lets the call through when auth is disabled."""
-    user = await get_optional_user_from_request(request)
-    if user is None:
-        return
-    if getattr(user, "system_role", None) != "admin":
-        raise HTTPException(status_code=403, detail="Tag management is admin-only")
-
-
 def _validate_tag_type(tag_type: str) -> str:
     if tag_type not in TAG_TYPES:
         raise HTTPException(status_code=422, detail=f"Invalid tag type '{tag_type}'. Allowed: {', '.join(TAG_TYPES)}")
@@ -96,10 +84,8 @@ async def list_tags(
 @router.post("", response_model=TagResponse, status_code=201, summary="Create Tag")
 async def create_tag(
     body: TagCreateRequest,
-    request: Request,
     store: TagStore = Depends(get_tag_store),
 ) -> TagResponse:
-    await _require_admin(request)
     _validate_tag_type(body.tag_type)
     try:
         record = await store.create_tag(
@@ -125,10 +111,8 @@ async def create_tag(
 async def update_tag(
     tag_id: str,
     body: TagUpdateRequest,
-    request: Request,
     store: TagStore = Depends(get_tag_store),
 ) -> TagResponse:
-    await _require_admin(request)
     update_data = body.model_dump(exclude_unset=True)
     if "name" in update_data and update_data["name"] is not None:
         update_data["name"] = update_data["name"].strip()
@@ -149,10 +133,8 @@ async def update_tag(
 @router.delete("/{tag_id}", status_code=204, summary="Delete Tag")
 async def delete_tag(
     tag_id: str,
-    request: Request,
     store: TagStore = Depends(get_tag_store),
 ) -> None:
-    await _require_admin(request)
     try:
         deleted = await store.delete_tag(tag_id)
         if not deleted:
@@ -195,10 +177,8 @@ async def set_resource_tags(
     target_type: str,
     target_id: str,
     body: TagAssignmentRequest,
-    request: Request,
     store: TagStore = Depends(get_tag_store),
 ) -> TagListResponse:
-    await _require_admin(request)
     _validate_tag_type(target_type)
     try:
         records = await store.set_target_tags(target_type, target_id, body.tag_ids)

offline-backend-20260512/backend/config.yaml

@@ -90,6 +90,16 @@ models:
     temperature: 1.0
     supports_vision: true
     supports_thinking: true
+  - name: Qwen/Qwen3.6-27B
+    display_name: Qwen/Qwen3.6-27B
+    use: langchain_openai:ChatOpenAI
+    model: Qwen/Qwen3.6-27B
+    api_key:  sk-ighylafootabtuthwueirmoycccyqtdgbifyhozduvolkdyt
+    base_url: https://api.siliconflow.cn/v1/
+    max_tokens: 4096
+    temperature: 1.0
+    supports_vision: true
+    supports_thinking: true
 # 识图(图片识别)配置。
 # - model_name:指定一个 models[] 中的模型名,作为“默认识图大模型”。view_image 工具会把
 #   图片交给该模型识别,并把识别出的文字结果返回给主对话模型 —— 这样即使主对话模型不支持
@@ -213,21 +223,15 @@ auth_login:
 
 database:
   backend: sqlite
-  sqlite_dir: .deer-flow/data
-  pool_size: 10
+  sqlite_dir: /app/backend/.deer-flow/data
+  pool_size: 20          # 基础连接池(postgres/mysql 生效;sqlite 忽略)
+  max_overflow: 20       # 高峰期可临时超出 pool_size 的额外连接数,真实上限 = pool_size + max_overflow
+  pool_timeout: 30       # 池满时等待空闲连接的秒数,超时报错
   echo_sql: false
 
 checkpointer:
-  type: postgres
-  # true: 新 checkpoint 写入 Postgres；false: 保留 PG 配置但继续使用 SQLite。
-  # 也可临时用环境变量 DEER_FLOW_USE_PG_CHECKPOINTER=0 覆盖。
-  use_postgres: true
-  connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow
-  sqlite_connection_string: .deer-flow/data/deerflow.db
-  dual_read_enabled: true
-  dual_read_sqlite_paths:
-    - .deer-flow/data/deerflow.db
-    - ../data/.deer-flow/data/checkpoints.db
+  type: sqlite
+  connection_string: /app/backend/.deer-flow/data/deerflow.db
 
 # 记忆子系统:跨会话持久化用户画像与 agent 私有记忆。
 # 三层覆盖优先级(高->低):

offline-backend-20260512/backend/packages/harness/deerflow/config/database_config.py

@@ -11,13 +11,13 @@ blocking, making a unified file safe for both workloads.  Writers
 that contend for the lock wait via the default 5-second sqlite3
 busy timeout rather than failing immediately.
 
-Postgres mode: both use the same database URL but maintain independent
-connection pools with different lifecycles.
-
-MySQL mode: application ORM data uses MySQL. LangGraph's checkpointer does
-not have a built-in MySQL backend in this project, so deployments that choose
-MySQL for business tables should configure the legacy ``checkpointer:``
-section separately (usually sqlite) to preserve conversation checkpoints.
+Postgres mode: both use the same database URL but maintain independent
+connection pools with different lifecycles.
+
+MySQL mode: application ORM data uses MySQL. LangGraph's checkpointer does
+not have a built-in MySQL backend in this project, so deployments that choose
+MySQL for business tables should configure the legacy ``checkpointer:``
+section separately (usually sqlite) to preserve conversation checkpoints.
 
 Memory mode: checkpointer uses MemorySaver, app uses in-memory stores.
 No database is initialized.
@@ -26,12 +26,12 @@ Sensitive values (postgres_url) should use $VAR syntax in config.yaml
 to reference environment variables from .env:
 
     database:
-      backend: postgres
-      postgres_url: $DATABASE_URL
-
-The $VAR resolution is handled by AppConfig.resolve_env_variables()
-before this config is instantiated -- DatabaseConfig itself does not
-need to do any environment variable processing.
+      backend: postgres
+      postgres_url: $DATABASE_URL
+
+The $VAR resolution is handled by AppConfig.resolve_env_variables()
+before this config is instantiated -- DatabaseConfig itself does not
+need to do any environment variable processing.
 """
 
 from __future__ import annotations
@@ -42,40 +42,54 @@ from typing import Literal
 from pydantic import BaseModel, Field
 
 
-class DatabaseConfig(BaseModel):
-    backend: Literal["memory", "sqlite", "postgres", "mysql"] = Field(
-        default="memory",
-        description=("Storage backend for application data. 'memory' for development (no persistence across restarts), 'sqlite' for single-node deployment, 'postgres' for production multi-node deployment, 'mysql' for MySQL-backed business tables."),
-    )
+class DatabaseConfig(BaseModel):
+    backend: Literal["memory", "sqlite", "postgres", "mysql"] = Field(
+        default="memory",
+        description=("Storage backend for application data. 'memory' for development (no persistence across restarts), 'sqlite' for single-node deployment, 'postgres' for production multi-node deployment, 'mysql' for MySQL-backed business tables."),
+    )
     sqlite_dir: str = Field(
         default=".deer-flow/data",
         description=("Directory for the SQLite database file. Both checkpointer and application data share {sqlite_dir}/deerflow.db."),
     )
-    postgres_url: str = Field(
-        default="",
-        description=(
-            "PostgreSQL connection URL, shared by checkpointer and app. "
+    postgres_url: str = Field(
+        default="",
+        description=(
+            "PostgreSQL connection URL, shared by checkpointer and app. "
             "Use $DATABASE_URL in config.yaml to reference .env. "
             "Example: postgresql://user:pass@host:5432/deerflow "
             "(the +asyncpg driver suffix is added automatically where needed)."
-        ),
-    )
-    mysql_url: str = Field(
-        default="",
-        description=(
-            "MySQL connection URL for the application ORM. "
-            "Use $MYSQL_DATABASE_URL in config.yaml to reference .env. "
-            "Example: mysql+asyncmy://user:pass@host:3306/deerflow?charset=utf8mb4. "
-            "If mysql:// is provided, the +asyncmy driver suffix is added automatically."
-        ),
-    )
+        ),
+    )
+    mysql_url: str = Field(
+        default="",
+        description=(
+            "MySQL connection URL for the application ORM. "
+            "Use $MYSQL_DATABASE_URL in config.yaml to reference .env. "
+            "Example: mysql+asyncmy://user:pass@host:3306/deerflow?charset=utf8mb4. "
+            "If mysql:// is provided, the +asyncmy driver suffix is added automatically."
+        ),
+    )
     echo_sql: bool = Field(
         default=False,
         description="Echo all SQL statements to log (debug only).",
     )
     pool_size: int = Field(
         default=5,
-        description="Connection pool size for the app ORM engine (postgres/mysql only).",
+        description="Connection pool size for the app ORM engine (postgres/mysql only).",
+    )
+    max_overflow: int = Field(
+        default=10,
+        description=(
+            "Extra connections allowed beyond pool_size under load (postgres/mysql only). "
+            "The real connection ceiling is pool_size + max_overflow. SQLAlchemy default is 10."
+        ),
+    )
+    pool_timeout: int = Field(
+        default=30,
+        description=(
+            "Seconds a caller waits for a free pooled connection before erroring (postgres/mysql only). "
+            "SQLAlchemy default is 30."
+        ),
     )
 
     # -- Derived helpers (not user-configured) --
@@ -108,16 +122,16 @@ class DatabaseConfig(BaseModel):
         """SQLAlchemy async URL for the application ORM engine."""
         if self.backend == "sqlite":
             return f"sqlite+aiosqlite:///{self.sqlite_path}"
-        if self.backend == "postgres":
-            url = self.postgres_url
-            if url.startswith("postgresql://"):
-                url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
-            return url
-        if self.backend == "mysql":
-            url = self.mysql_url
-            if not url:
-                raise ValueError("database.mysql_url is required for backend='mysql'")
-            if url.startswith("mysql://"):
-                url = url.replace("mysql://", "mysql+asyncmy://", 1)
-            return url
-        raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}")
+        if self.backend == "postgres":
+            url = self.postgres_url
+            if url.startswith("postgresql://"):
+                url = url.replace("postgresql://", "postgresql+asyncpg://", 1)
+            return url
+        if self.backend == "mysql":
+            url = self.mysql_url
+            if not url:
+                raise ValueError("database.mysql_url is required for backend='mysql'")
+            if url.startswith("mysql://"):
+                url = url.replace("mysql://", "mysql+asyncmy://", 1)
+            return url
+        raise ValueError(f"No SQLAlchemy URL for backend={self.backend!r}")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/__init__.py

@@ -0,0 +1,5 @@
+"""Persistence models for admin analytics snapshots."""
+
+from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
+
+__all__ = ["AdminLeaderboardDailyStatRow"]

offline-backend-20260512/backend/packages/harness/deerflow/persistence/admin_stats/model.py

@@ -0,0 +1,31 @@
+"""ORM models for admin analytics snapshots."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import Index, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime, PortableJSON
+
+
+class AdminLeaderboardDailyStatRow(Base):
+    """One precomputed leaderboard snapshot for a single local calendar day."""
+
+    __tablename__ = "admin_leaderboard_daily_stats"
+    __table_args__ = (
+        Index("ix_admin_leaderboard_daily_stats_status", "status"),
+        Index("ix_admin_leaderboard_daily_stats_updated_at", "updated_at"),
+    )
+
+    stat_date: Mapped[str] = mapped_column(String(10), primary_key=True)
+    settings_hash: Mapped[str] = mapped_column(String(64), primary_key=True)
+    status: Mapped[str] = mapped_column(String(16), default="ready")
+    payload_json: Mapped[dict] = mapped_column(PortableJSON(), default=dict)
+    message: Mapped[str | None] = mapped_column(Text)
+    error: Mapped[str | None] = mapped_column(Text)
+    generated_at: Mapped[datetime | None] = mapped_column(BeijingDateTime())
+    created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
+    updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))

offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py

@@ -11,16 +11,16 @@ None and fall back to in-memory implementations.
 from __future__ import annotations
 
 import asyncio
-import json
-import logging
-import os
-import sys
-from importlib.util import find_spec
-from pathlib import Path
-
-from sqlalchemy import text
-from sqlalchemy.engine import make_url
-from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
+import json
+import logging
+import os
+import sys
+from importlib.util import find_spec
+from pathlib import Path
+
+from sqlalchemy import text
+from sqlalchemy.engine import make_url
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine
 
 
 def _json_serializer(obj: object) -> str:
@@ -133,7 +133,7 @@ async def _run_alembic_upgrade(url: str) -> None:
         logger.exception("Auto-applying Alembic migrations failed; continuing with create_all schema")
 
 
-async def _auto_create_postgres_db(url: str) -> None:
+async def _auto_create_postgres_db(url: str) -> None:
     """Connect to the ``postgres`` maintenance DB and CREATE DATABASE.
 
     The target database name is extracted from *url*.  The connection is
@@ -156,55 +156,57 @@ async def _auto_create_postgres_db(url: str) -> None:
         async with maint_engine.connect() as conn:
             await conn.execute(text(f'CREATE DATABASE "{db_name}"'))
         logger.info("Auto-created PostgreSQL database: %s", db_name)
-    finally:
-        await maint_engine.dispose()
-
-
-async def _auto_create_mysql_db(url: str) -> None:
-    """Create the target MySQL database when it is missing.
-
-    ``create_all`` can create tables, but MySQL requires the database/schema to
-    exist before the application URL can connect. We connect to the built-in
-    ``mysql`` maintenance database on the same server and issue a conservative
-    ``CREATE DATABASE IF NOT EXISTS``.
-    """
-
-    parsed = make_url(url)
-    db_name = parsed.database
-    if not db_name:
-        raise ValueError("Cannot auto-create MySQL database: no database name in URL")
-
-    maint_url = parsed.set(database="mysql")
-    maint_engine = create_async_engine(
-        maint_url.render_as_string(hide_password=False),
-        isolation_level="AUTOCOMMIT",
-        pool_pre_ping=True,
-    )
-    quoted = "`" + db_name.replace("`", "``") + "`"
-    try:
-        async with maint_engine.connect() as conn:
-            await conn.execute(
-                text(
-                    f"CREATE DATABASE IF NOT EXISTS {quoted} "
-                    "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
-                )
-            )
-        logger.info("Auto-created MySQL database if missing: %s", db_name)
-    finally:
-        await maint_engine.dispose()
-
-
-def _is_mysql_unknown_database_error(exc: Exception) -> bool:
-    message = str(exc).lower()
-    return "unknown database" in message or "1049" in message
-
-
-async def init_engine(
+    finally:
+        await maint_engine.dispose()
+
+
+async def _auto_create_mysql_db(url: str) -> None:
+    """Create the target MySQL database when it is missing.
+
+    ``create_all`` can create tables, but MySQL requires the database/schema to
+    exist before the application URL can connect. We connect to the built-in
+    ``mysql`` maintenance database on the same server and issue a conservative
+    ``CREATE DATABASE IF NOT EXISTS``.
+    """
+
+    parsed = make_url(url)
+    db_name = parsed.database
+    if not db_name:
+        raise ValueError("Cannot auto-create MySQL database: no database name in URL")
+
+    maint_url = parsed.set(database="mysql")
+    maint_engine = create_async_engine(
+        maint_url.render_as_string(hide_password=False),
+        isolation_level="AUTOCOMMIT",
+        pool_pre_ping=True,
+    )
+    quoted = "`" + db_name.replace("`", "``") + "`"
+    try:
+        async with maint_engine.connect() as conn:
+            await conn.execute(
+                text(
+                    f"CREATE DATABASE IF NOT EXISTS {quoted} "
+                    "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
+                )
+            )
+        logger.info("Auto-created MySQL database if missing: %s", db_name)
+    finally:
+        await maint_engine.dispose()
+
+
+def _is_mysql_unknown_database_error(exc: Exception) -> bool:
+    message = str(exc).lower()
+    return "unknown database" in message or "1049" in message
+
+
+async def init_engine(
     backend: str,
     *,
     url: str = "",
     echo: bool = False,
     pool_size: int = 5,
+    max_overflow: int = 10,
+    pool_timeout: int = 30,
     sqlite_dir: str = "",
 ) -> None:
     """Create the async engine and session factory, then auto-create tables.
@@ -213,7 +215,9 @@ async def init_engine(
         backend: "memory", "sqlite", "postgres", or "mysql".
         url: SQLAlchemy async URL (for sqlite/postgres/mysql).
         echo: Echo SQL to log.
-        pool_size: Postgres connection pool size.
+        pool_size: Base connection pool size (postgres/mysql).
+        max_overflow: Extra connections allowed beyond pool_size under load (postgres/mysql).
+        pool_timeout: Seconds to wait for a free pooled connection before erroring (postgres/mysql).
         sqlite_dir: Directory to create for SQLite (ensured to exist).
     """
     global _engine, _session_factory
@@ -271,6 +275,8 @@ async def init_engine(
             url,
             echo=echo,
             pool_size=pool_size,
+            max_overflow=max_overflow,
+            pool_timeout=pool_timeout,
             pool_pre_ping=True,
             json_serializer=_json_serializer,
         )
@@ -279,6 +285,8 @@ async def init_engine(
             url,
             echo=echo,
             pool_size=pool_size,
+            max_overflow=max_overflow,
+            pool_timeout=pool_timeout,
             pool_pre_ping=True,
             pool_recycle=1800,
             json_serializer=_json_serializer,
@@ -302,37 +310,39 @@ async def init_engine(
         # This is expected during initial scaffolding or minimal installs.
         logger.debug("deerflow.persistence.models not found; skipping auto-create tables")
 
-    try:
-        async with _engine.begin() as conn:
-            await conn.run_sync(Base.metadata.create_all)
-    except Exception as exc:
-        if backend == "postgres" and "does not exist" in str(exc):
+    try:
+        async with _engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+    except Exception as exc:
+        if backend == "postgres" and "does not exist" in str(exc):
             # Database not yet created — attempt to auto-create it, then retry.
             await _auto_create_postgres_db(url)
             # Rebuild engine against the now-existing database
             await _engine.dispose()
-            _engine = create_async_engine(url, echo=echo, pool_size=pool_size, pool_pre_ping=True, json_serializer=_json_serializer)
-            _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
-            async with _engine.begin() as conn:
-                await conn.run_sync(Base.metadata.create_all)
-        elif backend == "mysql" and _is_mysql_unknown_database_error(exc):
-            # Database not yet created. Create it, rebuild the app engine
-            # against the target DB, then let SQLAlchemy create all tables.
-            await _auto_create_mysql_db(url)
-            await _engine.dispose()
-            _engine = create_async_engine(
-                url,
-                echo=echo,
-                pool_size=pool_size,
-                pool_pre_ping=True,
-                pool_recycle=1800,
-                json_serializer=_json_serializer,
-            )
-            _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
-            async with _engine.begin() as conn:
-                await conn.run_sync(Base.metadata.create_all)
-        else:
-            raise
+            _engine = create_async_engine(url, echo=echo, pool_size=pool_size, max_overflow=max_overflow, pool_timeout=pool_timeout, pool_pre_ping=True, json_serializer=_json_serializer)
+            _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
+            async with _engine.begin() as conn:
+                await conn.run_sync(Base.metadata.create_all)
+        elif backend == "mysql" and _is_mysql_unknown_database_error(exc):
+            # Database not yet created. Create it, rebuild the app engine
+            # against the target DB, then let SQLAlchemy create all tables.
+            await _auto_create_mysql_db(url)
+            await _engine.dispose()
+            _engine = create_async_engine(
+                url,
+                echo=echo,
+                pool_size=pool_size,
+                max_overflow=max_overflow,
+                pool_timeout=pool_timeout,
+                pool_pre_ping=True,
+                pool_recycle=1800,
+                json_serializer=_json_serializer,
+            )
+            _session_factory = async_sessionmaker(_engine, expire_on_commit=False)
+            async with _engine.begin() as conn:
+                await conn.run_sync(Base.metadata.create_all)
+        else:
+            raise
 
     # Apply in-place schema fixes (column widenings, etc.) that ``create_all``
     # cannot do on its own. Failures don't abort startup — see the helper.
@@ -351,6 +361,8 @@ async def init_engine_from_config(config) -> None:
         url=config.app_sqlalchemy_url,
         echo=config.echo_sql,
         pool_size=config.pool_size,
+        max_overflow=getattr(config, "max_overflow", 10),
+        pool_timeout=getattr(config, "pool_timeout", 30),
         sqlite_dir=config.sqlite_dir if config.backend == "sqlite" else "",
     )
 

offline-backend-20260512/backend/packages/harness/deerflow/persistence/llm_metrics/model.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from sqlalchemy import Float, Integer, String, Text
+from sqlalchemy import Float, Index, Integer, String, Text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from deerflow.persistence.base import Base
@@ -15,6 +15,10 @@ class LlmCallMetricRow(Base):
     """One row per completed (or failed) LLM provider invocation."""
 
     __tablename__ = "llm_call_metrics"
+    __table_args__ = (
+        Index("ix_llm_call_metrics_created_user", "created_at", "user_id"),
+        Index("ix_llm_call_metrics_run_created", "run_id", "created_at"),
+    )
 
     id: Mapped[str] = mapped_column(String(64), primary_key=True)
     # Wall-clock time the LLM call ended. Indexed because admin views always

offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260608_01_admin_leaderboard_daily_stats.py

@@ -0,0 +1,85 @@
+"""Create admin leaderboard daily snapshots and analytics indexes.
+
+Revision ID: 20260608_01
+Revises: 20260607_02
+Create Date: 2026-06-08
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy import inspect
+
+from deerflow.persistence.types import PortableJSON
+
+
+revision = "20260608_01"
+down_revision = "20260607_02"
+branch_labels = None
+depends_on = None
+
+
+def _has_table(table_name: str) -> bool:
+    inspector = inspect(op.get_bind())
+    return table_name in inspector.get_table_names()
+
+
+def _has_index(table_name: str, index_name: str) -> bool:
+    if not _has_table(table_name):
+        return False
+    inspector = inspect(op.get_bind())
+    return any(idx["name"] == index_name for idx in inspector.get_indexes(table_name))
+
+
+def _create_index(table_name: str, index_name: str, columns: list[str]) -> None:
+    if _has_table(table_name) and not _has_index(table_name, index_name):
+        op.create_index(index_name, table_name, columns)
+
+
+def _drop_index(table_name: str, index_name: str) -> None:
+    if _has_table(table_name) and _has_index(table_name, index_name):
+        op.drop_index(index_name, table_name=table_name)
+
+
+def upgrade() -> None:
+    if not _has_table("admin_leaderboard_daily_stats"):
+        op.create_table(
+            "admin_leaderboard_daily_stats",
+            sa.Column("stat_date", sa.String(10), nullable=False),
+            sa.Column("settings_hash", sa.String(64), nullable=False),
+            sa.Column("status", sa.String(16), nullable=False, server_default="ready"),
+            sa.Column("payload_json", PortableJSON(), nullable=True),
+            sa.Column("message", sa.Text(), nullable=True),
+            sa.Column("error", sa.Text(), nullable=True),
+            sa.Column("generated_at", sa.DateTime(timezone=True), nullable=True),
+            sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+            sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+            sa.PrimaryKeyConstraint("stat_date", "settings_hash", name="pk_admin_leaderboard_daily_stats"),
+        )
+        op.create_index("ix_admin_leaderboard_daily_stats_status", "admin_leaderboard_daily_stats", ["status"])
+        op.create_index("ix_admin_leaderboard_daily_stats_updated_at", "admin_leaderboard_daily_stats", ["updated_at"])
+
+    _create_index("runs", "ix_runs_created_user_status", ["created_at", "user_id", "status"])
+    _create_index("runs", "ix_runs_user_created", ["user_id", "created_at"])
+    _create_index("runs", "ix_runs_thread_created", ["thread_id", "created_at"])
+    _create_index("tool_call_metrics", "ix_tool_call_metrics_created_user", ["created_at", "user_id"])
+    _create_index("tool_call_metrics", "ix_tool_call_metrics_run_created", ["run_id", "created_at"])
+    _create_index("llm_call_metrics", "ix_llm_call_metrics_created_user", ["created_at", "user_id"])
+    _create_index("llm_call_metrics", "ix_llm_call_metrics_run_created", ["run_id", "created_at"])
+    _create_index("threads_meta", "ix_threads_meta_created_user", ["created_at", "user_id"])
+    _create_index("threads_meta", "ix_threads_meta_user_created", ["user_id", "created_at"])
+
+
+def downgrade() -> None:
+    _drop_index("threads_meta", "ix_threads_meta_user_created")
+    _drop_index("threads_meta", "ix_threads_meta_created_user")
+    _drop_index("llm_call_metrics", "ix_llm_call_metrics_run_created")
+    _drop_index("llm_call_metrics", "ix_llm_call_metrics_created_user")
+    _drop_index("tool_call_metrics", "ix_tool_call_metrics_run_created")
+    _drop_index("tool_call_metrics", "ix_tool_call_metrics_created_user")
+    _drop_index("runs", "ix_runs_thread_created")
+    _drop_index("runs", "ix_runs_user_created")
+    _drop_index("runs", "ix_runs_created_user_status")
+    if _has_table("admin_leaderboard_daily_stats"):
+        op.drop_table("admin_leaderboard_daily_stats")

offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py

@@ -21,6 +21,7 @@ from deerflow.persistence.agents.model import (
     AgentRow,
     AgentSkillRow,
 )
+from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
 from deerflow.persistence.memory_config.model import UserMemoryConfigRow
 from deerflow.persistence.feedback.model import FeedbackRow
 from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
@@ -58,6 +59,7 @@ __all__ = [
     "AgentFavoriteRow",
     "AgentRow",
     "AgentSkillRow",
+    "AdminLeaderboardDailyStatRow",
     "LlmCallMetricRow",
     "NotificationRow",
     "RecommendedQuestionRow",

offline-backend-20260512/backend/packages/harness/deerflow/persistence/run/model.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from sqlalchemy import Index, String, Text
+from sqlalchemy import Index, String, Text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from deerflow.persistence.base import Base
@@ -47,4 +47,9 @@ class RunRow(Base):
     created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
     updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
 
-    __table_args__ = (Index("ix_runs_thread_status", "thread_id", "status"),)
+    __table_args__ = (
+        Index("ix_runs_thread_status", "thread_id", "status"),
+        Index("ix_runs_created_user_status", "created_at", "user_id", "status"),
+        Index("ix_runs_user_created", "user_id", "created_at"),
+        Index("ix_runs_thread_created", "thread_id", "created_at"),
+    )

offline-backend-20260512/backend/packages/harness/deerflow/persistence/thread_meta/model.py

@@ -4,15 +4,19 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from sqlalchemy import String
+from sqlalchemy import Index, String
 from sqlalchemy.orm import Mapped, mapped_column
 
 from deerflow.persistence.base import Base
 from deerflow.persistence.types import BeijingDateTime, PortableJSON
 
 
-class ThreadMetaRow(Base):
-    __tablename__ = "threads_meta"
+class ThreadMetaRow(Base):
+    __tablename__ = "threads_meta"
+    __table_args__ = (
+        Index("ix_threads_meta_created_user", "created_at", "user_id"),
+        Index("ix_threads_meta_user_created", "user_id", "created_at"),
+    )
 
     thread_id: Mapped[str] = mapped_column(String(64), primary_key=True)
     assistant_id: Mapped[str | None] = mapped_column(String(128), index=True)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/tool_metrics/model.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from datetime import UTC, datetime
 
-from sqlalchemy import Integer, String, Text
+from sqlalchemy import Index, Integer, String, Text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from deerflow.persistence.base import Base
@@ -21,6 +21,10 @@ class ToolCallMetricRow(Base):
     """
 
     __tablename__ = "tool_call_metrics"
+    __table_args__ = (
+        Index("ix_tool_call_metrics_created_user", "created_at", "user_id"),
+        Index("ix_tool_call_metrics_run_created", "run_id", "created_at"),
+    )
 
     id: Mapped[str] = mapped_column(String(64), primary_key=True)
     created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), index=True)

offline-backend-20260512/backend/scripts/sql/leaderboard_indexes_mysql.sql

@@ -0,0 +1,85 @@
+-- ============================================================================
+-- 排行榜(admin leaderboard)统计所需的复合索引 —— MySQL 手动执行版
+-- ============================================================================
+--
+-- 背景:
+--   这些索引声明在各 ORM model 的 __table_args__ 里,但 Base.metadata.create_all
+--   对【已存在的表】不会补建索引,而 alembic 自动迁移在本项目里是关闭的
+--   (engine.py 中 `# await _run_alembic_upgrade(url)` 被注释)。
+--   因此在【已有数据的 MySQL 部署】上,这些索引很可能并未真正创建,排行榜的
+--   单日聚合查询会退化为全表扫描,拖慢并占用连接。
+--
+-- 用法:
+--   1) 先选中目标库(与 database.mysql_url 里的库名一致):
+--        USE your_deerflow_db;
+--   2) 直接整文件执行即可。本脚本是【幂等】的:已存在的索引会跳过,不会报错,
+--      可重复执行。
+--
+-- 注意:
+--   * 大表上建索引可能耗时;MySQL 8.0 在线 DDL 默认 ALGORITHM=INPLACE 一般不阻塞
+--     写入,但仍建议在低峰期执行,或先在备库/预发验证。
+--   * 涉及列均为 String(64) / String(20) 等短列,utf8mb4 下复合索引长度远低于
+--     3072 字节上限,无需前缀索引。
+--   * 索引名与 migration 20260608_01 保持一致,日后若启用 alembic 也不会重复创建。
+-- ============================================================================
+
+-- ---- 通用幂等封装:仅当索引不存在时才创建 --------------------------------------
+DELIMITER //
+DROP PROCEDURE IF EXISTS _create_index_if_missing //
+CREATE PROCEDURE _create_index_if_missing(
+    IN p_table   VARCHAR(64),
+    IN p_index   VARCHAR(64),
+    IN p_columns VARCHAR(255)
+)
+BEGIN
+    DECLARE v_exists INT DEFAULT 0;
+    SELECT COUNT(*) INTO v_exists
+      FROM information_schema.statistics
+     WHERE table_schema = DATABASE()
+       AND table_name   = p_table
+       AND index_name   = p_index;
+
+    IF v_exists = 0 THEN
+        SET @ddl = CONCAT('CREATE INDEX `', p_index, '` ON `', p_table, '` (', p_columns, ')');
+        PREPARE stmt FROM @ddl;
+        EXECUTE stmt;
+        DEALLOCATE PREPARE stmt;
+        SELECT CONCAT('CREATED  ', p_table, '.', p_index) AS result;
+    ELSE
+        SELECT CONCAT('SKIPPED  ', p_table, '.', p_index, ' (already exists)') AS result;
+    END IF;
+END //
+DELIMITER ;
+
+-- ---- runs ---------------------------------------------------------------------
+CALL _create_index_if_missing('runs', 'ix_runs_created_user_status', '`created_at`, `user_id`, `status`');
+CALL _create_index_if_missing('runs', 'ix_runs_user_created',        '`user_id`, `created_at`');
+CALL _create_index_if_missing('runs', 'ix_runs_thread_created',      '`thread_id`, `created_at`');
+
+-- ---- tool_call_metrics -------------------------------------------------------
+CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_created_user', '`created_at`, `user_id`');
+CALL _create_index_if_missing('tool_call_metrics', 'ix_tool_call_metrics_run_created',  '`run_id`, `created_at`');
+
+-- ---- llm_call_metrics --------------------------------------------------------
+CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_created_user', '`created_at`, `user_id`');
+CALL _create_index_if_missing('llm_call_metrics', 'ix_llm_call_metrics_run_created',  '`run_id`, `created_at`');
+
+-- ---- threads_meta ------------------------------------------------------------
+CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_created_user', '`created_at`, `user_id`');
+CALL _create_index_if_missing('threads_meta', 'ix_threads_meta_user_created', '`user_id`, `created_at`');
+
+-- ---- 清理临时存储过程 --------------------------------------------------------
+DROP PROCEDURE IF EXISTS _create_index_if_missing;
+
+-- ---- 执行后自检:确认 9 个索引都已存在 ---------------------------------------
+SELECT table_name, index_name
+  FROM information_schema.statistics
+ WHERE table_schema = DATABASE()
+   AND index_name IN (
+       'ix_runs_created_user_status', 'ix_runs_user_created', 'ix_runs_thread_created',
+       'ix_tool_call_metrics_created_user', 'ix_tool_call_metrics_run_created',
+       'ix_llm_call_metrics_created_user', 'ix_llm_call_metrics_run_created',
+       'ix_threads_meta_created_user', 'ix_threads_meta_user_created'
+   )
+ GROUP BY table_name, index_name
+ ORDER BY table_name, index_name;