feat: 公开用户排行榜页面 + 修复 checkpointer 连接与前端依赖
- 新增公开排行榜页面与后端 public_leaderboard 路由 - 完善 skills/agents 路由 - checkpointer Postgres 改用连接池(自愈), 解决连接关闭后定时任务连环失败 - 修复前端 mermaid 依赖未声明导致 Vite 解析失败 (新增 pnpm-lock) Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.8 committed
1 parent
8113b822c5
commit
80d2db6315
26 files changed
+1877
-967
No files matched your search
@@ -250,6 +250,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
|
||||
| **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET /runs/{rid}`, run-file preview/edit/download; `GET /runs/{rid}/messages` - agent messages for the run's underlying agent run (used to inspect a stuck/running run; authorized via `_load_run`); `GET /runs/{rid}/html` - generated HTML document for an `html_page` run; `DELETE /runs/{rid}` - owner-scoped, with an admin fallback so admins can delete any user's run; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |
|
||||
| **HTML Page Favorites** (`/api/html-page-favorites`) | Bookmarked generated HTML pages, owner-scoped. `GET /` - list; `POST /` `{page_id, title?, description?, tags?}` - bookmark a `scheduled_html_pages` row (extracts a reusable style/layout summary via GLM5 at favorite time); a page can only be favorited once per user (`DuplicateFavoriteError` → HTTP 409). `GET/DELETE /{id}` - read/delete (the create-task reference picker deletes favorites here). A favorite can be selected as a style reference when creating a new `html_page` task. |
|
||||
| **Public Scheduled Tasks** (`/api/public/scheduled-tasks`) | **Unauthenticated** mirror for embedding. Adds `GET /runs/{rid}/html` and `GET /{task_id}/latest-html` for the `html_page` preview; the run result keeps `result.html` (only file paths are redacted). |
|
||||
| **Public Leaderboard** (`/api/public/leaderboard`) | **Unauthenticated** read-only user leaderboard snapshot powering the public publish page (`#/public/leaderboard`). `GET /` returns the same `AdminLeaderboardResponse` as the admin endpoint, reusing `_load_or_schedule_daily_leaderboard` and honoring the system-wide leaderboard settings (excluded users, scheduled/admin/failed inclusion). No admin controls, no mutations. Router: `app/gateway/routers/public_leaderboard.py`. |
|
||||
|
||||
Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` → Gateway.
|
||||
|
||||
|
||||
@@ -2,12 +2,12 @@ import asyncio
|
||||
import logging
|
||||
import os
|
||||
import queue
|
||||
import sys
|
||||
import threading
|
||||
from collections.abc import AsyncGenerator
|
||||
from contextlib import asynccontextmanager
|
||||
from datetime import datetime
|
||||
from time import perf_counter
|
||||
import sys
|
||||
import threading
|
||||
from collections.abc import AsyncGenerator
|
||||
from contextlib import asynccontextmanager
|
||||
from datetime import datetime
|
||||
from time import perf_counter
|
||||
|
||||
from dotenv import load_dotenv
|
||||
from fastapi import FastAPI
|
||||
@@ -29,13 +29,13 @@ from app.gateway.routers import (
|
||||
ai_writing,
|
||||
admin_users,
|
||||
agents,
|
||||
artifacts,
|
||||
assistants_compat,
|
||||
auth,
|
||||
browser_context,
|
||||
channels,
|
||||
checkpoint_migration,
|
||||
feedback,
|
||||
artifacts,
|
||||
assistants_compat,
|
||||
auth,
|
||||
browser_context,
|
||||
channels,
|
||||
checkpoint_migration,
|
||||
feedback,
|
||||
html_page_favorites,
|
||||
intent,
|
||||
llm_metrics,
|
||||
@@ -45,6 +45,7 @@ from app.gateway.routers import (
|
||||
multi_agent,
|
||||
notifications,
|
||||
public_embed,
|
||||
public_leaderboard,
|
||||
public_scheduled_tasks,
|
||||
recommend,
|
||||
recommended_questions,
|
||||
@@ -381,7 +382,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 进程真正
|
||||
@@ -407,344 +408,344 @@ 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 _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 _leaderboard_scheduler_sql_for_log(stmt) -> str:
|
||||
try:
|
||||
return str(stmt.compile(compile_kwargs={"literal_binds": True}))
|
||||
except Exception:
|
||||
return str(stmt)
|
||||
|
||||
|
||||
async def _execute_leaderboard_scheduler_query(session, step: str, stmt, **context):
|
||||
sql = _leaderboard_scheduler_sql_for_log(stmt)
|
||||
logger.info(
|
||||
"Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
datetime.now().isoformat(),
|
||||
sql,
|
||||
)
|
||||
start = perf_counter()
|
||||
try:
|
||||
result = await session.execute(stmt)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
(perf_counter() - start) * 1000,
|
||||
)
|
||||
raise
|
||||
logger.info(
|
||||
"Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
datetime.now().isoformat(),
|
||||
(perf_counter() - start) * 1000,
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
async def _time_leaderboard_scheduler_db_operation(step: str, operation, *, sql: str = "", **context):
|
||||
logger.info(
|
||||
"Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
datetime.now().isoformat(),
|
||||
sql,
|
||||
)
|
||||
start = perf_counter()
|
||||
try:
|
||||
result = await operation()
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
(perf_counter() - start) * 1000,
|
||||
)
|
||||
raise
|
||||
logger.info(
|
||||
"Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
datetime.now().isoformat(),
|
||||
(perf_counter() - start) * 1000,
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
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:
|
||||
stmt = (
|
||||
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)
|
||||
)
|
||||
rows = (
|
||||
await _execute_leaderboard_scheduler_query(
|
||||
session,
|
||||
"scheduler.due_dates",
|
||||
stmt,
|
||||
settings_hash=settings_hash,
|
||||
)
|
||||
).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 _time_leaderboard_scheduler_db_operation(
|
||||
"scheduler.claim.get",
|
||||
lambda: session.get(
|
||||
AdminLeaderboardDailyStatRow,
|
||||
{"stat_date": stat_date, "settings_hash": settings_hash},
|
||||
),
|
||||
sql="SELECT admin_leaderboard_daily_stats by primary key",
|
||||
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 _time_leaderboard_scheduler_db_operation(
|
||||
"scheduler.claim.commit",
|
||||
lambda: session.commit(),
|
||||
sql="COMMIT",
|
||||
stat_date=stat_date,
|
||||
settings_hash=settings_hash,
|
||||
)
|
||||
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:
|
||||
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 _leaderboard_scheduler_sql_for_log(stmt) -> str:
|
||||
try:
|
||||
return str(stmt.compile(compile_kwargs={"literal_binds": True}))
|
||||
except Exception:
|
||||
return str(stmt)
|
||||
|
||||
|
||||
async def _execute_leaderboard_scheduler_query(session, step: str, stmt, **context):
|
||||
sql = _leaderboard_scheduler_sql_for_log(stmt)
|
||||
logger.info(
|
||||
"Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
datetime.now().isoformat(),
|
||||
sql,
|
||||
)
|
||||
start = perf_counter()
|
||||
try:
|
||||
result = await session.execute(stmt)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
(perf_counter() - start) * 1000,
|
||||
)
|
||||
raise
|
||||
logger.info(
|
||||
"Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
datetime.now().isoformat(),
|
||||
(perf_counter() - start) * 1000,
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
async def _time_leaderboard_scheduler_db_operation(step: str, operation, *, sql: str = "", **context):
|
||||
logger.info(
|
||||
"Leaderboard snapshot DB query start: step=%s stat_date=%s job=%s settings_hash=%s started_at=%s sql=%s",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
datetime.now().isoformat(),
|
||||
sql,
|
||||
)
|
||||
start = perf_counter()
|
||||
try:
|
||||
result = await operation()
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Leaderboard snapshot DB query failed: step=%s stat_date=%s job=%s settings_hash=%s elapsed_ms=%.2f",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
(perf_counter() - start) * 1000,
|
||||
)
|
||||
raise
|
||||
logger.info(
|
||||
"Leaderboard snapshot DB query done: step=%s stat_date=%s job=%s settings_hash=%s finished_at=%s elapsed_ms=%.2f",
|
||||
step,
|
||||
context.get("stat_date") or context.get("stat_dates") or "",
|
||||
context.get("job_key") or "",
|
||||
context.get("settings_hash") or "",
|
||||
datetime.now().isoformat(),
|
||||
(perf_counter() - start) * 1000,
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
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:
|
||||
stmt = (
|
||||
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)
|
||||
)
|
||||
rows = (
|
||||
await _execute_leaderboard_scheduler_query(
|
||||
session,
|
||||
"scheduler.due_dates",
|
||||
stmt,
|
||||
settings_hash=settings_hash,
|
||||
)
|
||||
).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 _time_leaderboard_scheduler_db_operation(
|
||||
"scheduler.claim.get",
|
||||
lambda: session.get(
|
||||
AdminLeaderboardDailyStatRow,
|
||||
{"stat_date": stat_date, "settings_hash": settings_hash},
|
||||
),
|
||||
sql="SELECT admin_leaderboard_daily_stats by primary key",
|
||||
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 _time_leaderboard_scheduler_db_operation(
|
||||
"scheduler.claim.commit",
|
||||
lambda: session.commit(),
|
||||
sql="COMMIT",
|
||||
stat_date=stat_date,
|
||||
settings_hash=settings_hash,
|
||||
)
|
||||
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``
|
||||
@@ -898,21 +899,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)")
|
||||
|
||||
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
|
||||
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())
|
||||
@@ -956,17 +957,17 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
except Exception:
|
||||
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")
|
||||
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 出口由
|
||||
@@ -1198,14 +1199,14 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
|
||||
# Artifacts API is mounted at /api/threads/{thread_id}/artifacts
|
||||
app.include_router(artifacts.router)
|
||||
|
||||
# Uploads API is mounted at /api/threads/{thread_id}/uploads
|
||||
app.include_router(uploads.router)
|
||||
|
||||
# Optional Chrome-extension page capture API is mounted at /api/browser-context
|
||||
app.include_router(browser_context.router)
|
||||
|
||||
# Thread cleanup API is mounted at /api/threads/{thread_id}
|
||||
app.include_router(threads.router)
|
||||
# Uploads API is mounted at /api/threads/{thread_id}/uploads
|
||||
app.include_router(uploads.router)
|
||||
|
||||
# Optional Chrome-extension page capture API is mounted at /api/browser-context
|
||||
app.include_router(browser_context.router)
|
||||
|
||||
# Thread cleanup API is mounted at /api/threads/{thread_id}
|
||||
app.include_router(threads.router)
|
||||
|
||||
# Agents API is mounted at /api/agents
|
||||
app.include_router(agents.router)
|
||||
@@ -1244,6 +1245,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
|
||||
# surface results without a DeerFlow login.
|
||||
app.include_router(public_scheduled_tasks.router)
|
||||
|
||||
# Public read-only user leaderboard publish page (/api/public/leaderboard)
|
||||
app.include_router(public_leaderboard.router)
|
||||
|
||||
# Public embed Q&A status polling (/api/public/embed/*)
|
||||
app.include_router(public_embed.router)
|
||||
app.include_router(html_page_favorites.router)
|
||||
@@ -1254,18 +1258,18 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
|
||||
# Admin LLM metrics audit + export
|
||||
app.include_router(llm_metrics.router)
|
||||
|
||||
# Admin window into other users' threads / messages / memory
|
||||
app.include_router(admin_users.router)
|
||||
app.include_router(admin_users.self_router)
|
||||
|
||||
# Admin tool / skill call metrics audit + export
|
||||
# Admin window into other users' threads / messages / memory
|
||||
app.include_router(admin_users.router)
|
||||
app.include_router(admin_users.self_router)
|
||||
|
||||
# Admin tool / skill call metrics audit + export
|
||||
app.include_router(tool_metrics.router)
|
||||
|
||||
# Channels API is mounted at /api/channels
|
||||
app.include_router(channels.router)
|
||||
|
||||
# Admin checkpoint migration API: SQLite legacy checkpoints -> Postgres.
|
||||
app.include_router(checkpoint_migration.router)
|
||||
app.include_router(channels.router)
|
||||
|
||||
# Admin checkpoint migration API: SQLite legacy checkpoints -> Postgres.
|
||||
app.include_router(checkpoint_migration.router)
|
||||
|
||||
# Assistants compatibility API (LangGraph Platform stub)
|
||||
app.include_router(assistants_compat.router)
|
||||
|
||||
@@ -15,7 +15,7 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.gateway.auth.password import verify_password_async
|
||||
from app.gateway.deps import get_agent_store, get_config, get_tag_store
|
||||
from app.gateway.deps import get_agent_store, get_config, get_tag_store
|
||||
from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async
|
||||
from deerflow.config.agents_api_config import get_agents_api_config
|
||||
from deerflow.config.agents_config import list_custom_agents, load_agent_config, load_agent_soul, validate_agent_id
|
||||
@@ -39,6 +39,7 @@ class AgentResponse(BaseModel):
|
||||
name: str = Field(..., description="Display name. Chinese and duplicate names are allowed.")
|
||||
description: str = Field(default="", description="Agent description")
|
||||
user_id: str | None = Field(default=None, description="Owner user id. Null means built-in.")
|
||||
owner_username: str | None = Field(default=None, description="Publisher username (email local part). Null for built-in agents.")
|
||||
published: bool = Field(default=False, description="Whether a user-created agent is public")
|
||||
is_private: bool = Field(default=False, description="Whether the agent lives in the password-gated private square instead of the public one")
|
||||
builtin: bool = Field(default=False, description="True when user_id is null")
|
||||
@@ -62,15 +63,15 @@ class AgentResponse(BaseModel):
|
||||
is_favorite: bool = Field(
|
||||
default=False,
|
||||
description="Whether the requesting user has favorited (收藏) this agent.",
|
||||
)
|
||||
|
||||
is_pinned: bool = Field(
|
||||
default=False,
|
||||
description="Whether the requesting user has pinned this agent.",
|
||||
)
|
||||
|
||||
|
||||
class AgentsListResponse(BaseModel):
|
||||
)
|
||||
|
||||
is_pinned: bool = Field(
|
||||
default=False,
|
||||
description="Whether the requesting user has pinned this agent.",
|
||||
)
|
||||
|
||||
|
||||
class AgentsListResponse(BaseModel):
|
||||
"""Response model for listing agents.
|
||||
|
||||
``total`` / ``page`` / ``page_size`` describe server-side pagination. When
|
||||
@@ -115,7 +116,7 @@ class AgentUpdateRequest(BaseModel):
|
||||
tool_groups: list[str] | None = Field(default=None, description="Updated tool group whitelist (config.yaml)")
|
||||
|
||||
|
||||
class AgentSelectorResponse(BaseModel):
|
||||
class AgentSelectorResponse(BaseModel):
|
||||
"""Response for the homepage agent selector dropdown.
|
||||
|
||||
The list is fully curated by admins (``featured_order`` on each row).
|
||||
@@ -125,10 +126,10 @@ class AgentSelectorResponse(BaseModel):
|
||||
agents: list[AgentResponse] = Field(
|
||||
default_factory=list,
|
||||
description="Admin-curated agents, sorted by featured_order ascending.",
|
||||
)
|
||||
|
||||
|
||||
class AgentFeaturedRequest(BaseModel):
|
||||
)
|
||||
|
||||
|
||||
class AgentFeaturedRequest(BaseModel):
|
||||
"""Request body for setting/clearing an agent's homepage featured rank."""
|
||||
|
||||
order: int | None = Field(
|
||||
@@ -159,6 +160,41 @@ def _is_admin_user(request: Request) -> bool:
|
||||
return getattr(user, "system_role", None) == "admin"
|
||||
|
||||
|
||||
def _email_local_part(email: str | None) -> str | None:
|
||||
"""Return the username portion (before ``@``) of an email address."""
|
||||
if not email:
|
||||
return None
|
||||
at = email.find("@")
|
||||
return email[:at] if at > 0 else email
|
||||
|
||||
|
||||
async def _resolve_usernames(user_ids: set[str]) -> dict[str, str]:
|
||||
"""Batch-resolve owner user ids to friendly usernames (email local part).
|
||||
|
||||
One lookup per distinct id; tolerates missing accounts (deleted users) and
|
||||
an uninitialised auth provider by simply omitting those ids from the map.
|
||||
"""
|
||||
user_ids = {uid for uid in user_ids if uid}
|
||||
if not user_ids:
|
||||
return {}
|
||||
try:
|
||||
from app.gateway.deps import get_local_provider
|
||||
|
||||
provider = get_local_provider()
|
||||
except Exception:
|
||||
return {}
|
||||
names: dict[str, str] = {}
|
||||
for uid in user_ids:
|
||||
try:
|
||||
user = await provider.get_user(uid)
|
||||
except Exception:
|
||||
user = None
|
||||
local = _email_local_part(getattr(user, "email", None)) if user is not None else None
|
||||
if local:
|
||||
names[uid] = local
|
||||
return names
|
||||
|
||||
|
||||
def _validate_display_name(name: str) -> str:
|
||||
normalized = name.strip()
|
||||
if not normalized:
|
||||
@@ -348,7 +384,7 @@ def _ensure_agent_dir_from_record(record: dict[str, Any]) -> None:
|
||||
logger.exception("Failed to self-heal agent dir for '%s'", agent_id)
|
||||
|
||||
|
||||
def _record_to_response(
|
||||
def _record_to_response(
|
||||
record: dict[str, Any],
|
||||
*,
|
||||
include_soul: bool = False,
|
||||
@@ -390,21 +426,21 @@ def _record_to_response(
|
||||
tool_groups=extras["tool_groups"],
|
||||
skills=extras["skills"],
|
||||
featured_order=record.get("featured_order"),
|
||||
)
|
||||
|
||||
|
||||
def _sort_agent_responses(responses: list[AgentResponse], sort: str) -> list[AgentResponse]:
|
||||
reverse = sort not in {"time_asc", "updated_asc", "name_asc"}
|
||||
if sort in {"name_asc", "name_desc"}:
|
||||
ordered = sorted(responses, key=lambda r: (r.name or "").lower(), reverse=reverse)
|
||||
elif sort in {"updated_desc", "updated_asc"}:
|
||||
ordered = sorted(responses, key=lambda r: str(r.updated_at or ""), reverse=reverse)
|
||||
else:
|
||||
ordered = sorted(responses, key=lambda r: str(r.created_at or ""), reverse=reverse)
|
||||
return sorted(ordered, key=lambda r: 0 if r.is_pinned else 1)
|
||||
|
||||
|
||||
async def _record_to_response_async(
|
||||
)
|
||||
|
||||
|
||||
def _sort_agent_responses(responses: list[AgentResponse], sort: str) -> list[AgentResponse]:
|
||||
reverse = sort not in {"time_asc", "updated_asc", "name_asc"}
|
||||
if sort in {"name_asc", "name_desc"}:
|
||||
ordered = sorted(responses, key=lambda r: (r.name or "").lower(), reverse=reverse)
|
||||
elif sort in {"updated_desc", "updated_asc"}:
|
||||
ordered = sorted(responses, key=lambda r: str(r.updated_at or ""), reverse=reverse)
|
||||
else:
|
||||
ordered = sorted(responses, key=lambda r: str(r.created_at or ""), reverse=reverse)
|
||||
return sorted(ordered, key=lambda r: 0 if r.is_pinned else 1)
|
||||
|
||||
|
||||
async def _record_to_response_async(
|
||||
record: dict[str, Any],
|
||||
*,
|
||||
include_soul: bool = False,
|
||||
@@ -478,20 +514,23 @@ async def _build_list_responses(
|
||||
_record_to_response(a, include_soul=False, include_extras=False) for a in records
|
||||
]
|
||||
ids = [r.id for r in responses]
|
||||
extras = await store.get_extras_for(ids)
|
||||
assignments = await tag_store.list_assignments("agent", ids)
|
||||
favorite_ids = await store.list_favorite_ids(user_id)
|
||||
pin_ids = await store.list_pin_ids(user_id)
|
||||
for resp in responses:
|
||||
extras = await store.get_extras_for(ids)
|
||||
assignments = await tag_store.list_assignments("agent", ids)
|
||||
favorite_ids = await store.list_favorite_ids(user_id)
|
||||
pin_ids = await store.list_pin_ids(user_id)
|
||||
usernames = await _resolve_usernames({r.user_id for r in responses if r.user_id})
|
||||
for resp in responses:
|
||||
cached = extras.get(resp.id)
|
||||
if cached is not None:
|
||||
resp.skills = cached.get("skills")
|
||||
resp.tool_groups = cached.get("tool_groups")
|
||||
resp.model = cached.get("model")
|
||||
resp.tags = assignments.get(resp.id, [])
|
||||
resp.is_favorite = resp.id in favorite_ids
|
||||
resp.is_pinned = resp.id in pin_ids
|
||||
return responses
|
||||
resp.tags = assignments.get(resp.id, [])
|
||||
resp.is_favorite = resp.id in favorite_ids
|
||||
resp.is_pinned = resp.id in pin_ids
|
||||
if resp.user_id:
|
||||
resp.owner_username = usernames.get(resp.user_id)
|
||||
return responses
|
||||
|
||||
|
||||
@router.get(
|
||||
@@ -509,14 +548,14 @@ async def list_agents(
|
||||
pattern="^(mine|square)$",
|
||||
description="Pagination scope. 'mine' = owned + favorited; 'square' = built-in/published public. Omit for the default combined visibility list.",
|
||||
),
|
||||
page: int | None = Query(default=None, ge=1, description="1-based page index. Providing this (or 'scope') enables server-side pagination."),
|
||||
page_size: int = Query(default=20, ge=1, le=200, description="Items per page when paginating."),
|
||||
sort: str = Query(
|
||||
default="time_desc",
|
||||
pattern="^(time_desc|time_asc|updated_desc|updated_asc|name_asc|name_desc)$",
|
||||
description="Sort order. Default is newest created first.",
|
||||
),
|
||||
admin: bool = Query(
|
||||
page: int | None = Query(default=None, ge=1, description="1-based page index. Providing this (or 'scope') enables server-side pagination."),
|
||||
page_size: int = Query(default=20, ge=1, le=200, description="Items per page when paginating."),
|
||||
sort: str = Query(
|
||||
default="time_desc",
|
||||
pattern="^(time_desc|time_asc|updated_desc|updated_asc|name_asc|name_desc)$",
|
||||
description="Sort order. Default is newest created first.",
|
||||
),
|
||||
admin: bool = Query(
|
||||
default=False,
|
||||
description=(
|
||||
"Admin-only override. When ``true`` and the caller is a system admin, "
|
||||
@@ -543,11 +582,11 @@ async def list_agents(
|
||||
scope=scope,
|
||||
search=search,
|
||||
tag_ids=tag_ids,
|
||||
page=eff_page,
|
||||
page_size=page_size,
|
||||
admin=admin,
|
||||
sort=sort,
|
||||
)
|
||||
page=eff_page,
|
||||
page_size=page_size,
|
||||
admin=admin,
|
||||
sort=sort,
|
||||
)
|
||||
responses = await _build_list_responses(store, tag_store, user_id, rows)
|
||||
return AgentsListResponse(agents=responses, total=total, page=eff_page, page_size=page_size)
|
||||
|
||||
@@ -559,11 +598,11 @@ async def list_agents(
|
||||
if search:
|
||||
needle = search.strip().lower()
|
||||
responses = [r for r in responses if needle in r.name.lower()]
|
||||
if tag_ids:
|
||||
wanted = set(tag_ids)
|
||||
responses = [r for r in responses if any(t.get("id") in wanted for t in r.tags)]
|
||||
responses = _sort_agent_responses(responses, sort)
|
||||
return AgentsListResponse(agents=responses, total=len(responses), page=1, page_size=len(responses))
|
||||
if tag_ids:
|
||||
wanted = set(tag_ids)
|
||||
responses = [r for r in responses if any(t.get("id") in wanted for t in r.tags)]
|
||||
responses = _sort_agent_responses(responses, sort)
|
||||
return AgentsListResponse(agents=responses, total=len(responses), page=1, page_size=len(responses))
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
@@ -576,14 +615,14 @@ async def list_agents(
|
||||
summary="Check Agent Name",
|
||||
description="Validate an agent display name. Duplicate names are allowed.",
|
||||
)
|
||||
async def check_agent_name(name: str) -> dict[str, Any]:
|
||||
_require_agents_api_enabled()
|
||||
normalized = _validate_display_name(name)
|
||||
return {"available": True, "name": normalized}
|
||||
|
||||
|
||||
@router.get(
|
||||
"/agents/selector",
|
||||
async def check_agent_name(name: str) -> dict[str, Any]:
|
||||
_require_agents_api_enabled()
|
||||
normalized = _validate_display_name(name)
|
||||
return {"available": True, "name": normalized}
|
||||
|
||||
|
||||
@router.get(
|
||||
"/agents/selector",
|
||||
response_model=AgentSelectorResponse,
|
||||
summary="Agent Homepage Selector",
|
||||
description=(
|
||||
@@ -635,12 +674,12 @@ async def get_agent(agent_id: str, request: Request) -> AgentResponse:
|
||||
# Detail view legitimately reads SOUL.md + config.yaml — hop to a
|
||||
# worker thread so the asyncio loop stays responsive under concurrent
|
||||
# detail fetches.
|
||||
response = await _record_to_response_async(record, include_soul=True, include_extras=True)
|
||||
favorite_ids = await store.list_favorite_ids(user_id)
|
||||
pin_ids = await store.list_pin_ids(user_id)
|
||||
response.is_favorite = agent_id in favorite_ids
|
||||
response.is_pinned = agent_id in pin_ids
|
||||
return response
|
||||
response = await _record_to_response_async(record, include_soul=True, include_extras=True)
|
||||
favorite_ids = await store.list_favorite_ids(user_id)
|
||||
pin_ids = await store.list_pin_ids(user_id)
|
||||
response.is_favorite = agent_id in favorite_ids
|
||||
response.is_pinned = agent_id in pin_ids
|
||||
return response
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
@@ -874,16 +913,16 @@ async def delete_agent(agent_id: str, request: Request) -> None:
|
||||
await get_tag_store(request).unassign_all("agent", agent_id)
|
||||
except Exception: # noqa: BLE001 — tag cleanup must not block deletion
|
||||
pass
|
||||
try:
|
||||
await store.delete_favorites_for_agent(agent_id)
|
||||
except Exception: # noqa: BLE001 — favorite cleanup must not block deletion
|
||||
pass
|
||||
try:
|
||||
await store.delete_pins_for_agent(agent_id)
|
||||
except Exception: # noqa: BLE001 - pin cleanup must not block deletion
|
||||
pass
|
||||
try:
|
||||
await store.delete_extras_for_agent(agent_id)
|
||||
try:
|
||||
await store.delete_favorites_for_agent(agent_id)
|
||||
except Exception: # noqa: BLE001 — favorite cleanup must not block deletion
|
||||
pass
|
||||
try:
|
||||
await store.delete_pins_for_agent(agent_id)
|
||||
except Exception: # noqa: BLE001 - pin cleanup must not block deletion
|
||||
pass
|
||||
try:
|
||||
await store.delete_extras_for_agent(agent_id)
|
||||
except Exception: # noqa: BLE001 — cache cleanup must not block deletion
|
||||
pass
|
||||
logger.info("Deleted agent '%s' from %s", agent_id, agent_dir)
|
||||
@@ -925,7 +964,7 @@ async def favorite_agent(agent_id: str, request: Request) -> None:
|
||||
summary="Unfavorite Agent",
|
||||
description="Remove an agent from the current user's favorites. No-op when not favorited.",
|
||||
)
|
||||
async def unfavorite_agent(agent_id: str, request: Request) -> None:
|
||||
async def unfavorite_agent(agent_id: str, request: Request) -> None:
|
||||
_require_agents_api_enabled()
|
||||
agent_id = _validate_storage_id(agent_id)
|
||||
|
||||
@@ -936,60 +975,60 @@ async def unfavorite_agent(agent_id: str, request: Request) -> None:
|
||||
logger.info("User '%s' unfavorited agent '%s'", user_id, agent_id)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("Failed to unfavorite agent '%s': %s", agent_id, e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to unfavorite agent: {str(e)}") from e
|
||||
|
||||
|
||||
@router.post(
|
||||
"/agents/{agent_id}/pin",
|
||||
status_code=204,
|
||||
summary="Pin Agent",
|
||||
description="Pin a visible agent to the top of the current user's lists. Idempotent.",
|
||||
)
|
||||
async def pin_agent(agent_id: str, request: Request) -> None:
|
||||
_require_agents_api_enabled()
|
||||
agent_id = _validate_storage_id(agent_id)
|
||||
|
||||
try:
|
||||
store = get_agent_store(request)
|
||||
user_id = _current_user_id(request)
|
||||
record = await store.get_visible(agent_id, user_id)
|
||||
if record is None:
|
||||
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
|
||||
await store.add_pin(user_id, agent_id)
|
||||
logger.info("User '%s' pinned agent '%s'", user_id, agent_id)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("Failed to pin agent '%s': %s", agent_id, e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to pin agent: {str(e)}") from e
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/agents/{agent_id}/pin",
|
||||
status_code=204,
|
||||
summary="Unpin Agent",
|
||||
description="Remove the current user's pin from an agent. No-op when not pinned.",
|
||||
)
|
||||
async def unpin_agent(agent_id: str, request: Request) -> None:
|
||||
_require_agents_api_enabled()
|
||||
agent_id = _validate_storage_id(agent_id)
|
||||
|
||||
try:
|
||||
store = get_agent_store(request)
|
||||
user_id = _current_user_id(request)
|
||||
await store.remove_pin(user_id, agent_id)
|
||||
logger.info("User '%s' unpinned agent '%s'", user_id, agent_id)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("Failed to unpin agent '%s': %s", agent_id, e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to unpin agent: {str(e)}") from e
|
||||
|
||||
|
||||
@router.put(
|
||||
"/agents/{agent_id}/featured",
|
||||
except Exception as e:
|
||||
logger.error("Failed to unfavorite agent '%s': %s", agent_id, e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to unfavorite agent: {str(e)}") from e
|
||||
|
||||
|
||||
@router.post(
|
||||
"/agents/{agent_id}/pin",
|
||||
status_code=204,
|
||||
summary="Pin Agent",
|
||||
description="Pin a visible agent to the top of the current user's lists. Idempotent.",
|
||||
)
|
||||
async def pin_agent(agent_id: str, request: Request) -> None:
|
||||
_require_agents_api_enabled()
|
||||
agent_id = _validate_storage_id(agent_id)
|
||||
|
||||
try:
|
||||
store = get_agent_store(request)
|
||||
user_id = _current_user_id(request)
|
||||
record = await store.get_visible(agent_id, user_id)
|
||||
if record is None:
|
||||
raise HTTPException(status_code=404, detail=f"Agent '{agent_id}' not found")
|
||||
await store.add_pin(user_id, agent_id)
|
||||
logger.info("User '%s' pinned agent '%s'", user_id, agent_id)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("Failed to pin agent '%s': %s", agent_id, e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to pin agent: {str(e)}") from e
|
||||
|
||||
|
||||
@router.delete(
|
||||
"/agents/{agent_id}/pin",
|
||||
status_code=204,
|
||||
summary="Unpin Agent",
|
||||
description="Remove the current user's pin from an agent. No-op when not pinned.",
|
||||
)
|
||||
async def unpin_agent(agent_id: str, request: Request) -> None:
|
||||
_require_agents_api_enabled()
|
||||
agent_id = _validate_storage_id(agent_id)
|
||||
|
||||
try:
|
||||
store = get_agent_store(request)
|
||||
user_id = _current_user_id(request)
|
||||
await store.remove_pin(user_id, agent_id)
|
||||
logger.info("User '%s' unpinned agent '%s'", user_id, agent_id)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("Failed to unpin agent '%s': %s", agent_id, e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to unpin agent: {str(e)}") from e
|
||||
|
||||
|
||||
@router.put(
|
||||
"/agents/{agent_id}/featured",
|
||||
response_model=AgentResponse,
|
||||
summary="Set Agent Featured Rank (admin)",
|
||||
description=(
|
||||
@@ -1031,19 +1070,19 @@ async def set_agent_featured(
|
||||
raise HTTPException(status_code=500, detail=f"Failed to set featured: {str(e)}") from e
|
||||
|
||||
|
||||
class PrivateSquareListRequest(BaseModel):
|
||||
class PrivateSquareListRequest(BaseModel):
|
||||
"""Request body for listing all private-square agents."""
|
||||
|
||||
password: str = Field(..., description="Plain-text private-square password. Verified against the admin-configured hash before the listing is returned.")
|
||||
search: str | None = Field(default=None, description="Case-insensitive fuzzy match on agent name (raw or desensitized).")
|
||||
tag_ids: list[str] | None = Field(default=None, description="Keep agents carrying any of these tags (OR).")
|
||||
page: int | None = Field(default=None, ge=1, description="1-based page index. Providing this enables server-side pagination.")
|
||||
page_size: int = Field(default=20, ge=1, le=200, description="Items per page when paginating.")
|
||||
sort: str = Field(
|
||||
default="time_desc",
|
||||
pattern="^(time_desc|time_asc|updated_desc|updated_asc|name_asc|name_desc)$",
|
||||
description="Sort order. Default is newest created first.",
|
||||
)
|
||||
page: int | None = Field(default=None, ge=1, description="1-based page index. Providing this enables server-side pagination.")
|
||||
page_size: int = Field(default=20, ge=1, le=200, description="Items per page when paginating.")
|
||||
sort: str = Field(
|
||||
default="time_desc",
|
||||
pattern="^(time_desc|time_asc|updated_desc|updated_asc|name_asc|name_desc)$",
|
||||
description="Sort order. Default is newest created first.",
|
||||
)
|
||||
|
||||
|
||||
class AgentPrivateUpdateRequest(BaseModel):
|
||||
@@ -1086,17 +1125,17 @@ async def list_private_square(body: PrivateSquareListRequest, request: Request)
|
||||
scope="private",
|
||||
search=body.search,
|
||||
tag_ids=body.tag_ids,
|
||||
page=body.page,
|
||||
page_size=body.page_size,
|
||||
sort=body.sort,
|
||||
)
|
||||
page=body.page,
|
||||
page_size=body.page_size,
|
||||
sort=body.sort,
|
||||
)
|
||||
responses = await _build_list_responses(store, tag_store, user_id, rows)
|
||||
return AgentsListResponse(agents=responses, total=total, page=body.page, page_size=body.page_size)
|
||||
|
||||
records = await store.list_private()
|
||||
responses = await _build_list_responses(store, tag_store, user_id, records)
|
||||
responses = _sort_agent_responses(responses, body.sort)
|
||||
return AgentsListResponse(agents=responses, total=len(responses), page=1, page_size=len(responses))
|
||||
responses = await _build_list_responses(store, tag_store, user_id, records)
|
||||
responses = _sort_agent_responses(responses, body.sort)
|
||||
return AgentsListResponse(agents=responses, total=len(responses), page=1, page_size=len(responses))
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
"""Public (unauthenticated) read-only user leaderboard.
|
||||
|
||||
Mounted under the ``/api/public/`` prefix so the auth middleware lets this route
|
||||
through without a session cookie. Powers the public "publish" page for the user
|
||||
leaderboard — a view-only snapshot with no admin controls and no mutations.
|
||||
|
||||
The snapshot honors the same system-wide leaderboard settings (excluded users,
|
||||
scheduled/admin/failed inclusion) as the authenticated admin endpoint, so any
|
||||
accounts an admin chose to hide stay hidden on the public page as well.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Query
|
||||
|
||||
from app.gateway.routers.admin_users import (
|
||||
AdminLeaderboardResponse,
|
||||
_load_or_schedule_daily_leaderboard,
|
||||
_resolve_range,
|
||||
)
|
||||
from deerflow.config.system_settings import load_system_settings
|
||||
from deerflow.persistence.engine import get_session_factory
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Whitelisted under ``auth_middleware._PUBLIC_PATH_PREFIXES`` (``/api/public/``).
|
||||
router = APIRouter(prefix="/api/public/leaderboard", tags=["public-leaderboard"])
|
||||
|
||||
|
||||
def _mask_email(email: str | None) -> str:
|
||||
"""Reduce an email to its local part so the public page can show a
|
||||
recognizable handle without leaking the full address (domain).
|
||||
|
||||
The leaderboard only makes sense if it names its top users, so the local
|
||||
part is kept intentionally; the privacy gain is dropping the domain and the
|
||||
raw account id from the unauthenticated payload.
|
||||
"""
|
||||
if not email:
|
||||
return ""
|
||||
return email.split("@", 1)[0].strip() or ""
|
||||
|
||||
|
||||
def _anonymize(resp: AdminLeaderboardResponse) -> AdminLeaderboardResponse:
|
||||
"""Strip PII from a leaderboard snapshot before it leaves over the public,
|
||||
unauthenticated endpoint: email → local part, account/thread ids → opaque
|
||||
short hashes (kept stable so they still work as React keys)."""
|
||||
|
||||
def hashed(value: str | None) -> str:
|
||||
if not value:
|
||||
return ""
|
||||
return hashlib.sha1(value.encode("utf-8")).hexdigest()[:12]
|
||||
|
||||
seen_users: set[int] = set()
|
||||
for bucket in (resp.users_by_activity, resp.users_by_questions, resp.users_by_tokens):
|
||||
for user in bucket:
|
||||
if id(user) in seen_users:
|
||||
continue
|
||||
seen_users.add(id(user))
|
||||
user.email = _mask_email(user.email)
|
||||
user.user_id = hashed(user.user_id)
|
||||
|
||||
for bucket in (resp.recent_questions, resp.frequent_questions):
|
||||
for question in bucket:
|
||||
question.email = _mask_email(question.email)
|
||||
question.user_id = hashed(question.user_id)
|
||||
question.thread_id = ""
|
||||
question.run_id = ""
|
||||
|
||||
return resp
|
||||
|
||||
|
||||
@router.get("", response_model=AdminLeaderboardResponse)
|
||||
@router.get("/", response_model=AdminLeaderboardResponse)
|
||||
async def get_public_leaderboard(
|
||||
days: int = Query(default=30, ge=1, le=365),
|
||||
since: str | None = Query(default=None, description="Inclusive range start, ISO datetime"),
|
||||
until: str | None = Query(default=None, description="Inclusive range end, ISO datetime"),
|
||||
limit: int = Query(default=20, ge=5, le=100),
|
||||
) -> AdminLeaderboardResponse:
|
||||
"""Return a read-only leaderboard snapshot for the public publish page."""
|
||||
session_factory = get_session_factory()
|
||||
if session_factory is None:
|
||||
raise HTTPException(status_code=503, detail="Analytics requires database persistence")
|
||||
|
||||
settings = load_system_settings().leaderboard
|
||||
since_dt, until_dt, _range_days = _resolve_range(days, since, until)
|
||||
resp = await _load_or_schedule_daily_leaderboard(
|
||||
session_factory,
|
||||
settings,
|
||||
since_dt,
|
||||
until_dt,
|
||||
limit=limit,
|
||||
)
|
||||
return _anonymize(resp)
|
||||
@@ -56,6 +56,44 @@ def _is_admin_user(request: Request) -> bool:
|
||||
return getattr(user, "system_role", None) == "admin"
|
||||
|
||||
|
||||
def _email_local_part(email: str | None) -> str | None:
|
||||
"""Return the username portion (before ``@``) of an email address."""
|
||||
if not email:
|
||||
return None
|
||||
at = email.find("@")
|
||||
return email[:at] if at > 0 else email
|
||||
|
||||
|
||||
async def _attach_owner_usernames(responses: list["SkillResponse"]) -> list["SkillResponse"]:
|
||||
"""Resolve each custom skill's ``owner_user_id`` to a friendly username.
|
||||
|
||||
One lookup per distinct owner id; tolerates deleted accounts and an
|
||||
uninitialised auth provider by leaving ``owner_username`` as None.
|
||||
"""
|
||||
owner_ids = {r.owner_user_id for r in responses if r.owner_user_id}
|
||||
if not owner_ids:
|
||||
return responses
|
||||
try:
|
||||
from app.gateway.deps import get_local_provider
|
||||
|
||||
provider = get_local_provider()
|
||||
except Exception:
|
||||
return responses
|
||||
names: dict[str, str] = {}
|
||||
for uid in owner_ids:
|
||||
try:
|
||||
user = await provider.get_user(uid)
|
||||
except Exception:
|
||||
user = None
|
||||
local = _email_local_part(getattr(user, "email", None)) if user is not None else None
|
||||
if local:
|
||||
names[uid] = local
|
||||
for resp in responses:
|
||||
if resp.owner_user_id:
|
||||
resp.owner_username = names.get(resp.owner_user_id)
|
||||
return responses
|
||||
|
||||
|
||||
def _can_view_skill_source(
|
||||
category: SkillCategory,
|
||||
owner_user_id: str | None,
|
||||
@@ -125,6 +163,7 @@ class SkillResponse(BaseModel):
|
||||
featured: bool = Field(default=False, description="Whether this skill is pinned to the top of the list (sort only)")
|
||||
source: str = Field(default="upload", description="Origin: 'agent' for agent-created, 'upload' for user-uploaded")
|
||||
owner_user_id: str | None = Field(default=None, description="Custom skill owner (null = built-in or legacy)")
|
||||
owner_username: str | None = Field(default=None, description="Publisher username (email local part). Null for built-in skills.")
|
||||
published: bool = Field(default=False, description="Whether the owner has made this skill visible to other users")
|
||||
builtin: bool = Field(default=False, description="True when owner_user_id is null (no human owner)")
|
||||
detail: str | None = Field(default=None, description="Human-authored long-form detail text (Markdown), editable by owner/admin")
|
||||
@@ -786,6 +825,7 @@ async def list_skills(
|
||||
)
|
||||
)
|
||||
responses = await _attach_and_filter_tags(tag_store, responses, search, tag_ids)
|
||||
responses = await _attach_owner_usernames(responses)
|
||||
responses = _sort_skill_responses(responses, sort)
|
||||
return SkillsListResponse(skills=responses)
|
||||
except HTTPException:
|
||||
@@ -1176,6 +1216,7 @@ async def list_custom_skills(
|
||||
display_sample=adapter.get("sample") if adapter else None,
|
||||
)
|
||||
)
|
||||
responses = await _attach_owner_usernames(responses)
|
||||
return SkillsListResponse(skills=responses)
|
||||
except Exception as e:
|
||||
logger.error("Failed to list custom skills: %s", e, exc_info=True)
|
||||
|
||||
+174
-130
@@ -1,39 +1,39 @@
|
||||
"""Async checkpointer factory.
|
||||
|
||||
Provides an **async context manager** for long-running async servers that need
|
||||
proper resource cleanup.
|
||||
|
||||
Supported backends: memory, sqlite, postgres.
|
||||
|
||||
Usage (e.g. FastAPI lifespan)::
|
||||
|
||||
from deerflow.runtime.checkpointer.async_provider import make_checkpointer
|
||||
|
||||
async with make_checkpointer() as checkpointer:
|
||||
app.state.checkpointer = checkpointer # InMemorySaver if not configured
|
||||
|
||||
For sync usage see :mod:`deerflow.runtime.checkpointer.provider`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
"""Async checkpointer factory.
|
||||
|
||||
Provides an **async context manager** for long-running async servers that need
|
||||
proper resource cleanup.
|
||||
|
||||
Supported backends: memory, sqlite, postgres.
|
||||
|
||||
Usage (e.g. FastAPI lifespan)::
|
||||
|
||||
from deerflow.runtime.checkpointer.async_provider import make_checkpointer
|
||||
|
||||
async with make_checkpointer() as checkpointer:
|
||||
app.state.checkpointer = checkpointer # InMemorySaver if not configured
|
||||
|
||||
For sync usage see :mod:`deerflow.runtime.checkpointer.provider`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import contextlib
|
||||
import logging
|
||||
from collections.abc import AsyncIterator
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from langgraph.types import Checkpointer
|
||||
|
||||
from deerflow.config.app_config import AppConfig, get_app_config
|
||||
from deerflow.runtime.checkpointer.provider import (
|
||||
POSTGRES_CONN_REQUIRED,
|
||||
POSTGRES_INSTALL,
|
||||
SQLITE_INSTALL,
|
||||
)
|
||||
|
||||
from langgraph.types import Checkpointer
|
||||
|
||||
from deerflow.config.app_config import AppConfig, get_app_config
|
||||
from deerflow.runtime.checkpointer.provider import (
|
||||
POSTGRES_CONN_REQUIRED,
|
||||
POSTGRES_INSTALL,
|
||||
SQLITE_INSTALL,
|
||||
)
|
||||
from deerflow.runtime.store._sqlite_utils import ensure_sqlite_parent_dir, resolve_sqlite_conn_str
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@@ -79,6 +79,52 @@ def _sqlite_conn_str_for_postgres_fallback(config) -> str:
|
||||
return paths[0] if paths else "store.db"
|
||||
|
||||
|
||||
def _pool_size_env(name: str, default: int) -> int:
|
||||
raw = os.getenv(name, "").strip()
|
||||
if not raw:
|
||||
return default
|
||||
try:
|
||||
return max(0, int(raw))
|
||||
except ValueError:
|
||||
logger.warning("Invalid %s=%r; using default %d", name, raw, default)
|
||||
return default
|
||||
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def _postgres_saver_from_pool(conn_string: str) -> AsyncIterator[Checkpointer]:
|
||||
"""Yield an ``AsyncPostgresSaver`` backed by a self-healing connection pool.
|
||||
|
||||
``AsyncPostgresSaver.from_conn_string`` opens a *single* long-lived
|
||||
connection. If that one connection dies (Postgres restart, idle timeout,
|
||||
network blip, or a prior broken pipeline), every later checkpointer call
|
||||
fails permanently with "the connection is closed" until the process is
|
||||
restarted. A pool with ``check=check_connection`` validates each connection
|
||||
on checkout and transparently replaces dead ones, so the runtime recovers
|
||||
on its own.
|
||||
"""
|
||||
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
|
||||
from psycopg.rows import dict_row
|
||||
from psycopg_pool import AsyncConnectionPool
|
||||
|
||||
pool = AsyncConnectionPool(
|
||||
conninfo=conn_string,
|
||||
min_size=_pool_size_env("DEER_FLOW_PG_POOL_MIN", 1),
|
||||
max_size=_pool_size_env("DEER_FLOW_PG_POOL_MAX", 10),
|
||||
max_idle=300.0,
|
||||
max_lifetime=3600.0,
|
||||
check=AsyncConnectionPool.check_connection,
|
||||
kwargs={"autocommit": True, "prepare_threshold": 0, "row_factory": dict_row},
|
||||
open=False,
|
||||
)
|
||||
await pool.open(wait=True)
|
||||
try:
|
||||
saver = AsyncPostgresSaver(pool)
|
||||
await saver.setup()
|
||||
yield saver
|
||||
finally:
|
||||
await pool.close()
|
||||
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def _sqlite_checkpointer_from_conn_str(raw: str) -> AsyncIterator[Checkpointer]:
|
||||
try:
|
||||
@@ -96,17 +142,17 @@ async def _sqlite_checkpointer_from_conn_str(raw: str) -> AsyncIterator[Checkpoi
|
||||
# ---------------------------------------------------------------------------
|
||||
# Async factory
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def _async_checkpointer(config) -> AsyncIterator[Checkpointer]:
|
||||
"""Async context manager that constructs and tears down a checkpointer."""
|
||||
if config.type == "memory":
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
|
||||
yield InMemorySaver()
|
||||
return
|
||||
|
||||
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def _async_checkpointer(config) -> AsyncIterator[Checkpointer]:
|
||||
"""Async context manager that constructs and tears down a checkpointer."""
|
||||
if config.type == "memory":
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
|
||||
yield InMemorySaver()
|
||||
return
|
||||
|
||||
if config.type == "sqlite":
|
||||
async with _sqlite_checkpointer_from_conn_str(config.connection_string or "store.db") as saver:
|
||||
yield saver
|
||||
@@ -121,15 +167,14 @@ async def _async_checkpointer(config) -> AsyncIterator[Checkpointer]:
|
||||
return
|
||||
|
||||
try:
|
||||
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
|
||||
except ImportError as exc:
|
||||
raise ImportError(POSTGRES_INSTALL) from exc
|
||||
|
||||
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver # noqa: F401
|
||||
except ImportError as exc:
|
||||
raise ImportError(POSTGRES_INSTALL) from exc
|
||||
|
||||
if not config.connection_string:
|
||||
raise ValueError(POSTGRES_CONN_REQUIRED)
|
||||
|
||||
async with AsyncPostgresSaver.from_conn_string(config.connection_string) as saver:
|
||||
await saver.setup()
|
||||
async with _postgres_saver_from_pool(config.connection_string) as saver:
|
||||
if not _dual_read_enabled(config):
|
||||
yield saver
|
||||
return
|
||||
@@ -172,87 +217,86 @@ async def _async_checkpointer(config) -> AsyncIterator[Checkpointer]:
|
||||
finally:
|
||||
await proxy.aclose_sources()
|
||||
return
|
||||
|
||||
raise ValueError(f"Unknown checkpointer type: {config.type!r}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public async context manager
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def _async_checkpointer_from_database(db_config) -> AsyncIterator[Checkpointer]:
|
||||
"""Async context manager that constructs a checkpointer from unified DatabaseConfig."""
|
||||
if db_config.backend == "memory":
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
|
||||
yield InMemorySaver()
|
||||
return
|
||||
|
||||
if db_config.backend == "sqlite":
|
||||
try:
|
||||
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
|
||||
except ImportError as exc:
|
||||
raise ImportError(SQLITE_INSTALL) from exc
|
||||
|
||||
conn_str = db_config.checkpointer_sqlite_path
|
||||
ensure_sqlite_parent_dir(conn_str)
|
||||
async with AsyncSqliteSaver.from_conn_string(conn_str) as saver:
|
||||
await saver.setup()
|
||||
yield saver
|
||||
return
|
||||
|
||||
if db_config.backend == "postgres":
|
||||
try:
|
||||
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver
|
||||
except ImportError as exc:
|
||||
raise ImportError(POSTGRES_INSTALL) from exc
|
||||
|
||||
if not db_config.postgres_url:
|
||||
raise ValueError("database.postgres_url is required for the postgres backend")
|
||||
|
||||
async with AsyncPostgresSaver.from_conn_string(db_config.postgres_url) as saver:
|
||||
await saver.setup()
|
||||
yield saver
|
||||
return
|
||||
|
||||
raise ValueError(f"Unknown database backend: {db_config.backend!r}")
|
||||
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def make_checkpointer(app_config: AppConfig | None = None) -> AsyncIterator[Checkpointer]:
|
||||
"""Async context manager that yields a checkpointer for the caller's lifetime.
|
||||
Resources are opened on enter and closed on exit -- no global state::
|
||||
|
||||
async with make_checkpointer(app_config) as checkpointer:
|
||||
app.state.checkpointer = checkpointer
|
||||
|
||||
Yields an ``InMemorySaver`` when no checkpointer is configured in *config.yaml*.
|
||||
|
||||
Priority:
|
||||
1. Legacy ``checkpointer:`` config section (backward compatible)
|
||||
2. Unified ``database:`` config section
|
||||
3. Default InMemorySaver
|
||||
"""
|
||||
|
||||
if app_config is None:
|
||||
app_config = get_app_config()
|
||||
|
||||
# Legacy: standalone checkpointer config takes precedence
|
||||
if app_config.checkpointer is not None:
|
||||
async with _async_checkpointer(app_config.checkpointer) as saver:
|
||||
yield saver
|
||||
return
|
||||
|
||||
# Unified database config
|
||||
db_config = getattr(app_config, "database", None)
|
||||
if db_config is not None and db_config.backend != "memory":
|
||||
async with _async_checkpointer_from_database(db_config) as saver:
|
||||
yield saver
|
||||
return
|
||||
|
||||
# Default: in-memory
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
|
||||
yield InMemorySaver()
|
||||
|
||||
raise ValueError(f"Unknown checkpointer type: {config.type!r}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Public async context manager
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def _async_checkpointer_from_database(db_config) -> AsyncIterator[Checkpointer]:
|
||||
"""Async context manager that constructs a checkpointer from unified DatabaseConfig."""
|
||||
if db_config.backend == "memory":
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
|
||||
yield InMemorySaver()
|
||||
return
|
||||
|
||||
if db_config.backend == "sqlite":
|
||||
try:
|
||||
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
|
||||
except ImportError as exc:
|
||||
raise ImportError(SQLITE_INSTALL) from exc
|
||||
|
||||
conn_str = db_config.checkpointer_sqlite_path
|
||||
ensure_sqlite_parent_dir(conn_str)
|
||||
async with AsyncSqliteSaver.from_conn_string(conn_str) as saver:
|
||||
await saver.setup()
|
||||
yield saver
|
||||
return
|
||||
|
||||
if db_config.backend == "postgres":
|
||||
try:
|
||||
from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver # noqa: F401
|
||||
except ImportError as exc:
|
||||
raise ImportError(POSTGRES_INSTALL) from exc
|
||||
|
||||
if not db_config.postgres_url:
|
||||
raise ValueError("database.postgres_url is required for the postgres backend")
|
||||
|
||||
async with _postgres_saver_from_pool(db_config.postgres_url) as saver:
|
||||
yield saver
|
||||
return
|
||||
|
||||
raise ValueError(f"Unknown database backend: {db_config.backend!r}")
|
||||
|
||||
|
||||
@contextlib.asynccontextmanager
|
||||
async def make_checkpointer(app_config: AppConfig | None = None) -> AsyncIterator[Checkpointer]:
|
||||
"""Async context manager that yields a checkpointer for the caller's lifetime.
|
||||
Resources are opened on enter and closed on exit -- no global state::
|
||||
|
||||
async with make_checkpointer(app_config) as checkpointer:
|
||||
app.state.checkpointer = checkpointer
|
||||
|
||||
Yields an ``InMemorySaver`` when no checkpointer is configured in *config.yaml*.
|
||||
|
||||
Priority:
|
||||
1. Legacy ``checkpointer:`` config section (backward compatible)
|
||||
2. Unified ``database:`` config section
|
||||
3. Default InMemorySaver
|
||||
"""
|
||||
|
||||
if app_config is None:
|
||||
app_config = get_app_config()
|
||||
|
||||
# Legacy: standalone checkpointer config takes precedence
|
||||
if app_config.checkpointer is not None:
|
||||
async with _async_checkpointer(app_config.checkpointer) as saver:
|
||||
yield saver
|
||||
return
|
||||
|
||||
# Unified database config
|
||||
db_config = getattr(app_config, "database", None)
|
||||
if db_config is not None and db_config.backend != "memory":
|
||||
async with _async_checkpointer_from_database(db_config) as saver:
|
||||
yield saver
|
||||
return
|
||||
|
||||
# Default: in-memory
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
|
||||
yield InMemorySaver()
|
||||
@@ -200,7 +200,7 @@ checkpointer:
|
||||
type: postgres
|
||||
# true: 新 checkpoint 写入 Postgres；false: 保留 PG 配置但继续使用 SQLite。
|
||||
# 也可临时用环境变量 DEER_FLOW_USE_PG_CHECKPOINTER=0 覆盖。
|
||||
use_postgres: true
|
||||
use_postgres: false
|
||||
connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow
|
||||
sqlite_connection_string: backend/.deer-flow/data/deerflow.db
|
||||
dual_read_enabled: true
|
||||
|
||||
Reference in new issue
Block a user