feat(监控): 管理员排行榜实时并发监控 + 真正停止对话 + 409 中文化

- 排行榜页顶部新增「实时并发监控」面板（admin-only）：当前并发大模型调用数 +
  调用列表（用户/智能体/类型/已运行时长），每行「停止」+「全部停止」经
  RunManager.cancel 真正中断 LLM 流式调用，减轻大模型压力
- 并发量折线图（峰值/均值，按分钟/小时）：新增轻量后台采样器写入新表
  concurrency_samples（create_all 自动建表），可按分/时聚合
- 大模型出字速度折线图（tokens/秒，按分钟/小时）：复用 llm_call_metrics 聚合，
  便于定位是模型/供应商慢
- 后台采样可开关、默认关闭(opt-in)、运行时切换免重启、对线上零影响
  （单条极小 INSERT、保留期清理、查询有上限、全程 best-effort）
- 409 中文化：上一轮未结束又发新问题时返回中文提示，前端 onError 友好映射

后端：runtime/runs/manager.py(list_active)、persistence/concurrency(新模块)、
config/system_settings(ConcurrencyMonitorSettings)、gateway/routers/admin_active_runs、
app.py(采样后台任务)、services.py(409)。前端：core/admin/{active-runs,hooks}、
components/workspace/admin/concurrency-monitor、AdminUserLeaderboardPage、threads/hooks。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

offline-backend-20260512/backend/CLAUDE.md

@@ -297,6 +297,15 @@ Two on-demand/automatic refresh paths layer on top:
 - **Manual "????"** ? `POST /leaderboard/refresh-today` force-rebuilds *today's* snapshot **synchronously** (bounded by the same semaphore/timeout the scheduler uses), then returns the merged range. Lets an admin pull the latest real-time records into the stats without waiting for the TTL. Frontend: a "????" button on `AdminUserLeaderboardPage` ? `useRefreshTodayLeaderboard` (`core/admin/hooks.ts`), which primes + invalidates the `["admin-activity","leaderboard"]` query.
 - **Daily finalize of the previous day** ? once per day, on/after `ADMIN_LEADERBOARD_FINALIZE_HOUR` (local Beijing, default `1` = ?? 1 ?), the scheduler tick force-recomputes *yesterday* a single time so runs/tool-calls that only landed near midnight are fully captured. The decision is the pure `deerflow.persistence.admin_stats.finalize.should_finalize_previous_day` (status/generated_at/now/hour) ? the rebuild stamps `generated_at` to today, which self-debounces it for the rest of the day (no extra state file). Tests: `tests/test_leaderboard_finalize.py`.
 
+### Admin Concurrency Monitor (实时并发监控 / 系统压力)
+
+A live admin monitor embedded **at the top of `AdminUserLeaderboardPage`** (`components/workspace/admin/concurrency-monitor.tsx`, `ConcurrencyMonitorPanel`) that answers "当前有几个并发的大模型调用、是谁、能不能直接停掉" and charts system pressure over time. Built on the existing in-memory `RunManager` (no new run plumbing).
+
+- **Live concurrency + 真正停止** — `RunManager.list_active()` returns every run whose status is `pending`/`running` (the registry keeps finished records around until `cleanup`, so it filters by status). `len(list_active())` is the current concurrent-call count. Router `app/gateway/routers/admin_active_runs.py` (`/api/admin/active-runs`, **admin only** via the shared `admin_users._require_admin`): `GET /` returns `{count, server_time, runs:[{run_id, thread_id, status, created_at, elapsed_seconds, user_id, user_name, agent_id, agent_name, thread_title, kind, multitask_strategy}]}` — each run best-effort enriched (batched DB) with the owning user's email (`threads_meta.user_id`→`users.email`) and the agent's display name (`agents.name`); `kind` is derived from thread metadata (圆桌会商 / 系统/后台 / 任务对话 / 对话). `POST /{run_id}/cancel` calls `RunManager.cancel(run_id, action="interrupt")`, which sets the abort event **and cancels the run's asyncio task** — i.e. it truly aborts the in-flight model stream, freeing capacity. `POST /cancel-all` stops every in-flight run.
+- **并发量折线图 (按分钟/小时)** — a lightweight background **sampler** (`_concurrency_sampler_loop` in `app/gateway/app.py`, singleton file-locked, started in the lifespan) records `len(list_active())` every `CONCURRENCY_SAMPLE_INTERVAL_SECONDS` (default 15s) into the new `concurrency_samples` table (`deerflow.persistence.concurrency`, `ConcurrencySampleRow{sampled_at, active_count}`, auto-created by `create_all`, **no migration**; store wired as `app.state.concurrency_sample_store`). ~Hourly it purges rows older than `CONCURRENCY_SAMPLE_RETENTION_HOURS` (default 168h). `GET /concurrency?granularity=minute|hour&hours=N` aggregates samples into per-bucket **peak + avg** (`ConcurrencySampleStore.aggregate`, bucketed in Python so it's identical across sqlite/mysql/postgres; fetch capped at 60k rows). Default lookback: minute→3h, hour→48h.
+- **大模型出字速度折线图 (tokens/sec, 按分钟/小时)** — `GET /token-speed?granularity=&hours=` aggregates the persisted `llm_call_metrics` (`status="success"`) into per-bucket avg/min/max tokens/sec (deriving tokens/sec from `output_tokens÷duration_ms` when the stored value is null). A dip pinpoints the model/provider as the bottleneck for that window — "知道具体原因出现在哪里".
+- **On/off + 不影响线上** — the whole feature is designed to never impact production: sampling is one tiny INSERT, retention-purged, all paths best-effort/guarded, queries bounded. A **runtime switch** lives in `system_settings.json → concurrency_monitor.enabled` (`ConcurrencyMonitorSettings`, **default false / opt-in**, mtime-hot-reloaded): the background loop always runs (just sleeps) but only samples when an admin turns it on, and it re-reads the flag **every tick** so toggling it on/off **takes effect immediately without a restart** (the on-demand live snapshot + 停止 stay available regardless; only the historical chart needs sampling on). `GET|PUT /api/admin/active-runs/settings` reads/writes it (admin only); the panel's 「后台采样」 Switch drives it. Env `CONCURRENCY_SAMPLE_ENABLED=0` is a hard master kill (never starts the loop). Frontend client/hooks: `core/admin/active-runs.ts` + `core/admin/hooks.ts` (`useActiveRuns` polls 4s while the panel is open, `useConcurrencySeries`/`useTokenSpeedSeries` poll 15s, `useCancelActiveRun`/`useCancelAllActiveRuns`, `useMonitorSettings`/`useUpdateMonitorSettings`). Tests: `tests/test_concurrency_monitor.py`.
+
 ### Scheduled Task Runtime
 
 `ScheduledTaskService` (`deerflow/runtime/scheduler/service.py`) polls due tasks every `poll_interval_seconds` and executes each in its own throwaway thread, serialized by a shared `_agent_run_lock`.

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

@@ -27,6 +27,7 @@ from app.gateway.csrf_middleware import CSRFMiddleware
 from app.gateway.deps import langgraph_runtime
 from app.gateway.routers import (
     ai_writing,
+    admin_active_runs,
     admin_users,
     agents,
     artifacts,
@@ -488,6 +489,18 @@ _LEADERBOARD_SNAPSHOT_STARTUP_DELAY_SECONDS = _env_float(
     "ADMIN_LEADERBOARD_BACKGROUND_STARTUP_DELAY_SECONDS",
     60.0,
 )
+# Concurrency (system-pressure) sampler: how often to snapshot the in-flight run
+# count, and how long to keep samples (drives the admin 并发量 line chart, which
+# aggregates per minute/hour). Sampling is in-memory cheap (one tiny INSERT).
+_CONCURRENCY_SAMPLE_ENABLED = _env_bool("CONCURRENCY_SAMPLE_ENABLED", True)
+_CONCURRENCY_SAMPLE_INTERVAL_SECONDS = _env_float(
+    "CONCURRENCY_SAMPLE_INTERVAL_SECONDS",
+    15.0,
+)
+_CONCURRENCY_SAMPLE_RETENTION_HOURS = _env_float(
+    "CONCURRENCY_SAMPLE_RETENTION_HOURS",
+    168.0,  # 7 days
+)
 # Hour (local Beijing time, 0-23) after which the scheduler force-recomputes the
 # *previous* day's snapshot once, so records that landed near midnight are fully
 # captured. Default 1 ? "???? 1 ????????".
@@ -828,6 +841,58 @@ async def _leaderboard_snapshot_scheduler_loop() -> None:
         _release_singleton_lock(lock_handle)
 
 
+async def _concurrency_sampler_loop(app: FastAPI) -> None:
+    """Periodically record the in-flight run count for the admin pressure chart.
+
+    Runs in the gateway process, singleton-locked so only one worker samples.
+    Each tick records ``len(run_manager.list_active())`` and, roughly hourly,
+    purges samples older than the retention window. Best-effort throughout — a
+    failed tick is logged and the loop continues.
+    """
+    if not _CONCURRENCY_SAMPLE_ENABLED:
+        logger.info("Concurrency sampler disabled by CONCURRENCY_SAMPLE_ENABLED")
+        return
+
+    store = getattr(app.state, "concurrency_sample_store", None)
+    run_mgr = getattr(app.state, "run_manager", None)
+    if store is None or run_mgr is None:
+        logger.info("Concurrency sampler not started (no store / run manager — memory backend?)")
+        return
+
+    from deerflow.config.runtime_paths import runtime_home
+
+    lock_handle = _acquire_singleton_lock(runtime_home() / ".concurrency_sampler.lock")
+    if lock_handle is None:
+        logger.info("Concurrency sampler: another gateway worker holds the lock; this worker skips")
+        return
+
+    interval = max(1.0, _CONCURRENCY_SAMPLE_INTERVAL_SECONDS)
+    purge_every_ticks = max(1, int(3600 / interval))  # ~once per hour
+    logger.info("Concurrency sampler started: interval=%.1fs retention=%.1fh", interval, _CONCURRENCY_SAMPLE_RETENTION_HOURS)
+    ticks = 0
+    try:
+        while True:
+            try:
+                # Runtime toggle (admin can turn sampling off without a restart).
+                # mtime-cached read, so this is cheap every tick.
+                from deerflow.config.system_settings import get_concurrency_monitor_settings
+
+                if get_concurrency_monitor_settings().enabled:
+                    active = await run_mgr.list_active()
+                    await store.record(len(active))
+                    ticks += 1
+                    if ticks % purge_every_ticks == 0:
+                        from datetime import UTC, datetime, timedelta
+
+                        cutoff = datetime.now(UTC) - timedelta(hours=_CONCURRENCY_SAMPLE_RETENTION_HOURS)
+                        await store.purge_older_than(cutoff)
+            except Exception:
+                logger.exception("Concurrency sampler tick failed (non-fatal)")
+            await asyncio.sleep(interval)
+    finally:
+        _release_singleton_lock(lock_handle)
+
+
 def _warn_if_multi_worker() -> None:
     """?? uvicorn / gunicorn ? worker ???? ERROR ???
 
@@ -1044,6 +1109,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         except Exception:
             logger.exception("Leaderboard snapshot scheduler failed to start (non-fatal)")
 
+        # Concurrency (system-pressure) sampler — lightweight, singleton-locked,
+        # runtime-toggleable via system_settings.concurrency_monitor.enabled.
+        try:
+            app.state.concurrency_sampler_task = asyncio.create_task(_concurrency_sampler_loop(app))
+        except Exception:
+            logger.exception("Concurrency sampler failed to start (non-fatal)")
+
         # Start IM channel service if any channels are configured
         try:
             from app.channels.service import start_channel_service
@@ -1102,6 +1174,16 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
             except Exception:
                 logger.exception("Leaderboard snapshot scheduler shutdown error")
 
+        concurrency_sampler_task = getattr(app.state, "concurrency_sampler_task", None)
+        if concurrency_sampler_task is not None:
+            concurrency_sampler_task.cancel()
+            try:
+                await concurrency_sampler_task
+            except asyncio.CancelledError:
+                pass
+            except Exception:
+                logger.exception("Concurrency sampler shutdown error")
+
         # ?? 7 ????AI ?? graph ???? checkpointer / in-flight task?
         # thread ??? LangGraph Server ? checkpointer ???lifespan ???
         # ``langgraph_runtime()`` ? contextmanager ???????????????
@@ -1434,6 +1516,8 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     app.include_router(admin_users.router)
     app.include_router(admin_users.self_router)
 
+    app.include_router(admin_active_runs.router)
+
     # Admin tool / skill call metrics audit + export
     app.include_router(tool_metrics.router)
 

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

@@ -116,6 +116,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
         app.state.llm_metrics_store = make_llm_metrics_store(sf)
 
+        from deerflow.persistence.concurrency import make_concurrency_sample_store
+
+        app.state.concurrency_sample_store = make_concurrency_sample_store(sf)
+
         from deerflow.persistence.tool_metrics import make_tool_metrics_store
 
         app.state.tool_metrics_store = make_tool_metrics_store(sf)
@@ -433,6 +437,11 @@ def get_llm_metrics_store(request: Request):
     return getattr(request.app.state, "llm_metrics_store", None)
 
 
+def get_concurrency_sample_store(request: Request):
+    """Return the concurrency-sample store, or None on the memory backend."""
+    return getattr(request.app.state, "concurrency_sample_store", None)
+
+
 def get_tool_metrics_store(request: Request):
     """Return the tool-call metrics store, or None when not configured."""
     return getattr(request.app.state, "tool_metrics_store", None)

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

@@ -0,0 +1,406 @@
+"""Admin-only live concurrency monitor for in-flight model runs.
+
+Surfaces every *currently running* call into the agent runtime (interactive
+chat, agent chat, roundtable seats, …) so an administrator can see how much
+concurrent pressure the system is putting on the LLM and **truly stop** any of
+them — ``RunManager.cancel`` aborts the run's asyncio task, which cancels the
+in-flight model stream, immediately freeing that capacity.
+
+All endpoints require ``system_role == "admin"`` (shared ``_require_admin``).
+The data source is the in-memory :class:`RunManager` registry (``list_active``),
+enriched best-effort with the owning user's email + the agent's display name
+from the DB. Enrichment failures degrade gracefully (the run still shows, just
+with rawer labels) — the monitor must never break because a lookup failed.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import UTC, datetime, timedelta
+
+from fastapi import APIRouter, Query, Request
+from pydantic import BaseModel
+from sqlalchemy import select
+
+from app.gateway.deps import get_concurrency_sample_store, get_run_manager
+from app.gateway.routers.admin_users import _require_admin
+from deerflow.config.system_settings import load_system_settings, save_system_settings
+from deerflow.persistence.agents.model import AgentRow
+from deerflow.persistence.engine import get_session_factory
+from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
+from deerflow.persistence.thread_meta.model import ThreadMetaRow
+from deerflow.persistence.user.model import UserRow
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/admin/active-runs", tags=["admin-active-runs"])
+
+# Lookback windows are clamped so an aggregation never scans an unbounded range.
+_MAX_LOOKBACK_HOURS = 24 * 14  # 14 days
+
+# Per-granularity default lookback (hours): minute view = recent, hour view = wider.
+_DEFAULT_LOOKBACK_HOURS = {"minute": 3, "hour": 48}
+
+# assistant_id values that are not real custom-agent ids (the default lead agent).
+_DEFAULT_ASSISTANT_IDS = {"", "default", "lead_agent", "agent"}
+
+
+class ActiveRunItem(BaseModel):
+    run_id: str
+    thread_id: str
+    status: str
+    created_at: str = ""
+    elapsed_seconds: int = 0
+    user_id: str = ""
+    user_name: str = ""
+    agent_id: str = ""
+    agent_name: str = ""
+    thread_title: str = ""
+    kind: str = "对话"
+    multitask_strategy: str = "reject"
+
+
+class ActiveRunsResponse(BaseModel):
+    count: int
+    server_time: str
+    runs: list[ActiveRunItem]
+
+
+class CancelRunResponse(BaseModel):
+    run_id: str
+    cancelled: bool
+
+
+class CancelAllResponse(BaseModel):
+    cancelled_count: int
+
+
+class ConcurrencyPoint(BaseModel):
+    bucket: str
+    peak: int
+    avg: float
+    samples: int
+
+
+class ConcurrencySeriesResponse(BaseModel):
+    granularity: str
+    since: str
+    until: str
+    enabled: bool
+    points: list[ConcurrencyPoint]
+
+
+class TokenSpeedPoint(BaseModel):
+    bucket: str
+    avg_tokens_per_sec: float
+    min_tokens_per_sec: float
+    max_tokens_per_sec: float
+    samples: int
+
+
+class TokenSpeedSeriesResponse(BaseModel):
+    granularity: str
+    since: str
+    until: str
+    points: list[TokenSpeedPoint]
+
+
+class MonitorSettingsResponse(BaseModel):
+    enabled: bool
+
+
+def _resolve_window(granularity: str, hours: float | None) -> tuple[str, datetime, datetime]:
+    """Clamp the granularity + lookback window into (gran, since, until)."""
+    gran = "hour" if granularity == "hour" else "minute"
+    lookback = hours if hours and hours > 0 else _DEFAULT_LOOKBACK_HOURS[gran]
+    lookback = min(float(lookback), _MAX_LOOKBACK_HOURS)
+    until = datetime.now(UTC)
+    since = until - timedelta(hours=lookback)
+    return gran, since, until
+
+
+def _bucket_start(dt: datetime, granularity: str) -> datetime:
+    if granularity == "hour":
+        return dt.replace(minute=0, second=0, microsecond=0)
+    return dt.replace(second=0, microsecond=0)
+
+
+def _derive_tokens_per_sec(tps, output_tokens, total_tokens, duration_ms) -> float | None:
+    """Use the stored tokens/sec, else derive from tokens ÷ duration."""
+    if tps not in (None, 0):
+        return float(tps)
+    numerator = output_tokens if output_tokens else total_tokens
+    if not numerator or not duration_ms or duration_ms <= 0:
+        return None
+    return round((float(numerator) / float(duration_ms)) * 1000.0, 3)
+
+
+def _elapsed_seconds(created_at: str, now: datetime) -> int:
+    """Seconds since the (UTC ISO) ``created_at`` stamp; 0 on any parse error."""
+    if not created_at:
+        return 0
+    try:
+        started = datetime.fromisoformat(created_at)
+    except (ValueError, TypeError):
+        return 0
+    if started.tzinfo is None:
+        started = started.replace(tzinfo=UTC)
+    delta = (now - started).total_seconds()
+    return max(0, int(delta))
+
+
+def _kind_for(metadata: dict | None) -> str:
+    """Coarse human label for the run's origin, from its thread metadata."""
+    meta = metadata or {}
+    thread_type = str(meta.get("thread_type") or "")
+    if thread_type == "roundtable":
+        return "圆桌会商"
+    if meta.get("system"):
+        return "系统/后台"
+    if meta.get("taskId"):
+        return "任务对话"
+    return "对话"
+
+
+async def _enrich(records) -> dict:
+    """Batch-resolve thread metadata, user emails and agent names.
+
+    Returns a dict of lookup maps; every query is wrapped so a failure just
+    yields empty maps (the runs still render with raw ids).
+    """
+    thread_ids = {r.thread_id for r in records if r.thread_id}
+    assistant_ids = {
+        r.assistant_id
+        for r in records
+        if r.assistant_id and r.assistant_id not in _DEFAULT_ASSISTANT_IDS
+    }
+
+    thread_meta: dict[str, ThreadMetaRow] = {}
+    user_emails: dict[str, str] = {}
+    agent_names: dict[str, str] = {}
+
+    try:
+        session_factory = get_session_factory()
+        async with session_factory() as session:
+            if thread_ids:
+                rows = (
+                    await session.execute(
+                        select(ThreadMetaRow).where(ThreadMetaRow.thread_id.in_(thread_ids))
+                    )
+                ).scalars().all()
+                thread_meta = {row.thread_id: row for row in rows}
+
+            user_ids = {row.user_id for row in thread_meta.values() if row.user_id}
+            if user_ids:
+                rows = (
+                    await session.execute(
+                        select(UserRow.id, UserRow.email).where(UserRow.id.in_(user_ids))
+                    )
+                ).all()
+                user_emails = {r[0]: (r[1] or "") for r in rows}
+
+            if assistant_ids:
+                rows = (
+                    await session.execute(
+                        select(AgentRow.id, AgentRow.name).where(AgentRow.id.in_(assistant_ids))
+                    )
+                ).all()
+                agent_names = {r[0]: (r[1] or "") for r in rows}
+    except Exception:  # pragma: no cover - enrichment is best-effort
+        logger.warning("Failed to enrich active runs (non-fatal)", exc_info=True)
+
+    return {"thread_meta": thread_meta, "user_emails": user_emails, "agent_names": agent_names}
+
+
+@router.get("", response_model=ActiveRunsResponse)
+@router.get("/", response_model=ActiveRunsResponse)
+async def list_active_runs(request: Request) -> ActiveRunsResponse:
+    """List every in-flight run with who/what is driving it (admin only)."""
+    await _require_admin(request)
+    run_mgr = get_run_manager(request)
+    records = await run_mgr.list_active()
+    now = datetime.now(UTC)
+    maps = await _enrich(records)
+    thread_meta = maps["thread_meta"]
+    user_emails = maps["user_emails"]
+    agent_names = maps["agent_names"]
+
+    items: list[ActiveRunItem] = []
+    for r in records:
+        meta_row = thread_meta.get(r.thread_id)
+        user_id = (meta_row.user_id if meta_row else "") or ""
+        metadata = (meta_row.metadata_json if meta_row else None) or r.metadata or {}
+        assistant_id = r.assistant_id or ""
+        if assistant_id in _DEFAULT_ASSISTANT_IDS:
+            agent_name = "默认助手"
+        else:
+            agent_name = agent_names.get(assistant_id) or assistant_id
+        items.append(
+            ActiveRunItem(
+                run_id=r.run_id,
+                thread_id=r.thread_id,
+                status=r.status.value,
+                created_at=r.created_at,
+                elapsed_seconds=_elapsed_seconds(r.created_at, now),
+                user_id=user_id,
+                user_name=user_emails.get(user_id) or user_id or "未知用户",
+                agent_id=assistant_id,
+                agent_name=agent_name,
+                thread_title=(meta_row.display_name if meta_row else "") or "",
+                kind=_kind_for(metadata),
+                multitask_strategy=r.multitask_strategy,
+            )
+        )
+
+    return ActiveRunsResponse(count=len(items), server_time=now.isoformat(), runs=items)
+
+
+@router.post("/{run_id}/cancel", response_model=CancelRunResponse)
+async def cancel_active_run(
+    run_id: str,
+    request: Request,
+    action: str = Query(default="interrupt", description="interrupt 保留检查点 / rollback 回滚"),
+) -> CancelRunResponse:
+    """Truly stop one in-flight run — cancels its task and the live model stream."""
+    await _require_admin(request)
+    run_mgr = get_run_manager(request)
+    normalized = "rollback" if action == "rollback" else "interrupt"
+    cancelled = await run_mgr.cancel(run_id, action=normalized)
+    return CancelRunResponse(run_id=run_id, cancelled=cancelled)
+
+
+@router.post("/cancel-all", response_model=CancelAllResponse)
+async def cancel_all_active_runs(request: Request) -> CancelAllResponse:
+    """Stop every in-flight run at once (admin emergency relief valve)."""
+    await _require_admin(request)
+    run_mgr = get_run_manager(request)
+    records = await run_mgr.list_active()
+    cancelled = 0
+    for r in records:
+        if await run_mgr.cancel(r.run_id, action="interrupt"):
+            cancelled += 1
+    return CancelAllResponse(cancelled_count=cancelled)
+
+
+@router.get("/concurrency", response_model=ConcurrencySeriesResponse)
+async def concurrency_series(
+    request: Request,
+    granularity: str = Query(default="minute", description="minute | hour"),
+    hours: float | None = Query(default=None, description="回看小时数(默认按粒度取)"),
+) -> ConcurrencySeriesResponse:
+    """Per-minute / per-hour 并发量 series (peak + avg) for the pressure chart."""
+    await _require_admin(request)
+    gran, since, until = _resolve_window(granularity, hours)
+    enabled = load_system_settings().concurrency_monitor.enabled
+    store = get_concurrency_sample_store(request)
+    points: list[ConcurrencyPoint] = []
+    if store is not None:
+        buckets = await store.aggregate(since=since, until=until, granularity=gran)
+        points = [
+            ConcurrencyPoint(bucket=b.bucket, peak=b.peak, avg=b.avg, samples=b.samples)
+            for b in buckets
+        ]
+    return ConcurrencySeriesResponse(
+        granularity=gran,
+        since=since.isoformat(),
+        until=until.isoformat(),
+        enabled=enabled,
+        points=points,
+    )
+
+
+@router.get("/token-speed", response_model=TokenSpeedSeriesResponse)
+async def token_speed_series(
+    request: Request,
+    granularity: str = Query(default="minute", description="minute | hour"),
+    hours: float | None = Query(default=None, description="回看小时数(默认按粒度取)"),
+) -> TokenSpeedSeriesResponse:
+    """Per-minute / per-hour 大模型出字速度 (tokens/sec) series.
+
+    Aggregated from the persisted LLM-call metrics: a dip in tokens/sec for a
+    bucket points at the model/provider being the bottleneck for that window.
+    """
+    await _require_admin(request)
+    gran, since, until = _resolve_window(granularity, hours)
+
+    buckets: dict[datetime, dict[str, float]] = {}
+    try:
+        session_factory = get_session_factory()
+        if session_factory is not None:
+            async with session_factory() as session:
+                rows = (
+                    await session.execute(
+                        select(
+                            LlmCallMetricRow.created_at,
+                            LlmCallMetricRow.tokens_per_sec,
+                            LlmCallMetricRow.output_tokens,
+                            LlmCallMetricRow.total_tokens,
+                            LlmCallMetricRow.duration_ms,
+                        )
+                        .where(
+                            LlmCallMetricRow.created_at >= since,
+                            LlmCallMetricRow.created_at <= until,
+                            LlmCallMetricRow.status == "success",
+                        )
+                        .order_by(LlmCallMetricRow.created_at)
+                        .limit(100_000)
+                    )
+                ).all()
+            for created_at, tps_raw, output_tokens, total_tokens, duration_ms in rows:
+                tps = _derive_tokens_per_sec(tps_raw, output_tokens, total_tokens, duration_ms)
+                if tps is None or tps <= 0 or created_at is None:
+                    continue
+                if created_at.tzinfo is None:
+                    created_at = created_at.replace(tzinfo=UTC)
+                key = _bucket_start(created_at, gran)
+                entry = buckets.get(key)
+                if entry is None:
+                    buckets[key] = {"sum": tps, "min": tps, "max": tps, "count": 1}
+                else:
+                    entry["sum"] += tps
+                    entry["min"] = min(entry["min"], tps)
+                    entry["max"] = max(entry["max"], tps)
+                    entry["count"] += 1
+    except Exception:  # pragma: no cover - chart must degrade gracefully
+        logger.warning("Failed to aggregate token-speed series (non-fatal)", exc_info=True)
+
+    points: list[TokenSpeedPoint] = []
+    for key in sorted(buckets):
+        entry = buckets[key]
+        count = max(1, int(entry["count"]))
+        points.append(
+            TokenSpeedPoint(
+                bucket=key.isoformat(),
+                avg_tokens_per_sec=round(entry["sum"] / count, 2),
+                min_tokens_per_sec=round(entry["min"], 2),
+                max_tokens_per_sec=round(entry["max"], 2),
+                samples=count,
+            )
+        )
+    return TokenSpeedSeriesResponse(
+        granularity=gran,
+        since=since.isoformat(),
+        until=until.isoformat(),
+        points=points,
+    )
+
+
+@router.get("/settings", response_model=MonitorSettingsResponse)
+async def get_monitor_settings(request: Request) -> MonitorSettingsResponse:
+    """Read the concurrency-monitor on/off switch (admin only)."""
+    await _require_admin(request)
+    return MonitorSettingsResponse(enabled=load_system_settings().concurrency_monitor.enabled)
+
+
+@router.put("/settings", response_model=MonitorSettingsResponse)
+async def update_monitor_settings(
+    request: Request,
+    body: MonitorSettingsResponse,
+) -> MonitorSettingsResponse:
+    """Toggle the background concurrency sampler on/off (admin only, no restart)."""
+    await _require_admin(request)
+    settings = load_system_settings()
+    updated = settings.model_copy(
+        update={"concurrency_monitor": settings.concurrency_monitor.model_copy(update={"enabled": bool(body.enabled)})}
+    )
+    save_system_settings(updated)
+    return MonitorSettingsResponse(enabled=updated.concurrency_monitor.enabled)

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

@@ -581,7 +581,12 @@ async def start_run(
             multitask_strategy=body.multitask_strategy,
         )
     except ConflictError as exc:
-        raise HTTPException(status_code=409, detail=str(exc)) from exc
+        # 同一会话上一轮回答还没结束，又提交了新问题 → 拒绝。给用户中文提示，
+        # 而不是把内部英文「Thread ... already has an active run」直接抛到前端。
+        raise HTTPException(
+            status_code=409,
+            detail="当前对话还在生成回答，请等待本轮回复结束，或先点「停止」再发送新的问题。",
+        ) from exc
     except UnsupportedStrategyError as exc:
         raise HTTPException(status_code=501, detail=str(exc)) from exc
     t_post_create = time.perf_counter()

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

@@ -234,6 +234,21 @@ class LoginWhitelistSettings(BaseModel):
     model_config = ConfigDict(extra="ignore")
 
 
+class ConcurrencyMonitorSettings(BaseModel):
+    """管理员「实时并发监控」开关(排行榜页的系统压力监控)。
+
+    ``enabled`` 控制**后台并发采样器**是否持续把当前并发调用数写库(供并发量
+    折线图按分钟/小时聚合)。**默认关**(opt-in,由管理员在排行榜页手动开启),
+    以对线上服务零影响为先;开启后采样器才开始写库,关闭则立即停止(每 tick 都
+    重读该开关,免重启),实时快照查询与「停止」始终可用、零常驻成本。设计上即便
+    开启也对线上零影响:采样仅一条极小 INSERT、有保留期清理、聚合查询有行数上限、
+    全程 best-effort。
+    """
+
+    enabled: bool = Field(default=False, description="是否启用后台并发采样(默认关;开→开始写库,关→停止写库、折线图不再新增历史点)")
+    model_config = ConfigDict(extra="ignore")
+
+
 class SystemSettings(BaseModel):
     """Top-level system settings document."""
 
@@ -246,6 +261,7 @@ class SystemSettings(BaseModel):
     leaderboard: LeaderboardSettings = Field(default_factory=LeaderboardSettings)
     skill_compression: SkillCompressionSettings = Field(default_factory=SkillCompressionSettings)
     agent_modules: AgentModuleSettings = Field(default_factory=AgentModuleSettings)
+    concurrency_monitor: ConcurrencyMonitorSettings = Field(default_factory=ConcurrencyMonitorSettings)
     model_config = ConfigDict(extra="ignore")
 
 
@@ -280,6 +296,15 @@ def get_skill_compression_settings() -> SkillCompressionSettings:
         return SkillCompressionSettings()
 
 
+def get_concurrency_monitor_settings() -> ConcurrencyMonitorSettings:
+    """Convenience accessor for the concurrency-monitor block (safe defaults on error)."""
+    try:
+        return load_system_settings().concurrency_monitor
+    except Exception:
+        logger.debug("Failed to read concurrency monitor settings; using defaults", exc_info=True)
+        return ConcurrencyMonitorSettings()
+
+
 def save_system_settings(settings: SystemSettings) -> SystemSettings:
     """Persist the settings document atomically, returning the saved value."""
     path = _settings_path()

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

@@ -0,0 +1,15 @@
+"""Concurrency (system-pressure) sampling persistence."""
+
+from deerflow.persistence.concurrency.model import ConcurrencySampleRow
+from deerflow.persistence.concurrency.store import (
+    ConcurrencyBucket,
+    ConcurrencySampleStore,
+    make_concurrency_sample_store,
+)
+
+__all__ = [
+    "ConcurrencyBucket",
+    "ConcurrencySampleRow",
+    "ConcurrencySampleStore",
+    "make_concurrency_sample_store",
+]

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

@@ -0,0 +1,31 @@
+"""ORM model for periodic concurrency (system-pressure) samples.
+
+A background sampler writes one row every few seconds holding the number of
+in-flight model runs at that instant. The admin monitor aggregates these into
+per-minute / per-hour buckets to draw the 并发量 pressure line chart, even for
+time windows before the admin opened the page.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import Integer
+from sqlalchemy.orm import Mapped, mapped_column
+
+from deerflow.persistence.base import Base
+from deerflow.persistence.types import BeijingDateTime
+
+
+class ConcurrencySampleRow(Base):
+    """One sampled snapshot of concurrent in-flight model runs."""
+
+    __tablename__ = "concurrency_samples"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    # Wall-clock time the sample was taken (Beijing time on disk, tz-aware in Python).
+    sampled_at: Mapped[datetime] = mapped_column(
+        BeijingDateTime(), index=True, default=lambda: datetime.now(UTC)
+    )
+    # Number of runs whose status was pending/running at ``sampled_at``.
+    active_count: Mapped[int] = mapped_column(Integer, default=0)

offline-backend-20260512/backend/packages/harness/deerflow/persistence/concurrency/store.py

@@ -0,0 +1,142 @@
+"""Async store for concurrency samples + minute/hour aggregation."""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
+
+from deerflow.persistence.concurrency.model import ConcurrencySampleRow
+
+logger = logging.getLogger(__name__)
+
+# How far back the aggregation will fetch raw rows before bucketing. A safety
+# cap so a degenerate range never tries to load the whole table into memory.
+_MAX_ROWS = 60_000
+
+
+@dataclass
+class ConcurrencyBucket:
+    """One minute/hour bucket of the concurrency series."""
+
+    bucket: str  # ISO start of the bucket
+    peak: int  # max concurrent runs observed in the bucket
+    avg: float  # mean concurrent runs in the bucket
+    samples: int  # how many raw samples fell into the bucket
+
+
+def _bucket_start(dt: datetime, granularity: str) -> datetime:
+    """Truncate a timestamp to the start of its minute/hour bucket."""
+    if granularity == "hour":
+        return dt.replace(minute=0, second=0, microsecond=0)
+    return dt.replace(second=0, microsecond=0)
+
+
+class ConcurrencySampleStore:
+    """Persists periodic concurrency snapshots and aggregates them."""
+
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._sf = session_factory
+
+    async def record(self, active_count: int, *, sampled_at: datetime | None = None) -> None:
+        """Insert one sample. Best-effort — never raises into the caller."""
+        try:
+            async with self._sf() as session:
+                session.add(
+                    ConcurrencySampleRow(
+                        active_count=max(0, int(active_count)),
+                        sampled_at=sampled_at or datetime.now(UTC),
+                    )
+                )
+                await session.commit()
+        except Exception:  # pragma: no cover - sampling must never break the app
+            logger.warning("Failed to record concurrency sample", exc_info=True)
+
+    async def aggregate(
+        self,
+        *,
+        since: datetime,
+        until: datetime,
+        granularity: str = "minute",
+    ) -> list[ConcurrencyBucket]:
+        """Return per-minute / per-hour buckets of peak + average concurrency.
+
+        Bucketing is done in Python (not SQL ``date_trunc``/``strftime``) so the
+        logic is identical across the sqlite/mysql/postgres backends.
+        """
+        gran = "hour" if granularity == "hour" else "minute"
+        try:
+            async with self._sf() as session:
+                rows = (
+                    await session.execute(
+                        select(ConcurrencySampleRow.sampled_at, ConcurrencySampleRow.active_count)
+                        .where(
+                            ConcurrencySampleRow.sampled_at >= since,
+                            ConcurrencySampleRow.sampled_at <= until,
+                        )
+                        .order_by(ConcurrencySampleRow.sampled_at)
+                        .limit(_MAX_ROWS)
+                    )
+                ).all()
+        except Exception:
+            logger.warning("Failed to aggregate concurrency samples", exc_info=True)
+            return []
+
+        buckets: dict[datetime, dict[str, float]] = {}
+        for sampled_at, active_count in rows:
+            if sampled_at is None:
+                continue
+            if sampled_at.tzinfo is None:
+                sampled_at = sampled_at.replace(tzinfo=UTC)
+            key = _bucket_start(sampled_at, gran)
+            value = int(active_count or 0)
+            entry = buckets.get(key)
+            if entry is None:
+                buckets[key] = {"peak": value, "sum": value, "count": 1}
+            else:
+                entry["peak"] = max(entry["peak"], value)
+                entry["sum"] += value
+                entry["count"] += 1
+
+        result: list[ConcurrencyBucket] = []
+        for key in sorted(buckets):
+            entry = buckets[key]
+            count = max(1, int(entry["count"]))
+            result.append(
+                ConcurrencyBucket(
+                    bucket=key.isoformat(),
+                    peak=int(entry["peak"]),
+                    avg=round(entry["sum"] / count, 2),
+                    samples=count,
+                )
+            )
+        return result
+
+    async def purge_older_than(self, cutoff: datetime) -> int:
+        """Delete samples older than ``cutoff``. Returns rows removed."""
+        try:
+            async with self._sf() as session:
+                result = await session.execute(
+                    delete(ConcurrencySampleRow).where(ConcurrencySampleRow.sampled_at < cutoff)
+                )
+                await session.commit()
+                return int(result.rowcount or 0)
+        except Exception:  # pragma: no cover
+            logger.warning("Failed to purge concurrency samples", exc_info=True)
+            return 0
+
+
+def make_concurrency_sample_store(
+    session_factory: async_sessionmaker[AsyncSession] | None,
+) -> ConcurrencySampleStore | None:
+    """Factory mirroring the other ``make_*_store`` helpers.
+
+    Returns ``None`` on the memory backend (no SQL session factory), in which
+    case the monitor simply reports a live snapshot with no historical chart.
+    """
+    if session_factory is None:
+        return None
+    return ConcurrencySampleStore(session_factory)

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

@@ -23,6 +23,7 @@ from deerflow.persistence.agents.model import (
 )
 from deerflow.persistence.admin_stats.model import AdminLeaderboardDailyStatRow
 from deerflow.persistence.business_mapping.model import BusinessMappingRow
+from deerflow.persistence.concurrency.model import ConcurrencySampleRow
 from deerflow.persistence.dashboard_sessions.model import DashboardSessionRow
 from deerflow.persistence.memory_config.model import UserMemoryConfigRow
 from deerflow.persistence.feedback.model import FeedbackRow
@@ -85,6 +86,7 @@ __all__ = [
     "AgentSkillRow",
     "AdminLeaderboardDailyStatRow",
     "BusinessMappingRow",
+    "ConcurrencySampleRow",
     "DashboardSessionRow",
     "KnowledgeEmbeddingRow",
     "KnowledgeEntityRow",

offline-backend-20260512/backend/packages/harness/deerflow/runtime/runs/manager.py

@@ -279,6 +279,20 @@ class RunManager:
         async with self._lock:
             return any(r.thread_id == thread_id and r.status in (RunStatus.pending, RunStatus.running) for r in self._runs.values())
 
+    async def list_active(self) -> list[RunRecord]:
+        """Return every in-flight run (status pending/running), oldest first.
+
+        Powers the admin concurrency monitor: each record represents one live
+        call into the model runtime (interactive chat, agent chat, roundtable
+        seat, …), so ``len(list_active())`` is the current concurrent-call count
+        and each record can be cancelled to relieve model pressure. Completed
+        records linger in ``_runs`` until :meth:`cleanup`, so we filter by status
+        rather than returning the whole registry.
+        """
+        async with self._lock:
+            # Dict insertion order == creation order, so this is oldest-first.
+            return [r for r in self._runs.values() if r.status in (RunStatus.pending, RunStatus.running)]
+
     async def cleanup(self, run_id: str, *, delay: float = 300) -> None:
         """Remove a run record after an optional delay."""
         if delay > 0: