-
问答前缀配置
+
问答前缀配置
管理员可在此配置问答页输入框中的「前缀开关」。开启后，用户在新会话中第一次提问时，
会将下方配置的前缀文本和原问题拼接后再发送给模型。
diff --git a/frontend-web/src/pages/PublicLeaderboardPage.tsx b/frontend-web/src/pages/PublicLeaderboardPage.tsx
new file mode 100644
index 00000000..495afb98
--- /dev/null
+++ b/frontend-web/src/pages/PublicLeaderboardPage.tsx
@@ -0,0 +1,350 @@
+"use client";
+
+import { useEffect, useMemo, useState } from "react";
+import {
+ Activity,
+ Award,
+ Crown,
+ Gauge,
+ Loader2,
+ Medal,
+ MessageSquareText,
+ Sparkles,
+ TrendingUp,
+ Users,
+ type LucideIcon,
+} from "lucide-react";
+
+import type {
+ AdminLeaderboardResponse,
+ AdminLeaderboardSkill,
+ AdminLeaderboardUser,
+} from "@/core/admin/api";
+import { getBackendBaseURL } from "@/core/config";
+
+const RANK_COLORS = ["#C1121F", "#E11D48", "#F59E0B", "#991B1B", "#B45309"] as const;
+
+function fmtNumber(value: number | undefined | null): string {
+ const n = Number(value ?? 0);
+ if (!Number.isFinite(n)) return "0";
+ return n.toLocaleString("zh-CN");
+}
+
+function usernameOf(email: string | undefined | null): string {
+ if (!email) return "匿名用户";
+ const at = email.indexOf("@");
+ return at > 0 ? email.slice(0, at) : email;
+}
+
+function fmtDate(value: string | undefined | null): string {
+ if (!value) return "";
+ const d = new Date(value);
+ if (Number.isNaN(d.getTime())) return "";
+ return d.toLocaleString("zh-CN", { hour12: false });
+}
+
+async function fetchPublicLeaderboard(days: number): Promise {
+ const res = await fetch(
+ `${getBackendBaseURL()}/api/public/leaderboard?days=${days}&limit=20`,
+ { headers: { Accept: "application/json" } },
+ );
+ if (!res.ok) {
+ const err = (await res.json().catch(() => ({}))) as { detail?: string };
+ throw new Error(err.detail ?? `加载排行榜失败 (HTTP ${res.status})`);
+ }
+ return (await res.json()) as AdminLeaderboardResponse;
+}
+
+export default function PublicLeaderboardPage() {
+ const [data, setData] = useState(null);
+ const [error, setError] = useState(null);
+ const [loading, setLoading] = useState(true);
+ const days = 30;
+
+ useEffect(() => {
+ let cancelled = false;
+ setLoading(true);
+ fetchPublicLeaderboard(days)
+ .then((res) => {
+ if (!cancelled) {
+ setData(res);
+ setError(null);
+ }
+ })
+ .catch((e) => {
+ if (!cancelled) setError(e instanceof Error ? e.message : String(e));
+ })
+ .finally(() => {
+ if (!cancelled) setLoading(false);
+ });
+ return () => {
+ cancelled = true;
+ };
+ }, [days]);
+
+ const pending = data?.status === "running" || data?.status === "refreshing";
+
+ // The snapshot may still be computing on first load — poll briefly until ready.
+ useEffect(() => {
+ if (!pending) return;
+ const timer = setInterval(() => {
+ fetchPublicLeaderboard(days)
+ .then((res) => setData(res))
+ .catch(() => undefined);
+ }, 3000);
+ return () => clearInterval(timer);
+ }, [pending, days]);
+
+ return (
+
+
+ {/* 标题区：仅展示，不含任何操作按钮 */}
+
+
+ 用户排行榜
+
+ 展示近 {data?.range_days ?? days} 天用户活跃、问答、Token 消耗与技能热度。
+ {data?.since && data?.until && (
+
+ （{fmtDate(data.since).slice(0, 10)} ~ {fmtDate(data.until).slice(0, 10)}）
+
+ )}
+
+
+
+ {loading && !data && (
+
+
+ 正在加载排行榜数据
+
+ )}
+
+ {error && (
+
+ 加载失败：{error}
+
+ )}
+
+ {data && (
+
+ {pending && (
+
+
+ 排行榜正在后台统计，数据会自动刷新。
+
+ )}
+
+ {/* 概览指标 */}
+
+
+ {/* 三类用户排名 */}
+
+ u.run_count}
+ unit="次运行"
+ />
+ u.qa_count}
+ unit="次问答"
+ />
+ u.total_tokens}
+ unit="Token"
+ />
+
+
+ {/* 技能热度 + 高频问题 */}
+
+
+ )}
+
+
+
+
+
+
+ {label}
+
+
+ {value}
+
+ {sub &&
{sub}
}
+
+
+
+ {title}
+
+ {top.length === 0 ? (
+
暂无数据
+ ) : (
+
+ {top.map((user, index) => (
+ -
+
+
+ {usernameOf(user.email)}
+
+
+ {fmtNumber(metric(user))}
+
+ {unit}
+
+ ))}
+
+ )}
+
+
+
+ 技能热度
+
+ {top.length === 0 ? (
+
暂无数据
+ ) : (
+
+ )}
+
+
+
+ {title}
+
+ {top.length === 0 ? (
+
暂无数据
+ ) : (
+
+ {top.map((text, index) => (
+ -
+ {index + 1}.
+ {text}
+
+ ))}
+
+ )}
+
+
-
推荐问题配置
+
推荐问题配置
管理首页展示的推荐问题，点击后直接发送到对话。拖拽可调整顺序。
diff --git a/offline-backend-20260512/backend/CLAUDE.md b/offline-backend-20260512/backend/CLAUDE.md
index f6bfc4c9..9b467307 100644
--- a/offline-backend-20260512/backend/CLAUDE.md
+++ b/offline-backend-20260512/backend/CLAUDE.md
@@ -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.
diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py
index b3d2fa64..283a790f 100644
--- a/offline-backend-20260512/backend/app/gateway/app.py
+++ b/offline-backend-20260512/backend/app/gateway/app.py
@@ -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)
diff --git a/offline-backend-20260512/backend/app/gateway/routers/agents.py b/offline-backend-20260512/backend/app/gateway/routers/agents.py
index a75df4c8..40de50f3 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/agents.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/agents.py
@@ -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:
diff --git a/offline-backend-20260512/backend/app/gateway/routers/public_leaderboard.py b/offline-backend-20260512/backend/app/gateway/routers/public_leaderboard.py
new file mode 100644
index 00000000..b856d6e2
--- /dev/null
+++ b/offline-backend-20260512/backend/app/gateway/routers/public_leaderboard.py
@@ -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)
diff --git a/offline-backend-20260512/backend/app/gateway/routers/skills.py b/offline-backend-20260512/backend/app/gateway/routers/skills.py
index b1c9ffe8..8ad19a0c 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/skills.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/skills.py
@@ -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)
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/checkpointer/async_provider.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/checkpointer/async_provider.py
index 4f2e5142..d7fa14e5 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/checkpointer/async_provider.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/checkpointer/async_provider.py
@@ -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()
diff --git a/offline-backend-20260512/config.yaml b/offline-backend-20260512/config.yaml
index 751c0156..aa098e5c 100644
--- a/offline-backend-20260512/config.yaml
+++ b/offline-backend-20260512/config.yaml
@@ -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
diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml
new file mode 100644
index 00000000..6ec1aeb7
--- /dev/null
+++ b/pnpm-lock.yaml
@@ -0,0 +1,246 @@
+lockfileVersion: '9.0'
+
+settings:
+ autoInstallPeers: true
+ excludeLinksFromLockfile: false
+
+importers:
+
+ .:
+ dependencies:
+ axios:
+ specifier: ^1.16.1
+ version: 1.17.0
+
+packages:
+
+ agent-base@6.0.2:
+ resolution: {integrity: sha512-RZNwNclF7+MS/8bDg70amg32dyeZGZxiDuQmZxKLAlQjr3jGyLx+4Kkk58UO7D2QdgFIQCovuSuZESne6RG6XQ==}
+ engines: {node: '>= 6.0.0'}
+
+ asynckit@0.4.0:
+ resolution: {integrity: sha512-Oei9OH4tRh0YqU3GxhX79dM/mwVgvbZJaSNaRk+bshkj0S5cfHcgYakreBjrHwatXKbz+IoIdYLxrKim2MjW0Q==}
+
+ axios@1.17.0:
+ resolution: {integrity: sha512-J8SwNxprqqpbfenehxWYXE7CW+wM1BB4w3+N+g+/Wx40xM4rsLrfPmHHxSWIxJLYDgSY/HqlFPIYb2/S3rxafw==}
+
+ call-bind-apply-helpers@1.0.2:
+ resolution: {integrity: sha512-Sp1ablJ0ivDkSzjcaJdxEunN5/XvksFJ2sMBFfq6x0ryhQV/2b/KwFe21cMpmHtPOSij8K99/wSfoEuTObmuMQ==}
+ engines: {node: '>= 0.4'}
+
+ combined-stream@1.0.8:
+ resolution: {integrity: sha512-FQN4MRfuJeHf7cBbBMJFXhKSDq+2kAArBlmRBvcvFE5BB1HZKXtSFASDhdlz9zOYwxh8lDdnvmMOe/+5cdoEdg==}
+ engines: {node: '>= 0.8'}
+
+ debug@4.4.3:
+ resolution: {integrity: sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==}
+ engines: {node: '>=6.0'}
+ peerDependencies:
+ supports-color: '*'
+ peerDependenciesMeta:
+ supports-color:
+ optional: true
+
+ delayed-stream@1.0.0:
+ resolution: {integrity: sha512-ZySD7Nf91aLB0RxL4KGrKHBXl7Eds1DAmEdcoVawXnLD7SDhpNgtuII2aAkg7a7QS41jxPSZ17p4VdGnMHk3MQ==}
+ engines: {node: '>=0.4.0'}
+
+ dunder-proto@1.0.1:
+ resolution: {integrity: sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A==}
+ engines: {node: '>= 0.4'}
+
+ es-define-property@1.0.1:
+ resolution: {integrity: sha512-e3nRfgfUZ4rNGL232gUgX06QNyyez04KdjFrF+LTRoOXmrOgFKDg4BCdsjW8EnT69eqdYGmRpJwiPVYNrCaW3g==}
+ engines: {node: '>= 0.4'}
+
+ es-errors@1.3.0:
+ resolution: {integrity: sha512-Zf5H2Kxt2xjTvbJvP2ZWLEICxA6j+hAmMzIlypy4xcBg1vKVnx89Wy0GbS+kf5cwCVFFzdCFh2XSCFNULS6csw==}
+ engines: {node: '>= 0.4'}
+
+ es-object-atoms@1.1.2:
+ resolution: {integrity: sha512-HWcBoN6NileqtSydK2FqHbS/LoDd2pqrnQHLyJzBj4kOp/ky2MWMN694xOfkK8/SnUsW2DH7EfyVlydKCsm1Zw==}
+ engines: {node: '>= 0.4'}
+
+ es-set-tostringtag@2.1.0:
+ resolution: {integrity: sha512-j6vWzfrGVfyXxge+O0x5sh6cvxAog0a/4Rdd2K36zCMV5eJ+/+tOAngRO8cODMNWbVRdVlmGZQL2YS3yR8bIUA==}
+ engines: {node: '>= 0.4'}
+
+ follow-redirects@1.16.0:
+ resolution: {integrity: sha512-y5rN/uOsadFT/JfYwhxRS5R7Qce+g3zG97+JrtFZlC9klX/W5hD7iiLzScI4nZqUS7DNUdhPgw4xI8W2LuXlUw==}
+ engines: {node: '>=4.0'}
+ peerDependencies:
+ debug: '*'
+ peerDependenciesMeta:
+ debug:
+ optional: true
+
+ form-data@4.0.5:
+ resolution: {integrity: sha512-8RipRLol37bNs2bhoV67fiTEvdTrbMUYcFTiy3+wuuOnUog2QBHCZWXDRijWQfAkhBj2Uf5UnVaiWwA5vdd82w==}
+ engines: {node: '>= 6'}
+
+ function-bind@1.1.2:
+ resolution: {integrity: sha512-7XHNxH7qX9xG5mIwxkhumTox/MIRNcOgDrxWsMt2pAr23WHp6MrRlN7FBSFpCpr+oVO0F744iUgR82nJMfG2SA==}
+
+ get-intrinsic@1.3.0:
+ resolution: {integrity: sha512-9fSjSaos/fRIVIp+xSJlE6lfwhES7LNtKaCBIamHsjr2na1BiABJPo0mOjjz8GJDURarmCPGqaiVg5mfjb98CQ==}
+ engines: {node: '>= 0.4'}
+
+ get-proto@1.0.1:
+ resolution: {integrity: sha512-sTSfBjoXBp89JvIKIefqw7U2CCebsc74kiY6awiGogKtoSGbgjYE/G/+l9sF3MWFPNc9IcoOC4ODfKHfxFmp0g==}
+ engines: {node: '>= 0.4'}
+
+ gopd@1.2.0:
+ resolution: {integrity: sha512-ZUKRh6/kUFoAiTAtTYPZJ3hw9wNxx+BIBOijnlG9PnrJsCcSjs1wyyD6vJpaYtgnzDrKYRSqf3OO6Rfa93xsRg==}
+ engines: {node: '>= 0.4'}
+
+ has-symbols@1.1.0:
+ resolution: {integrity: sha512-1cDNdwJ2Jaohmb3sg4OmKaMBwuC48sYni5HUw2DvsC8LjGTLK9h+eb1X6RyuOHe4hT0ULCW68iomhjUoKUqlPQ==}
+ engines: {node: '>= 0.4'}
+
+ has-tostringtag@1.0.2:
+ resolution: {integrity: sha512-NqADB8VjPFLM2V0VvHUewwwsw0ZWBaIdgo+ieHtK3hasLz4qeCRjYcqfB6AQrBggRKppKF8L52/VqdVsO47Dlw==}
+ engines: {node: '>= 0.4'}
+
+ hasown@2.0.4:
+ resolution: {integrity: sha512-T2UbfbBEF32wiepXIsMlTW9+dDYC6wMh/t/vYA4tuOMKqWz/n3vr1NFSxQiyP+zk2mXsoMA/i/7qV6LKut1t1A==}
+ engines: {node: '>= 0.4'}
+
+ https-proxy-agent@5.0.1:
+ resolution: {integrity: sha512-dFcAjpTQFgoLMzC2VwU+C/CbS7uRL0lWmxDITmqm7C+7F0Odmj6s9l6alZc6AELXhrnggM2CeWSXHGOdX2YtwA==}
+ engines: {node: '>= 6'}
+
+ math-intrinsics@1.1.0:
+ resolution: {integrity: sha512-/IXtbwEk5HTPyEwyKX6hGkYXxM9nbj64B+ilVJnC/R6B0pH5G4V3b0pVbL7DBj4tkhBAppbQUlf6F6Xl9LHu1g==}
+ engines: {node: '>= 0.4'}
+
+ mime-db@1.52.0:
+ resolution: {integrity: sha512-sPU4uV7dYlvtWJxwwxHD0PuihVNiE7TyAbQ5SWxDCB9mUYvOgroQOwYQQOKPJ8CIbE+1ETVlOoK1UC2nU3gYvg==}
+ engines: {node: '>= 0.6'}
+
+ mime-types@2.1.35:
+ resolution: {integrity: sha512-ZDY+bPm5zTTF+YpCrAU9nK0UgICYPT0QtT1NZWFv4s++TNkcgVaT0g6+4R2uI4MjQjzysHB1zxuWL50hzaeXiw==}
+ engines: {node: '>= 0.6'}
+
+ ms@2.1.3:
+ resolution: {integrity: sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==}
+
+ proxy-from-env@2.1.0:
+ resolution: {integrity: sha512-cJ+oHTW1VAEa8cJslgmUZrc+sjRKgAKl3Zyse6+PV38hZe/V6Z14TbCuXcan9F9ghlz4QrFr2c92TNF82UkYHA==}
+ engines: {node: '>=10'}
+
+snapshots:
+
+ agent-base@6.0.2:
+ dependencies:
+ debug: 4.4.3
+ transitivePeerDependencies:
+ - supports-color
+
+ asynckit@0.4.0: {}
+
+ axios@1.17.0:
+ dependencies:
+ follow-redirects: 1.16.0
+ form-data: 4.0.5
+ https-proxy-agent: 5.0.1
+ proxy-from-env: 2.1.0
+ transitivePeerDependencies:
+ - debug
+ - supports-color
+
+ call-bind-apply-helpers@1.0.2:
+ dependencies:
+ es-errors: 1.3.0
+ function-bind: 1.1.2
+
+ combined-stream@1.0.8:
+ dependencies:
+ delayed-stream: 1.0.0
+
+ debug@4.4.3:
+ dependencies:
+ ms: 2.1.3
+
+ delayed-stream@1.0.0: {}
+
+ dunder-proto@1.0.1:
+ dependencies:
+ call-bind-apply-helpers: 1.0.2
+ es-errors: 1.3.0
+ gopd: 1.2.0
+
+ es-define-property@1.0.1: {}
+
+ es-errors@1.3.0: {}
+
+ es-object-atoms@1.1.2:
+ dependencies:
+ es-errors: 1.3.0
+
+ es-set-tostringtag@2.1.0:
+ dependencies:
+ es-errors: 1.3.0
+ get-intrinsic: 1.3.0
+ has-tostringtag: 1.0.2
+ hasown: 2.0.4
+
+ follow-redirects@1.16.0: {}
+
+ form-data@4.0.5:
+ dependencies:
+ asynckit: 0.4.0
+ combined-stream: 1.0.8
+ es-set-tostringtag: 2.1.0
+ hasown: 2.0.4
+ mime-types: 2.1.35
+
+ function-bind@1.1.2: {}
+
+ get-intrinsic@1.3.0:
+ dependencies:
+ call-bind-apply-helpers: 1.0.2
+ es-define-property: 1.0.1
+ es-errors: 1.3.0
+ es-object-atoms: 1.1.2
+ function-bind: 1.1.2
+ get-proto: 1.0.1
+ gopd: 1.2.0
+ has-symbols: 1.1.0
+ hasown: 2.0.4
+ math-intrinsics: 1.1.0
+
+ get-proto@1.0.1:
+ dependencies:
+ dunder-proto: 1.0.1
+ es-object-atoms: 1.1.2
+
+ gopd@1.2.0: {}
+
+ has-symbols@1.1.0: {}
+
+ has-tostringtag@1.0.2:
+ dependencies:
+ has-symbols: 1.1.0
+
+ hasown@2.0.4:
+ dependencies:
+ function-bind: 1.1.2
+
+ https-proxy-agent@5.0.1:
+ dependencies:
+ agent-base: 6.0.2
+ debug: 4.4.3
+ transitivePeerDependencies:
+ - supports-color
+
+ math-intrinsics@1.1.0: {}
+
+ mime-db@1.52.0: {}
+
+ mime-types@2.1.35:
+ dependencies:
+ mime-db: 1.52.0
+
+ ms@2.1.3: {}
+
+ proxy-from-env@2.1.0: {}