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

2026-06-26 20:50:47 +08:00
commit 89232d7521
@@ -86,6 +86,8 @@ The frontend (`frontend-web/`) is a React 19 + Vite SPA:
 - **敏感词管理 (Sensitive Word Management)**: configures the **展示层脱敏映射** (原始词 → 替换词). **Gated on the username `lqq`** (not the admin role) — `getAccountDisplayName().toLowerCase() === "lqq"` (username = email `@` prefix, else `user_id`, same as the profile page); the dropdown entry, the route guard, and the page-level guard all use this check. `src/lib/desensitize.ts` ships a frozen seed map `SEED_DESENSITIZE_MAP` (the deployment seed) plus a live mutable `DESENSITIZE_MAP` (initialized from the seed so `desensitize()` works before any fetch). `DesensitizeWordsLoader` (mounted in `App.tsx` `AppProviders`) fetches `/api/sensitive-words` on startup and, if non-empty, calls `replaceDesensitizeMap(...)` to make the backend config the source of truth (empty/failed → keep seed). Page `SensitiveWordsPage.tsx` (route `/page/strategy/admin/sensitive-words`, opened from the 「敏感词管理」 item in the **bottom-left settings dropdown** of `components/page-sidebar.tsx`, present in **both** full + collapsed sidebar modes) lists/adds/edits/启停/deletes words and saves the **whole set atomically** (replace-all); a 「导入种子词」 button merges any missing `SEED_DESENSITIZE_MAP` entries into the editor. Client + hooks in `strategy-components/api/sensitive-words.ts` (`wordsToMap` folds enabled rows into the runtime map). Backend: `/api/sensitive-words` (global/shared; open read, write allowed for admin **or** the `lqq` account), mirroring 菜单管理's store shape. **侧边栏 + 导航栏 menu labels are desensitized** at the data source: `useSidebarMenuItems`/`useActiveThirdLevel` (`core/page-layout/use-sidebar-menu.ts`) map every node's `label` through a desensitizer built **directly from the `/api/sensitive-words` query** (`makeDesensitizer(wordsToMap(words))`, falling back to the global seed `desensitize` while empty) — this sidesteps the `replaceDesensitizeMap` effect-timing race. Only `label` is rewritten; `path`/`id`/`icon` are untouched so routing/highlight are unaffected. All nav surfaces (`page-sidebar.tsx`, `page-sidebar-dark.tsx`, `page-sidebar-v1.tsx`, `strategy-components/Header.tsx`, `PageRoutes.tsx`) consume this hook and are covered; `MenuManagementPage` uses the raw tree so admins still edit real labels.
 - **技能地址管理 (Skill Address Management, admin-only)**: 扫描运行中的技能（`skills/{public,custom}` 下所有 `.md`/`.py`）里的 **http(s) URL 与 IP/IP:端口**，按唯一地址聚合全部出现位置（跨 .md/.py），支持**批量替换**（如把 UAT 地址一次性改成生产）。主体抽成可复用组件 `SkillAddressesPanel`（`pages/SkillAddressesPage.tsx` 导出，默认导出 `SkillAddressesPage` 仍保留旧路由 `/page/strategy/admin/skill-addresses` 作直链回退），**作为「技能地址管理」标签页嵌入管理员「技能管理」页 `pages/AdminSkillsPage.tsx`**（该页加 Tabs：「技能列表」+「技能地址管理」；入口在「设置和更多」下拉的「技能管理」`workspace-nav-menu.tsx` → `/page/workspace/admin/skills`，不再单列于 `components/page-sidebar.tsx` 左下角设置下拉）：打开自动扫描并带**进度条**（扫描逐技能 / 应用逐文件 X/N，应对 150+ 技能），URL/IP 类型过滤 + 搜索 + 「片段批量套用」（如所有 `uat.4.cn` → `prod.4.cn`）、每地址「替换为」输入框（可展开看 `file:line` 片段）、内联**预览**、**应用**、**多关键词检索**（空格分隔全部命中），以及**修改记录**面板——每次应用生成一条记录（显示时间/改了哪些地址 from→to/文件数/处数），可**针对单次记录回退**（二次确认）。client + NDJSON 流式解析在 `strategy-components/api/skill-addresses.ts`。后端 `/api/skill-addresses`（**仅管理员**）：所有文件 IO 走 `asyncio.to_thread` 不阻塞事件循环；`GET /scan`(+`/scan/stream`)、`POST /preview`、`POST /apply`(+`/apply/stream`)、`GET /backups`、`POST /rollback/{id}`；**子串安全替换**（基于 token span，不会误伤子串）、`expected_total` 乐观锁（409）、写前自动磁盘备份到 `{skills_root}/.address-edits/`（同时写一份 `{id}.manifest.json` 记录本次替换明细，供修改记录展示）、应用后刷新技能系统提示缓存。纯引擎 `deerflow/skills/address_scan.py`，router `app/gateway/routers/skill_addresses.py`，tests `tests/test_skill_addresses.py`。范围仅运行中技能，不动 `deploy/minimal/skills/`。
 - **按钮管理 (Task Button Management, admin-only)**: configures the **可配置跳转按钮** rendered in the left task panel of the task deep-link workspaces (rwfx / 3qfx / xdfx, see `TaskFxWorkspace`). Each button = `{business, label, linkType, target, appendTaskId, idParam, idKind, openMode, loginParams, appendAuth, authTokenParam, authNameParam, enabled, sortOrder}`; `linkType` is `url` (外链 `window.open`, optional append of `?{idParam}={id}` — **`idParam` is the editable query-param name**, e.g. `id`/`task_id`), `business` (内部切换到另一业务, reuses `resolveTaskDeeplinkTarget`, always task id), or `purpose` (弹出当前任务的「任务目的」表格 via the shared `TaskPurposeDialog`, no target — configurable for any page incl. xdfx). **`url` buttons support two independent, coexisting login-append mechanisms**: (1) **`loginParams` (追加登录参数, like 应用管理/Light App Center)** — a list of `{key, source, customValue}` resolved at click time: `runTaskButton` reads `userInfo` (via `readUserInfo` from `strategy-components/lib/light-app`, embed-isolated through `authStorage`) and appends `?{key}={value}` for each (`source` = a userInfo field such as `accessToken`/`yUserId`, or `other` → literal `customValue`); empty list = nothing appended. Reuses `LOGIN_SOURCE_OPTIONS` from light-app; stored as JSON column `task_buttons.login_params_json` (`PortableJSON`, migration `20260625_01`). (2) **`appendAuth` (追加登录态)** — appends `?{authTokenParam}={token1}&{authNameParam}={username}` (default param names `token`/`username`) so the跳转目标外部系统 can identify the current user — e.g. `…/action/sentiment?token=<JWT>&username=lizhaoxing`. **Gated on token login**: `token1` = the `?authToken=` used by token-exchange login; `username` = the raw login name the backend resolved from `getTokenInfo`. Both are stashed at login by `persistLoginCredentials` into `login.token1`/`login.username` (`core/auth/login-credentials.ts`, scoped via [[scoped-storage]] so嵌入/独立会话隔离 + 随独立会话迁移). Only token login writes them → a non-token-login session has no `token1`, so `runTaskButton` silently skips the auth append (the jump still works, just without those two params). The frontend reads `username` straight from the new `UsernameLoginResponse.username` field (the stored email local-part is lossy — lowercased/normalized — so not used), **not** by decoding the JWT. Stored as columns `task_buttons.{append_auth, auth_token_param, auth_name_param}` (migration `20260625_02`). The admin page renders both editors for `url` buttons (loginParams 虚线框 + 追加登录态 switch). **`idKind` (`task`/`action`) only matters for xdfx**: its deep-link `?taskId=` is an 行动id and the 任务id is resolved at login (`fetchTaskIdFromAction`) — `TaskContext` now also stashes the raw `actionId`, so a url button can append either (default `task`). Page `TaskButtonsPage.tsx` (route `/page/strategy/admin/task-buttons`, **admin-gated**, entry「按钮管理」in the bottom-left settings dropdown of `components/page-sidebar.tsx`, both full + collapsed modes) has per-business tabs, edits name/type/target/启停/排序/增删, saves the **whole set atomically** (replace-all), plus 「导入默认」 to merge code-seed defaults. **Seed/fallback** (like 敏感词管理): only when the whole table is empty (fresh deploy, never saved) do the workspace + page fall back to `defaultTaskButtons(cfg)` — rwfx's url buttons (编辑/上一步/下一步/3q详情) built from the **live** `/api/public/config/task-deeplink` plus cross-business switch buttons for **all three** businesses (each switches to the other two — rwfx→三情/行动, 3qfx→任务/行动, xdfx→任务/三情). Once saved, the store is the sole source of truth. **Bottom 快捷跳转 bar groups by prev/next**: `TaskJumpBar` renders only「上一步/下一步」-type buttons in the bottom bar — a `prev` group (id 含 `prev` 或文案「上一步」, 描边次要样式 + ←图标) rendered **左侧**, then a `next` group (id 含 `next` 或文案「下一步」, 实心主色) on the **右侧** (container `justify-end`); everything else goes to the right-edge floating menu. The default 上一步 button (`rwfx-prev`, `url`, target = `DASHBOARD_AGENT_ROUTE` `/page/strategy/dashboard-agent`, `appendTaskId` + `idParam=taskId` → `?taskId=任务id`, `openMode=self`) 默认指向**应用内**「大屏绘制」页，在 **both** 单智能体 (`TaskFxWorkspace`) 与 多智能体会商 (`RoundtablePlanningPage` 深链 rwfx-A) 都显示（共用 `TaskJumpBar`）。**`runTaskButton` 的 `url` 类型支持应用内路由**：target 以单个 `/` 开头时按内部路由处理——`self` 走 react-router 同窗导航（不刷新），`blank` 拼成当前源的 hash 路由新标签打开；外链行为不变。 The old hardcoded jump buttons were migrated out: `TaskFxWorkspace` no longer hardcodes 切换到三Line truncated
+- **实时并发监控 (Concurrency / System-Pressure Monitor, admin-only)**: a live panel mounted at the **top of the admin 排行榜 page** (`components/workspace/admin/concurrency-monitor.tsx` `ConcurrencyMonitorPanel`, rendered in `pages/AdminUserLeaderboardPage.tsx`) showing **当前并发的大模型调用数 + 列表（用户/智能体/类型/已运行时长）**, each row with a 「停止」button (and a 「全部停止」) that **truly aborts** that run's in-flight model stream via `RunManager.cancel` (`POST /api/admin/active-runs/{run_id}/cancel`) — to relieve pressure when 使用人数过多. Two line charts (minute/hour toggle): **并发量折线图** (peak/avg, `GET /api/admin/active-runs/concurrency`, fed by a lightweight background sampler into `concurrency_samples`) and **大模型出字速度折线图** (tokens/sec, `GET /api/admin/active-runs/token-speed`, aggregated from `llm_call_metrics`) so a dip pinpoints where the bottleneck is. A **「后台采样」Switch** (`system_settings.json → concurrency_monitor.enabled`, **default off / opt-in**, `GET|PUT /api/admin/active-runs/settings`) turns the background sampling on/off **at runtime, no restart** (the live concurrency count + 「停止」 work even while sampling is off — only the 并发量 history chart needs it on); everything is designed to **never impact production** (one tiny INSERT/15s, retention-purged, best-effort). Client/hooks: `core/admin/active-runs.ts` + `core/admin/hooks.ts` (`useActiveRuns`/`useConcurrencySeries`/`useTokenSpeedSeries`/`useCancel*`/`useMonitorSettings`). See the backend CLAUDE.md「Admin Concurrency Monitor」.
+- **会话冲突 409 中文化**: when a user submits a new question while the previous turn is still streaming, the backend's `RunManager.create_or_reject` rejects with HTTP 409 (`services.py` now returns a **Chinese** detail「当前对话还在生成回答…」). `core/threads/hooks.ts` `onError` detects the conflict (`isActiveRunConflictError`: 409 / "already has an active run" / the Chinese text) and shows a friendly `toast.info` instead of a raw error.
 - **State**: TanStack Query for server state, context providers (`SubtasksProvider`, `ArtifactsProvider`, `PromptInputProvider`) scoped to `WorkspaceRoutes`
 - **LLM streaming**: `@langchain/langgraph-sdk` streams messages from the backend via SSE
 - **UI**: Radix UI primitives + TailwindCSS v4 + shadcn-style components in `src/components/ui/`
@@ -0,0 +1,472 @@
+"use client";
+
+import {
+  Activity,
+  Gauge,
+  Loader2,
+  OctagonX,
+  RefreshCw,
+  Square,
+  Zap,
+} from "lucide-react";
+import { useMemo, useState } from "react";
+import { toast } from "sonner";
+
+import { Badge } from "@/components/ui/badge";
+import { Button } from "@/components/ui/button";
+import { Switch } from "@/components/ui/switch";
+import type {
+  ActiveRunItem,
+  ConcurrencyPoint,
+  Granularity,
+  TokenSpeedPoint,
+} from "@/core/admin/active-runs";
+import {
+  useActiveRuns,
+  useCancelActiveRun,
+  useCancelAllActiveRuns,
+  useConcurrencySeries,
+  useMonitorSettings,
+  useTokenSpeedSeries,
+  useUpdateMonitorSettings,
+} from "@/core/admin/hooks";
+import { cn } from "@/lib/utils";
+
+/** lookback window (hours) per granularity. */
+const LOOKBACK: Record<Granularity, number> = { minute: 3, hour: 48 };
+
+function fmtBucket(iso: string, granularity: Granularity): string {
+  const date = new Date(iso);
+  if (Number.isNaN(date.getTime())) return iso;
+  const mm = String(date.getMonth() + 1).padStart(2, "0");
+  const dd = String(date.getDate()).padStart(2, "0");
+  const hh = String(date.getHours()).padStart(2, "0");
+  const mi = String(date.getMinutes()).padStart(2, "0");
+  return granularity === "hour" ? `${mm}-${dd} ${hh}:00` : `${hh}:${mi}`;
+}
+
+function fmtElapsed(seconds: number): string {
+  if (seconds < 60) return `${seconds}s`;
+  const m = Math.floor(seconds / 60);
+  const s = seconds % 60;
+  if (m < 60) return `${m}m${s.toString().padStart(2, "0")}s`;
+  const h = Math.floor(m / 60);
+  return `${h}h${(m % 60).toString().padStart(2, "0")}m`;
+}
+
+interface LineSeries {
+  name: string;
+  color: string;
+  values: (number | null)[];
+}
+
+/** Minimal dependency-free SVG line chart (1–2 series + area under the first). */
+function LineChart({
+  labels,
+  series,
+  unit,
+  emptyHint,
+}: {
+  labels: string[];
+  series: LineSeries[];
+  unit: string;
+  emptyHint: string;
+}) {
+  const width = 760;
+  const height = 220;
+  const pad = { top: 18, right: 16, bottom: 26, left: 40 };
+  const cw = width - pad.left - pad.right;
+  const ch = height - pad.top - pad.bottom;
+
+  const allValues = series.flatMap((s) => s.values.filter((v): v is number => v != null));
+  const max = Math.max(1, ...allValues);
+  const n = labels.length;
+
+  if (n === 0) {
+    return (
+      <div className="flex h-[220px] items-center justify-center text-sm text-[#5F6368] dark:text-slate-400">
+        {emptyHint}
+      </div>
+    );
+  }
+
+  const xFor = (i: number) => pad.left + (n <= 1 ? cw / 2 : (i / (n - 1)) * cw);
+  const yFor = (v: number) => pad.top + ch - (v / max) * ch;
+  const grid = [0, 0.25, 0.5, 0.75, 1];
+
+  const linePath = (vals: (number | null)[]) => {
+    let d = "";
+    let started = false;
+    vals.forEach((v, i) => {
+      if (v == null) {
+        started = false;
+        return;
+      }
+      d += `${started ? "L" : "M"} ${xFor(i).toFixed(1)} ${yFor(v).toFixed(1)} `;
+      started = true;
+    });
+    return d.trim();
+  };
+
+  const first = series[0];
+  const areaPath = (() => {
+    if (!first) return "";
+    const line = linePath(first.values);
+    if (!line) return "";
+    return `${line} L ${xFor(n - 1).toFixed(1)} ${(pad.top + ch).toFixed(1)} L ${xFor(0).toFixed(1)} ${(pad.top + ch).toFixed(1)} Z`;
+  })();
+
+  const labelStep = Math.max(1, Math.ceil(n / 7));
+
+  return (
+    <svg viewBox={`0 0 ${width} ${height}`} className="w-full" role="img" aria-label="时序折线图">
+      <defs>
+        <linearGradient id="cm-area" x1="0" x2="0" y1="0" y2="1">
+          <stop offset="0%" stopColor={first?.color ?? "#C1121F"} stopOpacity="0.18" />
+          <stop offset="100%" stopColor={first?.color ?? "#C1121F"} stopOpacity="0.02" />
+        </linearGradient>
+      </defs>
+      {grid.map((r) => {
+        const y = pad.top + ch * r;
+        return (
+          <g key={r}>
+            <line
+              x1={pad.left}
+              x2={width - pad.right}
+              y1={y}
+              y2={y}
+              stroke="currentColor"
+              className="text-[#E5E7EB] dark:text-[#3F3F46]"
+              strokeWidth="1"
+            />
+            <text x={6} y={y + 4} className="fill-[#9CA3AF] text-[10px]">
+              {Math.round(max * (1 - r))}
+            </text>
+          </g>
+        );
+      })}
+      {areaPath && <path d={areaPath} fill="url(#cm-area)" />}
+      {series.map((s) => (
+        <path
+          key={s.name}
+          d={linePath(s.values)}
+          fill="none"
+          stroke={s.color}
+          strokeWidth="2"
+          strokeLinecap="round"
+          strokeLinejoin="round"
+        />
+      ))}
+      {labels.map((label, i) => {
+        if (i !== 0 && i !== n - 1 && i % labelStep !== 0) return null;
+        return (
+          <text key={`${label}-${i}`} x={xFor(i)} y={height - 8} textAnchor="middle" className="fill-[#6B7280] text-[10px]">
+            {label}
+          </text>
+        );
+      })}
+      {/* hover dots with native tooltip */}
+      {series.map((s) =>
+        s.values.map((v, i) =>
+          v == null ? null : (
+            <circle key={`${s.name}-${i}`} cx={xFor(i)} cy={yFor(v)} r={2.5} fill={s.color}>
+              <title>{`${labels[i]} · ${s.name}: ${v}${unit}`}</title>
+            </circle>
+          ),
+        ),
+      )}
+    </svg>
+  );
+}
+
+function ChartCard({
+  icon: Icon,
+  title,
+  subtitle,
+  children,
+}: {
+  icon: typeof Activity;
+  title: string;
+  subtitle: string;
+  children: React.ReactNode;
+}) {
+  return (
+    <section className="rounded-lg border border-[#F1D6D6] bg-white p-4 shadow-sm dark:border-[#213D5E] dark:bg-[#02244C]">
+      <div className="mb-3 flex items-center gap-2">
+        <Icon className="size-4 text-[#C1121F] dark:text-[#7FACF2]" />
+        <h3 className="font-semibold text-[#231815] dark:text-white">{title}</h3>
+        <span className="ml-auto text-xs text-[#8B6B6B] dark:text-[#7FACF2]">{subtitle}</span>
+      </div>
+      {children}
+    </section>
+  );
+}
+
+function ActiveRunRow({
+  run,
+  onStop,
+  stopping,
+}: {
+  run: ActiveRunItem;
+  onStop: (runId: string) => void;
+  stopping: boolean;
+}) {
+  return (
+    <div className="grid grid-cols-[minmax(0,1.4fr)_minmax(0,1.2fr)_72px_64px_64px_72px] items-center gap-3 border-b border-[#F1D6D6] px-3 py-2 text-sm last:border-b-0 dark:border-[#213D5E]">
+      <div className="min-w-0">
+        <div className="truncate font-medium text-[#202124] dark:text-white">{run.user_name || "未知用户"}</div>
+        <div className="truncate text-xs text-[#5F6368] dark:text-slate-400" title={run.thread_title}>
+          {run.thread_title || run.thread_id}
+        </div>
+      </div>
+      <div className="min-w-0 truncate text-[#202124] dark:text-slate-100">{run.agent_name}</div>
+      <Badge variant="outline" className="justify-center">
+        {run.kind}
+      </Badge>
+      <span
+        className={cn(
+          "text-center text-xs font-medium",
+          run.status === "running" ? "text-[#0F766E] dark:text-teal-300" : "text-[#B45309] dark:text-amber-300",
+        )}
+      >
+        {run.status === "running" ? "运行中" : "排队中"}
+      </span>
+      <span className="text-center text-xs tabular-nums text-[#5F6368] dark:text-slate-400">
+        {fmtElapsed(run.elapsed_seconds)}
+      </span>
+      <Button
+        size="sm"
+        variant="outline"
+        className="h-7 border-[#C1121F]/40 text-[#C1121F] hover:bg-[#FFF1F1] dark:border-rose-400/40 dark:text-rose-200"
+        disabled={stopping}
+        onClick={() => onStop(run.run_id)}
+      >
+        {stopping ? <Loader2 className="size-3.5 animate-spin" /> : <Square className="size-3.5" />}
+        停止
+      </Button>
+    </div>
+  );
+}
+
+export function ConcurrencyMonitorPanel() {
+  const [open, setOpen] = useState(true);
+  const [granularity, setGranularity] = useState<Granularity>("minute");
+
+  const { data: live, isLoading: liveLoading, refetch, isRefetching } = useActiveRuns(open);
+  const concurrency = useConcurrencySeries(granularity, LOOKBACK[granularity], open);
+  const tokenSpeed = useTokenSpeedSeries(granularity, LOOKBACK[granularity], open);
+  const { data: settings } = useMonitorSettings();
+  const updateSettings = useUpdateMonitorSettings();
+  const cancelOne = useCancelActiveRun();
+  const cancelAll = useCancelAllActiveRuns();
+
+  const [stoppingId, setStoppingId] = useState<string | null>(null);
+
+  const count = live?.count ?? 0;
+  const runs = live?.runs ?? [];
+  const samplerOn = settings?.enabled ?? false;
+
+  const concurrencyChart = useMemo(() => {
+    const points: ConcurrencyPoint[] = concurrency.data?.points ?? [];
+    return {
+      labels: points.map((p) => fmtBucket(p.bucket, granularity)),
+      series: [
+        { name: "峰值并发", color: "#C1121F", values: points.map((p) => p.peak) },
+        { name: "平均并发", color: "#F59E0B", values: points.map((p) => p.avg) },
+      ] as LineSeries[],
+    };
+  }, [concurrency.data, granularity]);
+
+  const tokenChart = useMemo(() => {
+    const points: TokenSpeedPoint[] = tokenSpeed.data?.points ?? [];
+    return {
+      labels: points.map((p) => fmtBucket(p.bucket, granularity)),
+      series: [
+        { name: "平均速度", color: "#2563EB", values: points.map((p) => p.avg_tokens_per_sec) },
+        { name: "最慢", color: "#9333EA", values: points.map((p) => p.min_tokens_per_sec) },
+      ] as LineSeries[],
+    };
+  }, [tokenSpeed.data, granularity]);
+
+  const handleStop = async (runId: string) => {
+    setStoppingId(runId);
+    try {
+      const res = await cancelOne.mutateAsync(runId);
+      toast[res.cancelled ? "success" : "info"](res.cancelled ? "已停止该对话调用" : "该调用已结束或不存在");
+    } catch (e) {
+      toast.error(e instanceof Error ? e.message : "停止失败");
+    } finally {
+      setStoppingId(null);
+    }
+  };
+
+  const handleStopAll = async () => {
+    if (runs.length === 0) return;
+    if (!window.confirm(`确定要停止当前全部 ${count} 个并发调用吗？这会中断这些用户正在生成的回答。`)) return;
+    try {
+      const res = await cancelAll.mutateAsync();
+      toast.success(`已停止 ${res.cancelled_count} 个调用`);
+    } catch (e) {
+      toast.error(e instanceof Error ? e.message : "全部停止失败");
+    }
+  };
+
+  const handleToggleSampler = async (next: boolean) => {
+    try {
+      await updateSettings.mutateAsync(next);
+      toast.success(next ? "已开启并发采样" : "已关闭并发采样（折线图不再新增历史点）");
+    } catch (e) {
+      toast.error(e instanceof Error ? e.message : "切换失败");
+    }
+  };
+
+  return (
+    <section className="space-y-4 rounded-xl border border-[#F1D6D6] bg-[#FFF7F7] p-4 shadow-sm dark:border-[#213D5E] dark:bg-[#011226]">
+      {/* header */}
+      <div className="flex flex-wrap items-center gap-3">
+        <div className="flex items-center gap-2">
+          <span className="flex size-9 items-center justify-center rounded-lg bg-[#C1121F] text-white dark:bg-[#2563EB]">
+            <Activity className="size-5" />
+          </span>
+          <div>
+            <div className="font-semibold text-[#231815] dark:text-white">实时并发监控</div>
+            <div className="text-xs text-[#8B6B6B] dark:text-[#7FACF2]">
+              监控大模型并发压力，可直接停止任意对话调用
+            </div>
+          </div>
+        </div>
+
+        <div className="ml-auto flex flex-wrap items-center gap-3">
+          <div className="flex items-center gap-2 rounded-lg border border-[#F1D6D6] bg-white px-3 py-1.5 dark:border-[#213D5E] dark:bg-[#02244C]">
+            <span className="text-xs text-[#5F6368] dark:text-slate-300">当前并发</span>
+            <span
+              className={cn(
+                "text-2xl font-semibold tabular-nums",
+                count > 0 ? "text-[#C1121F] dark:text-[#7FACF2]" : "text-[#202124] dark:text-white",
+              )}
+            >
+              {count}
+            </span>
+          </div>
+          <div className="flex items-center gap-2 rounded-lg border border-[#F1D6D6] bg-white px-3 py-1.5 dark:border-[#213D5E] dark:bg-[#02244C]">
+            <span className="text-xs text-[#5F6368] dark:text-slate-300">后台采样</span>
+            <Switch checked={samplerOn} onCheckedChange={handleToggleSampler} disabled={updateSettings.isPending} />
+          </div>
+          <Button variant="outline" size="sm" onClick={() => refetch()} disabled={isRefetching}>
+            <RefreshCw className={cn("mr-1 size-4", isRefetching && "animate-spin")} />
+            刷新
+          </Button>
+          <Button
+            variant="outline"
+            size="sm"
+            className="border-[#C1121F]/50 text-[#C1121F] hover:bg-[#FFF1F1] dark:border-rose-400/40 dark:text-rose-200"
+            onClick={handleStopAll}
+            disabled={cancelAll.isPending || runs.length === 0}
+          >
+            <OctagonX className="mr-1 size-4" />
+            全部停止
+          </Button>
+          <Button variant="ghost" size="sm" onClick={() => setOpen((v) => !v)}>
+            {open ? "收起" : "展开"}
+          </Button>
+        </div>
+      </div>
+
+      {open && (
+        <>
+          {!samplerOn && (
+            <div className="rounded-lg border border-amber-300/60 bg-amber-50 px-3 py-2 text-xs text-amber-800 dark:border-amber-300/30 dark:bg-amber-300/10 dark:text-amber-100">
+              后台采样默认关闭：实时并发数与「停止」功能始终可用。需要并发量历史折线图时，打开右上角「后台采样」开关即可开始记录（对线上服务影响极小，可随时关闭）。
+            </div>
+          )}
+
+          {/* granularity toggle */}
+          <div className="flex items-center gap-2">
+            <span className="text-xs text-[#5F6368] dark:text-slate-300">统计粒度</span>
+            {(["minute", "hour"] as Granularity[]).map((g) => (
+              <Button
+                key={g}
+                size="sm"
+                variant={granularity === g ? "default" : "outline"}
+                className={cn("h-7", granularity === g && "bg-[#C1121F] hover:bg-[#A50F1A] dark:bg-[#2563EB]")}
+                onClick={() => setGranularity(g)}
+              >
+                {g === "minute" ? "按分钟" : "按小时"}
+              </Button>
+            ))}
+            <span className="text-xs text-[#8B6B6B] dark:text-[#7FACF2]">
+              {granularity === "minute" ? "近 3 小时" : "近 48 小时"}
+            </span>
+          </div>
+
+          {/* charts */}
+          <div className="grid gap-3 xl:grid-cols-2">
+            <ChartCard icon={Gauge} title="并发量（系统压力）" subtitle="峰值 / 平均并发调用数">
+              {concurrency.isLoading ? (
+                <div className="flex h-[220px] items-center justify-center text-sm text-[#5F6368] dark:text-slate-400">
+                  <Loader2 className="mr-2 size-4 animate-spin" /> 加载中
+                </div>
+              ) : (
+                <LineChart
+                  labels={concurrencyChart.labels}
+                  series={concurrencyChart.series}
+                  unit=""
+                  emptyHint="暂无并发采样数据（采样关闭或时间窗口内无记录）"
+                />
+              )}
+            </ChartCard>
+            <ChartCard icon={Zap} title="大模型出字速度" subtitle="tokens/秒（越低说明模型越慢）">
+              {tokenSpeed.isLoading ? (
+                <div className="flex h-[220px] items-center justify-center text-sm text-[#5F6368] dark:text-slate-400">
+                  <Loader2 className="mr-2 size-4 animate-spin" /> 加载中
+                </div>
+              ) : (
+                <LineChart
+                  labels={tokenChart.labels}
+                  series={tokenChart.series}
+                  unit=" t/s"
+                  emptyHint="暂无大模型调用记录"
+                />
+              )}
+            </ChartCard>
+          </div>
+
+          {/* active run list */}
+          <section className="rounded-lg border border-[#F1D6D6] bg-white dark:border-[#213D5E] dark:bg-[#02244C]">
+            <div className="flex items-center gap-2 border-b border-[#F1D6D6] px-3 py-2 dark:border-[#213D5E]">
+              <span className="font-semibold text-[#231815] dark:text-white">正在进行的调用</span>
+              <Badge variant="outline">{count}</Badge>
+              {liveLoading && <Loader2 className="size-3.5 animate-spin text-[#8B6B6B]" />}
+            </div>
+            {runs.length === 0 ? (
+              <div className="py-10 text-center text-sm text-[#5F6368] dark:text-slate-400">
+                当前没有正在进行的大模型调用
+              </div>
+            ) : (
+              <div>
+                <div className="grid grid-cols-[minmax(0,1.4fr)_minmax(0,1.2fr)_72px_64px_64px_72px] gap-3 border-b border-[#F1D6D6] px-3 py-2 text-xs text-[#8B6B6B] dark:border-[#213D5E] dark:text-[#7FACF2]">
+                  <span>用户 / 会话</span>
+                  <span>智能体</span>
+                  <span className="text-center">类型</span>
+                  <span className="text-center">状态</span>
+                  <span className="text-center">已运行</span>
+                  <span className="text-center">操作</span>
+                </div>
+                {runs.map((run) => (
+                  <ActiveRunRow
+                    key={run.run_id}
+                    run={run}
+                    onStop={handleStop}
+                    stopping={stoppingId === run.run_id}
+                  />
+                ))}
+              </div>
+            )}
+          </section>
+        </>
+      )}
+    </section>
+  );
+}
+
+export default ConcurrencyMonitorPanel;
@@ -0,0 +1,159 @@
+import { apiFetch } from "@/core/api";
+import { getBackendBaseURL } from "@/core/config";
+import { OptType, writeYLog } from "@/core/log/y-log";
+
+/** One in-flight model run as seen by the admin concurrency monitor. */
+export interface ActiveRunItem {
+  run_id: string;
+  thread_id: string;
+  status: string;
+  created_at: string;
+  elapsed_seconds: number;
+  user_id: string;
+  user_name: string;
+  agent_id: string;
+  agent_name: string;
+  thread_title: string;
+  kind: string;
+  multitask_strategy: string;
+}
+
+export interface ActiveRunsResponse {
+  count: number;
+  server_time: string;
+  runs: ActiveRunItem[];
+}
+
+export interface ConcurrencyPoint {
+  bucket: string;
+  peak: number;
+  avg: number;
+  samples: number;
+}
+
+export interface ConcurrencySeriesResponse {
+  granularity: string;
+  since: string;
+  until: string;
+  enabled: boolean;
+  points: ConcurrencyPoint[];
+}
+
+export interface TokenSpeedPoint {
+  bucket: string;
+  avg_tokens_per_sec: number;
+  min_tokens_per_sec: number;
+  max_tokens_per_sec: number;
+  samples: number;
+}
+
+export interface TokenSpeedSeriesResponse {
+  granularity: string;
+  since: string;
+  until: string;
+  points: TokenSpeedPoint[];
+}
+
+export type Granularity = "minute" | "hour";
+
+const base = () => `${getBackendBaseURL()}/api/admin/active-runs`;
+
+async function parse<T>(res: Response, what: string): Promise<T> {
+  if (!res.ok) {
+    const err = (await res.json().catch(() => ({}))) as { detail?: string };
+    throw new Error(err.detail ?? `${what} (HTTP ${res.status})`);
+  }
+  return res.json() as Promise<T>;
+}
+
+/** Live snapshot of all in-flight runs. Polled — intentionally NOT y-logged. */
+export async function loadActiveRuns(): Promise<ActiveRunsResponse> {
+  return parse<ActiveRunsResponse>(await apiFetch(base()), "Failed to load active runs");
+}
+
+/** Concurrency time-series (peak/avg per minute|hour). Polled — not y-logged. */
+export async function loadConcurrencySeries(
+  granularity: Granularity,
+  hours?: number,
+): Promise<ConcurrencySeriesResponse> {
+  const params = new URLSearchParams({ granularity });
+  if (hours) params.set("hours", String(hours));
+  return parse<ConcurrencySeriesResponse>(
+    await apiFetch(`${base()}/concurrency?${params}`),
+    "Failed to load concurrency series",
+  );
+}
+
+/** Token-speed time-series (tokens/sec per minute|hour). Polled — not y-logged. */
+export async function loadTokenSpeedSeries(
+  granularity: Granularity,
+  hours?: number,
+): Promise<TokenSpeedSeriesResponse> {
+  const params = new URLSearchParams({ granularity });
+  if (hours) params.set("hours", String(hours));
+  return parse<TokenSpeedSeriesResponse>(
+    await apiFetch(`${base()}/token-speed?${params}`),
+    "Failed to load token speed series",
+  );
+}
+
+/** Truly stop one in-flight run (cancels the live model stream). */
+export async function cancelActiveRun(runId: string): Promise<{ run_id: string; cancelled: boolean }> {
+  const url = `${base()}/${encodeURIComponent(runId)}/cancel`;
+  const result = await parse<{ run_id: string; cancelled: boolean }>(
+    await apiFetch(url, { method: "POST" }),
+    "Failed to cancel run",
+  );
+  writeYLog({
+    optType: OptType.other,
+    optDetail: "CM助手-全量-并发监控-停止对话调用",
+    url,
+    reqParam: { runId },
+    returnParam: result,
+  });
+  return result;
+}
+
+/** Stop every in-flight run at once. */
+export async function cancelAllActiveRuns(): Promise<{ cancelled_count: number }> {
+  const url = `${base()}/cancel-all`;
+  const result = await parse<{ cancelled_count: number }>(
+    await apiFetch(url, { method: "POST" }),
+    "Failed to cancel all runs",
+  );
+  writeYLog({
+    optType: OptType.other,
+    optDetail: "CM助手-全量-并发监控-全部停止",
+    url,
+    returnParam: result,
+  });
+  return result;
+}
+
+/** Read the background-sampler on/off switch. */
+export async function getMonitorSettings(): Promise<{ enabled: boolean }> {
+  return parse<{ enabled: boolean }>(
+    await apiFetch(`${base()}/settings`),
+    "Failed to load monitor settings",
+  );
+}
+
+/** Toggle the background concurrency sampler (admin only, no restart). */
+export async function updateMonitorSettings(enabled: boolean): Promise<{ enabled: boolean }> {
+  const url = `${base()}/settings`;
+  const result = await parse<{ enabled: boolean }>(
+    await apiFetch(url, {
+      method: "PUT",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ enabled }),
+    }),
+    "Failed to update monitor settings",
+  );
+  writeYLog({
+    optType: enabled ? OptType.enabled : OptType.disabled,
+    optDetail: `CM助手-全量-并发监控-${enabled ? "开启" : "关闭"}采样`,
+    url,
+    reqParam: { enabled },
+  });
+  return result;
+}
@@ -17,6 +17,96 @@ import {
  updateAdminV1Fact,
  updateAdminV1Section,
 } from "./api";
+import {
+  cancelActiveRun,
+  cancelAllActiveRuns,
+  getMonitorSettings,
+  loadActiveRuns,
+  loadConcurrencySeries,
+  loadTokenSpeedSeries,
+  updateMonitorSettings,
+  type Granularity,
+} from "./active-runs";
+
+// ── 实时并发监控 (admin) ─────────────────────────────────────────────────────
+
+/** Live snapshot of in-flight runs. Polls every `intervalMs` while `enabled`. */
+export function useActiveRuns(enabled = true, intervalMs = 4000) {
+  return useQuery({
+    queryKey: ["admin-active-runs", "live"],
+    queryFn: loadActiveRuns,
+    enabled,
+    refetchInterval: enabled ? intervalMs : false,
+    refetchOnWindowFocus: false,
+  });
+}
+
+/** 并发量时序 (peak/avg per minute|hour). */
+export function useConcurrencySeries(
+  granularity: Granularity,
+  hours: number | undefined,
+  enabled = true,
+  intervalMs = 15000,
+) {
+  return useQuery({
+    queryKey: ["admin-active-runs", "concurrency", granularity, hours ?? null],
+    queryFn: () => loadConcurrencySeries(granularity, hours),
+    enabled,
+    refetchInterval: enabled ? intervalMs : false,
+    refetchOnWindowFocus: false,
+  });
+}
+
+/** 大模型 token 出字速度时序 (tokens/sec per minute|hour). */
+export function useTokenSpeedSeries(
+  granularity: Granularity,
+  hours: number | undefined,
+  enabled = true,
+  intervalMs = 15000,
+) {
+  return useQuery({
+    queryKey: ["admin-active-runs", "token-speed", granularity, hours ?? null],
+    queryFn: () => loadTokenSpeedSeries(granularity, hours),
+    enabled,
+    refetchInterval: enabled ? intervalMs : false,
+    refetchOnWindowFocus: false,
+  });
+}
+
+export function useCancelActiveRun() {
+  const qc = useQueryClient();
+  return useMutation({
+    mutationFn: (runId: string) => cancelActiveRun(runId),
+    onSuccess: () => qc.invalidateQueries({ queryKey: ["admin-active-runs", "live"] }),
+  });
+}
+
+export function useCancelAllActiveRuns() {
+  const qc = useQueryClient();
+  return useMutation({
+    mutationFn: () => cancelAllActiveRuns(),
+    onSuccess: () => qc.invalidateQueries({ queryKey: ["admin-active-runs", "live"] }),
+  });
+}
+
+export function useMonitorSettings(enabled = true) {
+  return useQuery({
+    queryKey: ["admin-active-runs", "settings"],
+    queryFn: getMonitorSettings,
+    enabled,
+    refetchOnWindowFocus: false,
+  });
+}
+
+export function useUpdateMonitorSettings() {
+  const qc = useQueryClient();
+  return useMutation({
+    mutationFn: (nextEnabled: boolean) => updateMonitorSettings(nextEnabled),
+    onSuccess: (data) => {
+      qc.setQueryData(["admin-active-runs", "settings"], data);
+    },
+  });
+}

 export function useUserLeaderboard(
  range: {
@@ -219,6 +219,19 @@ function isRunNotFoundError(message: string): boolean {
  return /run\s+[\w-]{8,}\s+not\s+found/i.test(message);
 }

+/**
+ * 同一会话上一轮还没生成完，又提交了新问题 → 后端返回 409
+ * 「already has an active run」。匹配 HTTP 409、英文兜底文案，或后端已中文化
+ * 的提示，统一映射成一句友好的中文。
+ */
+function isActiveRunConflictError(message: string): boolean {
+  return (
+    /\b409\b/.test(message) ||
+    /already has an active run/i.test(message) ||
+    /还在生成回答/.test(message)
+  );
+}
+
 function buildCurrentUserLabel(): string {
  const auth = getStoredAuth();
  if (!auth) {
@@ -555,6 +568,12 @@ export function useThreadStream({
        }
        return;
      }
+      // 上一轮还没结束又发了新问题（后端 409）：给一句明确的中文提示，而不是
+      // 把原始的英文/HTTP 错误抛给用户。
+      if (isActiveRunConflictError(message)) {
+        toast.info("当前对话还在生成回答，请等待本轮回复结束，或先点「停止」再发送新的问题。");
+        return;
+      }
      toast.error(message);
    },
    onFinish(state) {
@@ -46,6 +46,7 @@ import { ScrollArea } from "@/components/ui/scroll-area";
 import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select";
 import { Tabs, TabsContent, TabsList, TabsTrigger } from "@/components/ui/tabs";
 import { MarkdownContent } from "@/components/workspace/messages/markdown-content";
+import { ConcurrencyMonitorPanel } from "@/components/workspace/admin/concurrency-monitor";
 import type {
  AdminLeaderboardDimension,
  AdminLeaderboardOverview,
@@ -1389,6 +1390,8 @@ export default function AdminUserLeaderboardPage() {
        </div>

          <div className="space-y-5 p-6">
+            <ConcurrencyMonitorPanel />
+
            {isLoading && (
              <div className="flex flex-col items-center justify-center gap-3 py-20 text-center text-sm text-[#5F6368] dark:text-slate-400">
                <div className="flex items-center gap-2">
@@ -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`.
@@ -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)

@@ -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)
@@ -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)
@@ -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()
@@ -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()
@@ -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",
+]
@@ -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)
@@ -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)
@@ -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",
@@ -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: