fix: isolate user concurrency from internal agent runs

fix: iframe底部选择智能体

See merge request knowledge-base-bj/deerflow-server!164

frontend-web/src/components/wiki/ReportMarkdownView.tsx

@@ -0,0 +1,189 @@
+"use client";
+
+import { BookOpen, ListTree } from "lucide-react";
+import { useCallback, useEffect, useMemo, useRef, useState } from "react";
+
+import { MarkdownContent } from "@/components/workspace/messages/markdown-content";
+import { Badge } from "@/components/ui/badge";
+import { reasoningPlugins } from "@/core/streamdown";
+import { createMarkdownAnchorResolver } from "@/core/utils/markdown";
+import { cn } from "@/lib/utils";
+
+import "./wiki.css";
+
+function stripMarkdownToText(markdown: string): string {
+  return markdown
+    .replace(/```[\s\S]*?```/g, " ")
+    .replace(/~~~[\s\S]*?~~~/g, " ")
+    .replace(/!\[[^\]]*\]\([^)]+\)/g, " ")
+    .replace(/\[([^\]]+)\]\([^)]+\)/g, "$1")
+    .replace(/^#{1,6}\s+/gm, "")
+    .replace(/^[>\-*+\d.)\s]+/gm, "")
+    .replace(/[`*_~|]/g, "")
+    .replace(/<[^>]+>/g, " ")
+    .replace(/\s+/g, " ")
+    .trim();
+}
+
+function getMarkdownExcerpt(markdown: string): string {
+  const text = stripMarkdownToText(markdown);
+  if (!text) return "";
+  return text.length > 150 ? `${text.slice(0, 150)}...` : text;
+}
+
+function countReadableWords(markdown: string): number {
+  const text = stripMarkdownToText(markdown);
+  if (!text) return 0;
+  const cjk = text.match(/[\u4e00-\u9fff]/g)?.length ?? 0;
+  const words = text.replace(/[\u4e00-\u9fff]/g, " ").match(/[A-Za-z0-9]+/g)?.length ?? 0;
+  return cjk + words;
+}
+
+function selectorForId(id: string): string {
+  const css = window.CSS;
+  if (css?.escape) return `#${css.escape(id)}`;
+  return `[id="${id.replace(/\\/g, "\\\\").replace(/"/g, '\\"')}"]`;
+}
+
+export function ReportMarkdownView({
+  content,
+  title,
+  subtitle,
+  badgeLabel = "报告阅读",
+  className,
+  compactHeader = false,
+}: {
+  content: string;
+  title?: string;
+  subtitle?: string;
+  badgeLabel?: string;
+  className?: string;
+  compactHeader?: boolean;
+}) {
+  const rootRef = useRef<HTMLDivElement>(null);
+  const anchorResolver = useMemo(() => createMarkdownAnchorResolver(content), [content]);
+  const headings = useMemo(
+    () => anchorResolver.headings.filter((item) => item.level >= 1 && item.level <= 3),
+    [anchorResolver],
+  );
+  const [activeHeadingId, setActiveHeadingId] = useState<string | null>(null);
+  const excerpt = useMemo(() => getMarkdownExcerpt(content), [content]);
+  const sectionCount = headings.length;
+  const wordCount = useMemo(() => countReadableWords(content), [content]);
+
+  useEffect(() => {
+    setActiveHeadingId(headings[0]?.id ?? null);
+  }, [headings]);
+
+  useEffect(() => {
+    const root = rootRef.current;
+    if (!root || headings.length === 0) return;
+
+    const targets = headings
+      .map((item) => root.querySelector<HTMLElement>(selectorForId(item.id)))
+      .filter((item): item is HTMLElement => !!item);
+    if (targets.length === 0) return;
+
+    const observer = new IntersectionObserver(
+      (entries) => {
+        const visible = entries
+          .filter((entry) => entry.isIntersecting)
+          .sort((a, b) => a.boundingClientRect.top - b.boundingClientRect.top)[0];
+        if (visible?.target.id) setActiveHeadingId(visible.target.id);
+      },
+      {
+        root: root.closest("[data-radix-scroll-area-viewport]") ?? null,
+        rootMargin: "-96px 0px -65% 0px",
+        threshold: [0, 1],
+      },
+    );
+
+    targets.forEach((target) => observer.observe(target));
+    return () => observer.disconnect();
+  }, [headings, content]);
+
+  const scrollToHeading = useCallback((id: string) => {
+    const root = rootRef.current;
+    const target =
+      root?.querySelector<HTMLElement>(selectorForId(id)) ?? document.getElementById(id);
+    setActiveHeadingId(id);
+    target?.scrollIntoView({ behavior: "smooth", block: "start" });
+  }, []);
+
+  return (
+    <div
+      ref={rootRef}
+      className={cn("report-markdown-shell grid min-h-0 gap-5 lg:grid-cols-[minmax(0,1fr)_300px]", className)}
+    >
+      <article className="report-markdown-paper min-w-0 overflow-hidden">
+        <header className={cn("report-markdown-header", compactHeader && "report-markdown-header-compact")}>
+          <div className="flex flex-wrap items-center gap-2">
+            <Badge variant="secondary" className="report-markdown-badge">
+              <BookOpen className="size-3.5" />
+              {badgeLabel}
+            </Badge>
+            {subtitle && (
+              <span className="max-w-full truncate text-xs text-slate-500 dark:text-slate-400">
+                {subtitle}
+              </span>
+            )}
+          </div>
+          {title && <h1 className="report-markdown-title">{title}</h1>}
+          {!compactHeader && excerpt && <p className="report-markdown-excerpt">{excerpt}</p>}
+          <div className="report-markdown-metrics">
+            <span>{sectionCount || 1} 个章节</span>
+            <span>约 {wordCount} 字</span>
+          </div>
+        </header>
+
+        <div className="wiki-markdown report-markdown-body">
+          <MarkdownContent
+            className="max-w-none"
+            content={content}
+            isLoading={false}
+            remarkPlugins={reasoningPlugins.remarkPlugins}
+            rehypePlugins={reasoningPlugins.rehypePlugins}
+          />
+        </div>
+      </article>
+
+      <aside className="hidden min-w-0 lg:block">
+        <div className="report-markdown-toc">
+          <div className="mb-3 flex items-center justify-between gap-2">
+            <div className="flex items-center gap-2 text-sm font-semibold text-slate-900 dark:text-slate-100">
+              <ListTree className="size-4" />
+              内容目录
+            </div>
+            <span className="rounded-md bg-slate-100 px-1.5 py-0.5 text-xs text-slate-500 dark:bg-slate-800 dark:text-slate-400">
+              {sectionCount || 0}
+            </span>
+          </div>
+          {headings.length > 0 ? (
+            <nav className="report-markdown-toc-list">
+              {headings.map((item) => (
+                <button
+                  key={item.id}
+                  type="button"
+                  className={cn(
+                    "report-markdown-toc-item",
+                    activeHeadingId === item.id && "report-markdown-toc-item-active",
+                    item.level === 2 && "report-markdown-toc-level-2",
+                    item.level === 3 && "report-markdown-toc-level-3",
+                  )}
+                  title={item.text}
+                  onClick={() => scrollToHeading(item.id)}
+                >
+                  {item.text}
+                </button>
+              ))}
+            </nav>
+          ) : (
+            <div className="rounded-md border border-dashed p-3 text-xs text-slate-500 dark:border-slate-800 dark:text-slate-400">
+              暂无标题目录
+            </div>
+          )}
+        </div>
+      </aside>
+    </div>
+  );
+}

frontend-web/src/components/wiki/wiki.css

@@ -111,6 +111,340 @@
   border-color: #eef1f5;
 }
 
+.public-report-markdown {
+  color: #334155;
+  font-feature-settings: "kern";
+  text-rendering: optimizeLegibility;
+}
+
+.public-report-markdown h1 {
+  font-size: 30px;
+  letter-spacing: 0;
+  margin-bottom: 24px;
+}
+
+.public-report-markdown h2 {
+  font-size: 22px;
+  margin-top: 42px;
+  margin-bottom: 16px;
+  padding-bottom: 10px;
+}
+
+.public-report-markdown h3 {
+  font-size: 18px;
+  margin-top: 30px;
+  margin-bottom: 12px;
+}
+
+.public-report-markdown p,
+.public-report-markdown li {
+  font-size: 16px;
+  line-height: 1.9;
+}
+
+.public-report-markdown p {
+  margin-bottom: 16px;
+}
+
+.public-report-markdown ul,
+.public-report-markdown ol {
+  margin: 14px 0 18px;
+  padding-left: 26px;
+}
+
+.public-report-markdown ul {
+  list-style-type: disc;
+}
+
+.public-report-markdown ol {
+  list-style-type: decimal;
+}
+
+.public-report-markdown li {
+  padding-left: 2px;
+}
+
+.public-report-markdown code {
+  border-radius: 6px;
+  background: #f1f5f9;
+  padding: 2px 6px;
+  color: #be123c;
+  font-size: 0.92em;
+}
+
+.public-report-markdown pre {
+  margin: 20px 0;
+  border-radius: 8px;
+  background: #0f172a;
+  padding: 16px;
+  overflow-x: auto;
+}
+
+.public-report-markdown pre code {
+  background: transparent;
+  padding: 0;
+  color: inherit;
+}
+
+.public-report-markdown table {
+  display: block;
+  overflow-x: auto;
+  border-radius: 8px;
+  border: 1px solid #e8ecf1;
+}
+
+.public-report-markdown th,
+.public-report-markdown td {
+  min-width: 120px;
+  vertical-align: top;
+}
+
+.public-report-markdown blockquote {
+  margin: 20px 0;
+}
+
+.dark .public-report-markdown {
+  color: #cbd5e1;
+}
+
+.dark .public-report-markdown code {
+  background: #1e293b;
+  color: #f9a8d4;
+}
+
+.dark .public-report-markdown pre {
+  background: #020617;
+}
+
+.dark .public-report-markdown table {
+  border-color: #334155;
+}
+
+.report-markdown-shell {
+  align-items: start;
+  padding: 2px;
+}
+
+.report-markdown-paper {
+  border: 1px solid rgba(203, 213, 225, 0.85);
+  border-radius: 8px;
+  background:
+    linear-gradient(180deg, rgba(248, 250, 252, 0.92), rgba(255, 255, 255, 0) 220px),
+    #ffffff;
+  box-shadow:
+    0 1px 2px rgba(15, 23, 42, 0.04),
+    0 18px 44px rgba(15, 23, 42, 0.07);
+}
+
+.report-markdown-header {
+  border-bottom: 1px solid #e8edf4;
+  padding: 26px 34px 22px;
+}
+
+.report-markdown-header-compact {
+  padding-bottom: 18px;
+}
+
+.report-markdown-badge {
+  gap: 6px;
+  border: 1px solid #dbe3ef;
+  background: #ffffff;
+  color: #475569;
+}
+
+.report-markdown-title {
+  margin: 14px 0 0;
+  color: #0f172a;
+  font-size: 26px;
+  font-weight: 700;
+  line-height: 1.35;
+  letter-spacing: 0;
+}
+
+.report-markdown-excerpt {
+  margin: 14px 0 0;
+  max-width: 840px;
+  color: #64748b;
+  font-size: 14px;
+  line-height: 1.8;
+}
+
+.report-markdown-metrics {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 8px;
+  margin-top: 16px;
+  color: #64748b;
+  font-size: 12px;
+}
+
+.report-markdown-metrics span {
+  border: 1px solid #e2e8f0;
+  border-radius: 6px;
+  background: rgba(255, 255, 255, 0.78);
+  padding: 4px 8px;
+}
+
+.report-markdown-body {
+  max-width: 920px;
+  margin: 0 auto;
+  padding: 32px 42px 44px;
+}
+
+.report-markdown-body h1 {
+  font-size: 28px;
+}
+
+.report-markdown-body h2 {
+  position: relative;
+  margin-top: 44px;
+  border-bottom-color: #e5edf7;
+  font-size: 22px;
+}
+
+.report-markdown-body h2::before {
+  content: "";
+  position: absolute;
+  left: -16px;
+  top: 0.34em;
+  width: 4px;
+  height: 1.05em;
+  border-radius: 999px;
+  background: #2563eb;
+}
+
+.report-markdown-body h3 {
+  color: #1e293b;
+  font-size: 18px;
+}
+
+.report-markdown-body p,
+.report-markdown-body li {
+  font-size: 16px;
+  line-height: 1.92;
+}
+
+.report-markdown-body blockquote {
+  border-left-color: #2563eb;
+  background: #f8fbff;
+}
+
+.report-markdown-body table {
+  box-shadow: 0 1px 0 rgba(15, 23, 42, 0.04);
+}
+
+.report-markdown-toc {
+  position: sticky;
+  top: 16px;
+  max-height: calc(100vh - 32px);
+  overflow: hidden;
+  border: 1px solid rgba(203, 213, 225, 0.82);
+  border-radius: 8px;
+  background: rgba(255, 255, 255, 0.92);
+  padding: 14px;
+  box-shadow: 0 12px 34px rgba(15, 23, 42, 0.06);
+  backdrop-filter: blur(10px);
+}
+
+.report-markdown-toc-list {
+  max-height: calc(100vh - 128px);
+  overflow: auto;
+  padding-right: 4px;
+}
+
+.report-markdown-toc-item {
+  display: block;
+  width: 100%;
+  overflow: hidden;
+  text-overflow: ellipsis;
+  white-space: nowrap;
+  border-left: 2px solid transparent;
+  border-radius: 6px;
+  padding: 7px 8px;
+  color: #64748b;
+  font-size: 12px;
+  line-height: 1.4;
+  text-align: left;
+  transition: background 160ms ease, border-color 160ms ease, color 160ms ease;
+}
+
+.report-markdown-toc-item:hover {
+  background: #f1f5f9;
+  color: #0f172a;
+}
+
+.report-markdown-toc-item-active {
+  border-left-color: #2563eb;
+  background: #eff6ff;
+  color: #1d4ed8;
+  font-weight: 600;
+}
+
+.report-markdown-toc-level-2 {
+  padding-left: 18px;
+}
+
+.report-markdown-toc-level-3 {
+  padding-left: 30px;
+  color: #94a3b8;
+}
+
+.dark .report-markdown-paper {
+  border-color: #1f2937;
+  background:
+    linear-gradient(180deg, rgba(15, 23, 42, 0.9), rgba(2, 6, 23, 0) 220px),
+    #020617;
+  box-shadow: 0 18px 44px rgba(0, 0, 0, 0.24);
+}
+
+.dark .report-markdown-header {
+  border-bottom-color: #1e293b;
+}
+
+.dark .report-markdown-badge,
+.dark .report-markdown-metrics span {
+  border-color: #334155;
+  background: #0f172a;
+  color: #cbd5e1;
+}
+
+.dark .report-markdown-title {
+  color: #f8fafc;
+}
+
+.dark .report-markdown-excerpt,
+.dark .report-markdown-metrics {
+  color: #94a3b8;
+}
+
+.dark .report-markdown-body h3 {
+  color: #e2e8f0;
+}
+
+.dark .report-markdown-body h2::before {
+  background: #60a5fa;
+}
+
+.dark .report-markdown-body blockquote {
+  background: #0f172a;
+  border-left-color: #60a5fa;
+}
+
+.dark .report-markdown-toc {
+  border-color: #1f2937;
+  background: rgba(2, 6, 23, 0.88);
+}
+
+.dark .report-markdown-toc-item:hover {
+  background: #0f172a;
+  color: #f8fafc;
+}
+
+.dark .report-markdown-toc-item-active {
+  background: rgba(37, 99, 235, 0.16);
+  color: #93c5fd;
+}
+
 /* Anchor target offset so headings clear the floating "目录" pill. */
 .wiki-markdown h1,
 .wiki-markdown h2,

frontend-web/src/components/workspace/agents/agent-selector.tsx

@@ -93,43 +93,47 @@ export function AgentSelector({
   return (
     <>
       <Dialog open={open} onOpenChange={setOpen}>
-        <DialogTrigger asChild>
-          {iconMode ? (
-            <Tooltip content={iconTooltipLabel}>
-              <Button
-                type="button"
-                variant="ghost"
-                size="icon"
-                disabled={disabled}
-                aria-label={selectedAgent ? `已选择智能体：${triggerLabel}` : "选择智能体"}
-                className={cn(
-                  "relative size-9 rounded-full text-muted-foreground shadow-none hover:bg-muted/60 hover:text-foreground",
-                  selectedAgent && "text-amber-600 hover:text-amber-700",
-                  selectedAgent && iconActiveClassName,
-                  iconButtonClassName,
-                )}
-              >
-                <BotIcon className="size-4" />
-                {selectedAgent && (
-                  <span
-                    role="button"
-                    tabIndex={0}
-                    aria-label="清除已选 agent"
-                    className="absolute -right-1 -top-1 inline-flex size-4 items-center justify-center rounded-full border bg-background text-foreground shadow-sm hover:bg-muted"
-                    onClick={handleClear}
-                    onKeyDown={(e) => {
-                      if (e.key === "Enter" || e.key === " ") {
-                        e.preventDefault();
-                        onAgentChange(null);
-                      }
-                    }}
-                  >
-                    <XIcon className="size-2.5" />
-                  </span>
-                )}
-              </Button>
-            </Tooltip>
-          ) : (
+        {iconMode ? (
+          <Tooltip content={iconTooltipLabel}>
+            <div>
+              <DialogTrigger asChild>
+                <Button
+                  type="button"
+                  variant="ghost"
+                  size="icon"
+                  disabled={disabled}
+                  aria-label={selectedAgent ? `已选择智能体：${triggerLabel}` : "选择智能体"}
+                  className={cn(
+                    "relative size-9 rounded-full text-muted-foreground shadow-none hover:bg-muted/60 hover:text-foreground",
+                    selectedAgent && "text-amber-600 hover:text-amber-700",
+                    selectedAgent && iconActiveClassName,
+                    iconButtonClassName,
+                  )}
+                >
+                  <BotIcon className="size-4" />
+                  {selectedAgent && (
+                    <span
+                      role="button"
+                      tabIndex={0}
+                      aria-label="清除已选 agent"
+                      className="absolute -right-1 -top-1 inline-flex size-4 items-center justify-center rounded-full border bg-background text-foreground shadow-sm hover:bg-muted"
+                      onClick={handleClear}
+                      onKeyDown={(e) => {
+                        if (e.key === "Enter" || e.key === " ") {
+                          e.preventDefault();
+                          onAgentChange(null);
+                        }
+                      }}
+                    >
+                      <XIcon className="size-2.5" />
+                    </span>
+                  )}
+                </Button>
+              </DialogTrigger>
+            </div>
+          </Tooltip>
+        ) : (
+          <DialogTrigger asChild>
             <PromptInputButton
               type="button"
               disabled={disabled}
@@ -167,8 +171,8 @@ export function AgentSelector({
                 </span>
               )}
             </PromptInputButton>
-          )}
-        </DialogTrigger>
+          </DialogTrigger>
+        )}
         <DialogContent className="overflow-hidden p-0 sm:max-w-md">
           <DialogTitle className="sr-only">选择 Agent</DialogTitle>
           <Command className="**:data-[slot=command-input-wrapper]:h-auto">

frontend-web/src/components/workspace/agents/skill-transfer-selector.tsx

@@ -45,11 +45,32 @@ function skillLabel(skill: Skill) {
 
 function matchesQuery(skill: Skill, q: string) {
   if (!q) return true;
-  return (
-    skill.name.toLowerCase().includes(q) ||
-    (skill.name_zh ?? "").toLowerCase().includes(q) ||
-    (skill.description ?? "").toLowerCase().includes(q)
-  );
+  const haystack = [
+    skill.name,
+    skill.name_zh,
+    skill.description,
+    skill.detail,
+    skill.category,
+    skill.license,
+    skill.source,
+    skill.state,
+    skill.owner_username,
+    skill.created_at,
+    skill.last_activity_at,
+    skill.use_count != null ? `使用 ${skill.use_count}` : null,
+    skill.view_count != null ? `查看 ${skill.view_count}` : null,
+    skill.patch_count != null ? `变迁 ${skill.patch_count}` : null,
+    ...(skill.tags ?? []).flatMap((tag) => [
+      tag.name,
+      tag.description,
+      tag.color,
+      tag.usage_count != null ? `标签使用 ${tag.usage_count}` : null,
+    ]),
+  ]
+    .filter((value): value is string => typeof value === "string" && value.length > 0)
+    .join(" ")
+    .toLowerCase();
+  return haystack.includes(q);
 }
 
 export function SkillTransferSelector({
@@ -159,7 +180,7 @@ export function SkillTransferSelector({
         <div className="relative flex-1">
           <SearchIcon className="text-muted-foreground absolute left-2 top-1/2 size-3.5 -translate-y-1/2" />
           <Input
-            placeholder="搜索全部技能(中英文名称、描述)…"
+            placeholder="搜索技能名称、描述、分类、标签、变迁..."
             value={search}
             onChange={(e) => {
               setSearch(e.currentTarget.value);
@@ -239,7 +260,7 @@ export function SkillTransferSelector({
       {/* Two-column transfer panes */}
       <div className="grid grid-cols-2 gap-3">
         <SkillPane
-          title="可用技能"
+          title="未勾选技能"
           count={availableSkills.length}
           items={availableView}
           variant="available"
@@ -248,10 +269,10 @@ export function SkillTransferSelector({
           onPageChange={setAvailablePage}
           onPick={add}
           onViewDetail={onViewDetail}
-          emptyHint={query || category !== ALL_CATEGORIES ? "没有匹配的技能" : "已全部启用"}
+          emptyHint={query || category !== ALL_CATEGORIES ? "没有匹配的技能" : "已全部勾选"}
         />
         <SkillPane
-          title="已启用技能"
+          title="已勾选技能"
           count={selectedSkills.length}
           items={selectedView}
           variant="selected"
@@ -260,7 +281,7 @@ export function SkillTransferSelector({
           onPageChange={setSelectedPage}
           onPick={remove}
           onViewDetail={onViewDetail}
-          emptyHint={query ? "没有匹配的已选技能" : "尚未启用任何技能"}
+          emptyHint={query ? "没有匹配的已选技能" : "尚未勾选任何技能"}
         />
       </div>
     </div>

frontend-web/src/components/workspace/settings/skill-settings-page.tsx

@@ -470,11 +470,11 @@ function SkillSettingsList({ onClose }: { onClose?: () => void }) {
           </Button>
         </div>
       </header>
-      <TagSearchBar
-        targetType="skill"
-        onSearchChange={setSearch}
-        tagIds={tagIds}
-        onTagIdsChange={setTagIds}
+      <TagSearchBar
+        targetType="skill"
+        onSearchChange={setSearch}
+        tagIds={tagIds}
+        onTagIdsChange={setTagIds}
         actions={
           <div className="flex items-center gap-2">
             <ViewModeToggle value={viewMode} onChange={setViewMode} />
@@ -497,11 +497,16 @@ function SkillSettingsList({ onClose }: { onClose?: () => void }) {
                 <SelectItem value="name_desc">名称 Z-A</SelectItem>
               </SelectContent>
             </Select>
-          </div>
-        }
-      />
-      {publisher && (
-        <div className="flex items-center gap-2">
+          </div>
+        }
+      />
+      {!isLoading && !error && (
+        <div className="text-muted-foreground flex justify-end text-sm tabular-nums">
+          共 {orderedSkills.length} 个技能
+        </div>
+      )}
+      {publisher && (
+        <div className="flex items-center gap-2">
           <span className="text-muted-foreground text-xs">筛选发布者：</span>
           <button
             type="button"

frontend-web/src/core/plugin-management/domain-agent.test.mjs

@@ -187,4 +187,4 @@ test("unmapped incoming url resets auto-selected agent back to default when ther
       forceDefaultAssistant: true,
     },
   );
-});
+});

frontend-web/src/core/system-settings/api.ts

@@ -449,12 +449,16 @@ export interface ModelConcurrencySettings {
   enabled: boolean;
   default_limit: number;
   model_limits: Record<string, number>;
+  user_default_limit: number;
+  user_model_limits: Record<string, number>;
 }
 
 export interface ModelConcurrencyUpdate {
   enabled?: boolean;
   default_limit?: number;
   model_limits?: Record<string, number>;
+  user_default_limit?: number;
+  user_model_limits?: Record<string, number>;
 }
 
 export async function loadSkillCompressionSettings(): Promise<SkillCompressionSettings> {

frontend-web/src/core/threads/hooks.ts

@@ -263,11 +263,24 @@ function isModelConcurrencyLimitError(message: string): boolean {
   return (
     /\b429\b/.test(message) ||
     /并发已达到上限/.test(message) ||
+    /并发调用已达到上限/.test(message) ||
     /使用人数较多/.test(message) ||
     /too many requests/i.test(message)
   );
 }
 
+function friendlySendLimitMessage(message: string): string | null {
+  if (isActiveRunConflictError(message)) {
+    return "当前对话还在生成回答，请等待本轮回复结束，或先点“停止”再发送新的问题。";
+  }
+  if (isModelConcurrencyLimitError(message)) {
+    return /当前账号对该模型/.test(message)
+      ? "当前账号对该模型的并发调用已达到上限，请等待已有回答结束后再试。"
+      : "当前使用人数较多，请稍等一下再试。";
+  }
+  return null;
+}
+
 function buildCurrentUserLabel(): string {
   const auth = getStoredAuth();
   if (!auth) {
@@ -604,14 +617,11 @@ export function useThreadStream({
         }
         return;
       }
-      // 上一轮还没结束又发了新问题（后端 409）：给一句明确的中文提示，而不是
-      // 把原始的英文/HTTP 错误抛给用户。
-      if (isActiveRunConflictError(message)) {
-        toast.info("当前对话还在生成回答，请等待本轮回复结束，或先点「停止」再发送新的问题。");
-        return;
-      }
-      if (isModelConcurrencyLimitError(message)) {
-        toast.info("当前使用人数较多，请稍等一下再试。");
+      // 上一轮还没结束又发了新问题（409）或并发达到上限（429）：
+      // 给一句明确的中文提示，而不是把原始 HTTP 错误抛给用户。
+      const friendly = friendlySendLimitMessage(message);
+      if (friendly) {
+        toast.info(friendly);
         return;
       }
       toast.error(message);
@@ -1008,11 +1018,31 @@ export function useThreadStream({
           },
         );
         void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
-      } catch (error) {
-        setOptimisticMessages([]);
-        setIsUploading(false);
-        throw error;
-      } finally {
+      } catch (error) {
+        const message =
+          error instanceof Error ? error.message : String(error);
+        const friendly = friendlySendLimitMessage(message);
+        if (friendly) {
+          setOptimisticMessages((messages) => {
+            const human = messages.find((item) => item.type === "human");
+            const kept = human ? [human] : [];
+            return [
+              ...kept,
+              {
+                type: "ai",
+                id: `opt-ai-limit-${Date.now()}`,
+                content: friendly,
+                additional_kwargs: { element: "task" },
+              } as Message,
+            ];
+          });
+          setIsUploading(false);
+          return;
+        }
+        setOptimisticMessages([]);
+        setIsUploading(false);
+        throw error;
+      } finally {
         sendInFlightRef.current = false;
       }
     },

frontend-web/src/pages/IframeChatPage.tsx

@@ -707,4 +707,4 @@ export default function IframeChatPage() {
       </ChatBox>
     </ThreadContext.Provider>
   );
-}
+}

frontend-web/src/pages/ModelConcurrencySettingsPage.tsx

@@ -13,6 +13,21 @@ import {
   useUpdateModelConcurrencySettings,
 } from "@/core/system-settings/hooks";
 
+function cleanLimitMap(values: Record<string, string>, label: string): Record<string, number> | null {
+  const cleaned: Record<string, number> = {};
+  for (const [name, rawValue] of Object.entries(values)) {
+    const trimmed = rawValue.trim();
+    if (!trimmed) continue;
+    const parsed = Math.round(Number(trimmed));
+    if (!Number.isFinite(parsed) || parsed < 1) {
+      toast.error(`${label}「${name}」必须是大于 0 的整数`);
+      return null;
+    }
+    cleaned[name] = Math.min(parsed, 10000);
+  }
+  return cleaned;
+}
+
 export default function ModelConcurrencySettingsPage({
   embedded = false,
 }: {
@@ -24,13 +39,16 @@ export default function ModelConcurrencySettingsPage({
 
   const [enabled, setEnabled] = useState(false);
   const [defaultLimit, setDefaultLimit] = useState(10);
+  const [userDefaultLimit, setUserDefaultLimit] = useState(0);
   const [modelLimits, setModelLimits] = useState<Record<string, string>>({});
+  const [userModelLimits, setUserModelLimits] = useState<Record<string, string>>({});
   const initializedRef = useRef(false);
 
   useEffect(() => {
     if (!data || initializedRef.current) return;
     setEnabled(data.enabled);
     setDefaultLimit(data.default_limit);
+    setUserDefaultLimit(data.user_default_limit ?? 0);
     setModelLimits(
       Object.fromEntries(
         Object.entries(data.model_limits ?? {}).map(([name, limit]) => [
@@ -39,39 +57,45 @@ export default function ModelConcurrencySettingsPage({
         ]),
       ),
     );
+    setUserModelLimits(
+      Object.fromEntries(
+        Object.entries(data.user_model_limits ?? {}).map(([name, limit]) => [
+          name,
+          String(limit),
+        ]),
+      ),
+    );
     initializedRef.current = true;
   }, [data]);
 
   const modelNames = useMemo(() => models.map((model) => model.name), [models]);
 
-  const handleModelLimitChange = (modelName: string, value: string) => {
-    setModelLimits((prev) => ({ ...prev, [modelName]: value }));
-  };
-
   const handleSave = async () => {
     const nextDefault = Math.max(1, Math.min(Math.round(defaultLimit) || 10, 10000));
-    const cleaned: Record<string, number> = {};
-    for (const [name, rawValue] of Object.entries(modelLimits)) {
-      const trimmed = rawValue.trim();
-      if (!trimmed) continue;
-      const parsed = Math.round(Number(trimmed));
-      if (!Number.isFinite(parsed) || parsed < 1) {
-        toast.error(`模型 ${name} 的并发上限必须是大于 0 的整数`);
-        return;
-      }
-      cleaned[name] = Math.min(parsed, 10000);
-    }
+    const nextUserDefault = Math.max(0, Math.min(Math.round(userDefaultLimit) || 0, 10000));
+    const cleanedModelLimits = cleanLimitMap(modelLimits, "模型总上限");
+    if (!cleanedModelLimits) return;
+    const cleanedUserModelLimits = cleanLimitMap(userModelLimits, "单用户上限");
+    if (!cleanedUserModelLimits) return;
 
     try {
       await update.mutateAsync({
         enabled,
         default_limit: nextDefault,
-        model_limits: cleaned,
+        model_limits: cleanedModelLimits,
+        user_default_limit: nextUserDefault,
+        user_model_limits: cleanedUserModelLimits,
       });
       setDefaultLimit(nextDefault);
+      setUserDefaultLimit(nextUserDefault);
       setModelLimits(
         Object.fromEntries(
-          Object.entries(cleaned).map(([name, limit]) => [name, String(limit)]),
+          Object.entries(cleanedModelLimits).map(([name, limit]) => [name, String(limit)]),
+        ),
+      );
+      setUserModelLimits(
+        Object.fromEntries(
+          Object.entries(cleanedUserModelLimits).map(([name, limit]) => [name, String(limit)]),
         ),
       );
       toast.success("已保存");
@@ -85,7 +109,7 @@ export default function ModelConcurrencySettingsPage({
       className={
         embedded
           ? "flex w-full flex-col gap-6"
-          : "mx-auto flex h-full min-h-0 w-full max-w-3xl flex-col gap-6 overflow-y-auto px-6 py-10"
+          : "mx-auto flex h-full min-h-0 w-full max-w-4xl flex-col gap-6 overflow-y-auto px-6 py-10"
       }
     >
       <div>
@@ -93,7 +117,7 @@ export default function ModelConcurrencySettingsPage({
           模型并发限制
         </h1>
         <p className="text-muted-foreground mt-1 text-sm">
-          按模型限制同时进行中的问答数。达到上限后，新请求会收到“当前使用人数较多，请稍等一下再试”的提示。
+          按模型控制系统总并发，也可以限制单个用户对某个模型同时发起的调用数。达到上限时会在发送问题后立即提示，问题会保留在对话中。
         </p>
       </div>
 
@@ -102,42 +126,59 @@ export default function ModelConcurrencySettingsPage({
           <Loader2Icon className="text-muted-foreground size-6 animate-spin" />
         </div>
       ) : error ? (
-        <div className="text-destructive rounded-xl border border-dashed py-16 text-center text-sm">
+        <div className="text-destructive rounded-lg border border-dashed py-16 text-center text-sm">
           加载失败：{error.message}
         </div>
       ) : (
-        <div className="flex flex-col gap-5 rounded-xl border bg-white px-5 py-5 dark:border-[#213D5E] dark:bg-[#02244C] dark:text-white">
+        <div className="flex flex-col gap-5 rounded-lg border bg-white px-5 py-5 dark:border-[#213D5E] dark:bg-[#02244C] dark:text-white">
           <div className="flex items-center justify-between gap-4">
             <div>
               <div className="text-sm font-medium">开启限制</div>
               <div className="text-muted-foreground mt-0.5 text-xs">
-                关闭后，不再按模型并发数拦截问答请求。
+                关闭后，不再按模型或用户并发数拦截问答请求。
               </div>
             </div>
             <Switch checked={enabled} onCheckedChange={setEnabled} />
           </div>
 
-          <div className="flex flex-col gap-1.5">
-            <p className="text-sm font-medium">默认上限</p>
-            <p className="text-muted-foreground text-xs">
-              未单独配置的模型，使用这个并发上限。
-            </p>
-            <Input
-              type="number"
-              min={1}
-              max={10000}
-              value={defaultLimit}
-              disabled={!enabled}
-              onChange={(e) => setDefaultLimit(Number(e.target.value))}
-              className="w-32"
-            />
+          <div className="grid gap-4 md:grid-cols-2">
+            <div className="flex flex-col gap-1.5">
+              <p className="text-sm font-medium">模型默认总上限</p>
+              <p className="text-muted-foreground text-xs">
+                未单独配置的模型，使用这个系统总并发上限。
+              </p>
+              <Input
+                type="number"
+                min={1}
+                max={10000}
+                value={defaultLimit}
+                disabled={!enabled}
+                onChange={(e) => setDefaultLimit(Number(e.target.value))}
+                className="w-32"
+              />
+            </div>
+            <div className="flex flex-col gap-1.5">
+              <p className="text-sm font-medium">单用户默认上限</p>
+              <p className="text-muted-foreground text-xs">
+                同一用户对同一模型的默认并发上限。填 0 表示不启用默认用户限制。
+              </p>
+              <Input
+                type="number"
+                min={0}
+                max={10000}
+                value={userDefaultLimit}
+                disabled={!enabled}
+                onChange={(e) => setUserDefaultLimit(Number(e.target.value))}
+                className="w-32"
+              />
+            </div>
           </div>
 
           <div className="flex flex-col gap-3">
             <div>
               <p className="text-sm font-medium">按模型单独设置</p>
               <p className="text-muted-foreground mt-0.5 text-xs">
-                留空表示使用默认上限。
+                留空表示使用对应默认值；单用户上限填空且默认值为 0 时，不限制该维度。
               </p>
             </div>
 
@@ -150,24 +191,42 @@ export default function ModelConcurrencySettingsPage({
                 modelNames.map((modelName) => (
                   <div
                     key={modelName}
-                    className="flex items-center justify-between gap-4 rounded-lg border px-4 py-3"
+                    className="grid gap-3 rounded-lg border px-4 py-3 md:grid-cols-[minmax(0,1fr)_8rem_8rem]"
                   >
                     <div className="min-w-0">
                       <div className="truncate text-sm font-medium">{modelName}</div>
                       <div className="text-muted-foreground text-xs">
-                        留空时使用默认上限 {defaultLimit || 10}
+                        默认总上限 {defaultLimit || 10}；默认单用户上限 {userDefaultLimit || "不限制"}
                       </div>
                     </div>
-                    <Input
-                      type="number"
-                      min={1}
-                      max={10000}
-                      value={modelLimits[modelName] ?? ""}
-                      disabled={!enabled}
-                      onChange={(e) => handleModelLimitChange(modelName, e.target.value)}
-                      className="w-32"
-                      placeholder={String(defaultLimit || 10)}
-                    />
+                    <label className="flex flex-col gap-1">
+                      <span className="text-muted-foreground text-xs">模型总上限</span>
+                      <Input
+                        type="number"
+                        min={1}
+                        max={10000}
+                        value={modelLimits[modelName] ?? ""}
+                        disabled={!enabled}
+                        onChange={(e) =>
+                          setModelLimits((prev) => ({ ...prev, [modelName]: e.target.value }))
+                        }
+                        placeholder={String(defaultLimit || 10)}
+                      />
+                    </label>
+                    <label className="flex flex-col gap-1">
+                      <span className="text-muted-foreground text-xs">单用户上限</span>
+                      <Input
+                        type="number"
+                        min={1}
+                        max={10000}
+                        value={userModelLimits[modelName] ?? ""}
+                        disabled={!enabled}
+                        onChange={(e) =>
+                          setUserModelLimits((prev) => ({ ...prev, [modelName]: e.target.value }))
+                        }
+                        placeholder={userDefaultLimit > 0 ? String(userDefaultLimit) : "不限制"}
+                      />
+                    </label>
                   </div>
                 ))
               )}

frontend-web/src/pages/NewAgentPage.tsx

@@ -3,9 +3,7 @@
 import {
   ArrowLeftIcon,
   CheckCircleIcon,
-  InfoIcon,
   SaveIcon,
-  SearchIcon,
 } from "lucide-react";
 import { useParams, useRouter } from "next/navigation";
 import { useCallback, useEffect, useMemo, useState } from "react";
@@ -38,6 +36,7 @@ import {
   type SetupAgentApprovalPayload,
 } from "@/components/workspace/messages";
 import { ThreadContext } from "@/components/workspace/messages/context";
+import { SkillTransferSelector } from "@/components/workspace/agents/skill-transfer-selector";
 import { SkillDetailDialog } from "@/components/workspace/skills/skill-detail-dialog";
 import { getAPIClient } from "@/core/api";
 import type { Agent } from "@/core/agents";
@@ -51,7 +50,7 @@ import { getStoredAuth } from "@/core/auth";
 import { useLiveAccount } from "@/core/auth/use-live-account";
 import { useI18n } from "@/core/i18n/hooks";
 import { useModels } from "@/core/models/hooks";
-import { isAgentSkillCandidate, type Skill } from "@/core/skills";
+import { isAgentSkillCandidate } from "@/core/skills";
 import { useSkills } from "@/core/skills/hooks";
 import { useThreadStream } from "@/core/threads/hooks";
 import { uuid } from "@/core/utils/uuid";
@@ -98,7 +97,6 @@ export default function NewAgentPage() {
   const [selectedSkills, setSelectedSkills] = useState<string[]>([]);
   const [isSkillSelectionTouched, setIsSkillSelectionTouched] = useState(false);
   const [detailSkillName, setDetailSkillName] = useState<string | null>(null);
-  const [skillSearch, setSkillSearch] = useState("");
   // 恢复已有 bootstrap 会话时该 thread 已存在，无需再 create（置为 true，跳过创建）。
   const [threadPrepared, setThreadPrepared] = useState(!!resumeThreadId);
   const { skills, isLoading: isSkillsLoading } = useSkills();
@@ -123,17 +121,6 @@ export default function NewAgentPage() {
     [skills, storedAuth],
   );
 
-  const filteredSkills = useMemo(() => {
-    const q = skillSearch.trim().toLowerCase();
-    if (!q) return enabledSkills;
-    return enabledSkills.filter(
-      (skill) =>
-        skill.name.toLowerCase().includes(q) ||
-        (skill.name_zh ?? "").toLowerCase().includes(q) ||
-        (skill.description ?? "").toLowerCase().includes(q),
-    );
-  }, [enabledSkills, skillSearch]);
-
   useEffect(() => {
     if (isSkillSelectionTouched) {
       return;
@@ -380,14 +367,10 @@ export default function NewAgentPage() {
     [handleChatSubmit],
   );
 
-  const toggleSkill = useCallback((skill: Skill) => {
+  const handleSkillSelectionChange = useCallback((next: string[]) => {
     setIsSkillSelectionTouched(true);
     setFormError("");
-    setSelectedSkills((prev) =>
-      prev.includes(skill.name)
-        ? prev.filter((name) => name !== skill.name)
-        : [...prev, skill.name],
-    );
+    setSelectedSkills(next);
   }, []);
 
   const busy = isCreatingAgent || isSavingDraft || thread.isLoading;
@@ -624,68 +607,13 @@ export default function NewAgentPage() {
                       已选 {selectedSkills.length} 个
                     </span>
                   </div>
-                  {enabledSkills.length > 0 && (
-                    <div className="relative">
-                      <SearchIcon className="text-muted-foreground absolute left-2 top-1/2 size-3.5 -translate-y-1/2" />
-                      <Input
-                        placeholder="搜索技能名称或描述…"
-                        value={skillSearch}
-                        onChange={(e) => setSkillSearch(e.currentTarget.value)}
-                        className="h-8 pl-7 text-xs"
-                      />
-                    </div>
-                  )}
-                  <div className="max-h-[240px] overflow-y-auto rounded-md border p-2 dark:border-[#1B477E] dark:bg-[#022248]">
-                    {isSkillsLoading ? (
-                      <p className="text-muted-foreground text-sm">技能加载中...</p>
-                    ) : enabledSkills.length === 0 ? (
-                      <p className="text-muted-foreground text-sm">
-                        当前没有可用技能。
-                      </p>
-                    ) : filteredSkills.length === 0 ? (
-                      <p className="text-muted-foreground text-sm">
-                        没有匹配的技能。
-                      </p>
-                    ) : (
-                      <div className="grid grid-cols-1 gap-2 sm:grid-cols-2">
-                        {filteredSkills.map((skill) => {
-                          const checked = selectedSkills.includes(skill.name);
-                          return (
-                            <div
-                              key={skill.name}
-                              className="flex w-full min-w-0 items-center"
-                            >
-                              <Button
-                                type="button"
-                                size="sm"
-                                variant={checked ? "default" : "outline"}
-                                onClick={() => toggleSkill(skill)}
-                                title={skill.description}
-                                className={cn(
-                                  "h-8 min-w-0 flex-1 rounded-r-none",
-                                  !checked && "text-muted-foreground",
-                                )}
-                              >
-                                <span className="truncate">
-                                  {skill.name_zh?.trim() ? skill.name_zh : skill.name}
-                                </span>
-                              </Button>
-                              <Button
-                                type="button"
-                                size="icon"
-                                variant={checked ? "default" : "outline"}
-                                onClick={() => setDetailSkillName(skill.name)}
-                                title={t.settings.skills.viewDetailAction}
-                                className="h-8 w-7 rounded-l-none border-l-0"
-                              >
-                                <InfoIcon className="size-3.5" />
-                              </Button>
-                            </div>
-                          );
-                        })}
-                      </div>
-                    )}
-                  </div>
+                  <SkillTransferSelector
+                    skills={enabledSkills}
+                    selected={selectedSkills}
+                    onChange={handleSkillSelectionChange}
+                    isLoading={isSkillsLoading}
+                    onViewDetail={setDetailSkillName}
+                  />
                 </div>
 
                 <SkillDetailDialog

frontend-web/src/pages/PublicScheduledTaskPage.tsx

@@ -2,6 +2,7 @@
 
 import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
 import {
+  BookOpen,
   Check,
   ChevronLeft,
   ChevronRight,
@@ -23,6 +24,11 @@ import { toast } from "sonner";
 import { Badge } from "@/components/ui/badge";
 import { Button } from "@/components/ui/button";
 import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
+import {
+  Collapsible,
+  CollapsibleContent,
+  CollapsibleTrigger,
+} from "@/components/ui/collapsible";
 import {
   Dialog,
   DialogContent,
@@ -42,10 +48,12 @@ import {
 } from "@/components/ui/tooltip";
 import { MarkdownContent } from "@/components/workspace/messages/markdown-content";
 import { HtmlPreviewFrame } from "@/components/html-preview-frame";
+import { ReportMarkdownView } from "@/components/wiki/ReportMarkdownView";
 import { getBackendBaseURL } from "@/core/config";
 import { usePublicBackground } from "@/core/embed/background";
 import { downloadMarkdownAsDocx } from "@/core/scheduled-tasks/markdown-docx";
 import { reasoningPlugins } from "@/core/streamdown";
+import { createMarkdownAnchorResolver } from "@/core/utils/markdown";
 import { cn } from "@/lib/utils";
 
 type PublicTask = {
@@ -264,6 +272,190 @@ function stripTimestampSuffix(title: string): string {
   return cleaned.trim() || title;
 }
 
+function stripMarkdownToText(markdown: string): string {
+  return markdown
+    .replace(/```[\s\S]*?```/g, " ")
+    .replace(/~~~[\s\S]*?~~~/g, " ")
+    .replace(/!\[[^\]]*\]\([^)]+\)/g, " ")
+    .replace(/\[([^\]]+)\]\([^)]+\)/g, "$1")
+    .replace(/^#{1,6}\s+/gm, "")
+    .replace(/^[>\-*+\d.)\s]+/gm, "")
+    .replace(/[`*_~|]/g, "")
+    .replace(/<[^>]+>/g, " ")
+    .replace(/\s+/g, " ")
+    .trim();
+}
+
+function getMarkdownExcerpt(markdown: string): string {
+  const text = stripMarkdownToText(markdown);
+  if (!text) return "";
+  return text.length > 180 ? `${text.slice(0, 180)}...` : text;
+}
+
+function countReadableWords(markdown: string): number {
+  const text = stripMarkdownToText(markdown);
+  if (!text) return 0;
+  const cjk = text.match(/[\u4e00-\u9fff]/g)?.length ?? 0;
+  const words = text.replace(/[\u4e00-\u9fff]/g, " ").match(/[A-Za-z0-9]+/g)?.length ?? 0;
+  return cjk + words;
+}
+
+function selectorForId(id: string): string {
+  const css = window.CSS;
+  if (css?.escape) return `#${css.escape(id)}`;
+  return `[id="${id.replace(/\\/g, "\\\\").replace(/"/g, '\\"')}"]`;
+}
+
+function KnowledgeMarkdownView({
+  content,
+  title,
+  subtitle,
+  className,
+}: {
+  content: string;
+  title?: string;
+  subtitle?: string;
+  className?: string;
+  isLoading?: boolean;
+  remarkPlugins?: unknown;
+  rehypePlugins?: unknown;
+}) {
+  const rootRef = useRef<HTMLDivElement>(null);
+  const anchorResolver = useMemo(() => createMarkdownAnchorResolver(content), [content]);
+  const headings = useMemo(
+    () => anchorResolver.headings.filter((item) => item.level >= 1 && item.level <= 3),
+    [anchorResolver],
+  );
+  const [activeHeadingId, setActiveHeadingId] = useState<string | null>(null);
+  const excerpt = useMemo(() => getMarkdownExcerpt(content), [content]);
+  const sectionCount = headings.length;
+  const wordCount = useMemo(() => countReadableWords(content), [content]);
+
+  useEffect(() => {
+    setActiveHeadingId(headings[0]?.id ?? null);
+  }, [headings]);
+
+  useEffect(() => {
+    const root = rootRef.current;
+    if (!root || headings.length === 0) return;
+
+    const targets = headings
+      .map((item) => root.querySelector<HTMLElement>(selectorForId(item.id)))
+      .filter((item): item is HTMLElement => !!item);
+    if (targets.length === 0) return;
+
+    const observer = new IntersectionObserver(
+      (entries) => {
+        const visible = entries
+          .filter((entry) => entry.isIntersecting)
+          .sort((a, b) => a.boundingClientRect.top - b.boundingClientRect.top)[0];
+        if (visible?.target.id) {
+          setActiveHeadingId(visible.target.id);
+        }
+      },
+      {
+        root: root.closest("[data-radix-scroll-area-viewport]") ?? null,
+        rootMargin: "-96px 0px -65% 0px",
+        threshold: [0, 1],
+      },
+    );
+
+    targets.forEach((target) => observer.observe(target));
+    return () => observer.disconnect();
+  }, [headings, content]);
+
+  const scrollToHeading = useCallback((id: string) => {
+    const root = rootRef.current;
+    const target =
+      root?.querySelector<HTMLElement>(selectorForId(id)) ?? document.getElementById(id);
+    setActiveHeadingId(id);
+    target?.scrollIntoView({ behavior: "smooth", block: "start" });
+  }, []);
+
+  return (
+    <div
+      ref={rootRef}
+      className={cn(
+        "grid min-h-0 gap-5 lg:grid-cols-[minmax(0,1fr)_300px]",
+        className,
+      )}
+    >
+      <article className="min-w-0 overflow-hidden rounded-lg border border-slate-200/80 bg-white shadow-sm dark:border-slate-800 dark:bg-slate-950">
+        <div className="border-b border-slate-100 bg-slate-50/90 px-6 py-5 dark:border-slate-800 dark:bg-slate-900/70 md:px-8">
+          <div className="flex flex-wrap items-center gap-2 text-xs text-slate-500 dark:text-slate-400">
+            <span className="inline-flex items-center gap-1 rounded-md bg-white px-2 py-1 font-medium text-slate-600 ring-1 ring-slate-200 dark:bg-slate-950 dark:text-slate-300 dark:ring-slate-800">
+              <BookOpen className="size-3.5" />
+              知识梳理
+            </span>
+            {subtitle && <span className="truncate">{subtitle}</span>}
+          </div>
+          {title && (
+            <h1 className="mt-3 text-xl font-semibold leading-snug text-slate-950 dark:text-slate-50 md:text-2xl">
+              {title}
+            </h1>
+          )}
+          {excerpt && (
+            <p className="mt-3 max-w-4xl text-sm leading-7 text-slate-600 dark:text-slate-300">
+              {excerpt}
+            </p>
+          )}
+          <div className="mt-4 flex flex-wrap gap-2 text-xs text-slate-500 dark:text-slate-400">
+            <span className="rounded-md bg-white px-2 py-1 ring-1 ring-slate-200 dark:bg-slate-950 dark:ring-slate-800">
+              {sectionCount || 1} 个章节
+            </span>
+            <span className="rounded-md bg-white px-2 py-1 ring-1 ring-slate-200 dark:bg-slate-950 dark:ring-slate-800">
+              约 {wordCount} 字
+            </span>
+          </div>
+        </div>
+        <div className="wiki-markdown public-report-markdown mx-auto w-full max-w-[980px] px-6 py-7 md:px-10 md:py-9">
+          <MarkdownContent
+            className="max-w-none"
+            content={content}
+            isLoading={false}
+            remarkPlugins={reasoningPlugins.remarkPlugins}
+            rehypePlugins={reasoningPlugins.rehypePlugins}
+          />
+        </div>
+      </article>
+
+      <aside className="hidden min-w-0 lg:block">
+        <div className="sticky top-4 rounded-lg border border-slate-200/80 bg-white p-4 shadow-sm dark:border-slate-800 dark:bg-slate-950">
+          <div className="mb-2 flex items-center gap-2 text-sm font-semibold text-slate-800 dark:text-slate-100">
+            <List className="size-4" />
+            内容目录
+          </div>
+          {headings.length > 0 ? (
+            <nav className="max-h-[calc(100vh-210px)] space-y-1 overflow-auto pr-1">
+              {headings.map((item) => (
+                <button
+                  key={item.id}
+                  type="button"
+                  className={cn(
+                    "block w-full truncate rounded-md border-l-2 border-transparent px-2 py-1.5 text-left text-xs leading-5 text-slate-600 transition-colors hover:bg-slate-100 hover:text-slate-950 dark:text-slate-300 dark:hover:bg-slate-900 dark:hover:text-slate-50",
+                    activeHeadingId === item.id &&
+                      "border-primary bg-primary/10 font-medium text-primary dark:text-primary",
+                    item.level === 2 && "pl-4",
+                    item.level === 3 && "pl-7 text-slate-500 dark:text-slate-400",
+                  )}
+                  title={item.text}
+                  onClick={() => scrollToHeading(item.id)}
+                >
+                  {item.text}
+                </button>
+              ))}
+            </nav>
+          ) : (
+            <div className="rounded-md border border-dashed p-3 text-xs text-slate-500 dark:border-slate-800 dark:text-slate-400">
+              暂无标题目录
+            </div>
+          )}
+        </div>
+      </aside>
+    </div>
+  );
+}
+
 const BODY_TAB_VALUE = "__body__";
 const HTML_TAB_VALUE = "__html__";
 const RUN_LIST_COLLAPSED_WIDTH = 56;
@@ -688,12 +880,10 @@ function PublicFullscreenPreviewContent({ preview }: { preview: PublicFullscreen
   }
   if (preview.mode === "markdown") {
     return (
-      <MarkdownContent
+      <ReportMarkdownView
         className="max-w-none"
         content={preview.content || "无正文内容"}
-        isLoading={false}
-        remarkPlugins={reasoningPlugins.remarkPlugins}
-        rehypePlugins={reasoningPlugins.rehypePlugins}
+        title={preview.title}
       />
     );
   }
@@ -701,12 +891,10 @@ function PublicFullscreenPreviewContent({ preview }: { preview: PublicFullscreen
     const content = preview.file.content || "无文件内容";
     if (isMarkdownFile(preview.file)) {
       return (
-        <MarkdownContent
+        <ReportMarkdownView
           className="max-w-none"
           content={content}
-          isLoading={false}
-          remarkPlugins={reasoningPlugins.remarkPlugins}
-          rehypePlugins={reasoningPlugins.rehypePlugins}
+          title={preview.title}
         />
       );
     }
@@ -731,6 +919,7 @@ function publicFullscreenHref(taskId: string, runId: string, tab: ContentTab): s
 function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) {
   const queryClient = useQueryClient();
   const [fullscreenPreview, setFullscreenPreview] = useState<PublicFullscreenPreview | null>(null);
+  const [runMetaOpen, setRunMetaOpen] = useState(false);
 
   // HTML page run: render a sandboxed iframe preview instead of the MD tabs.
   const isHtmlRun =
@@ -970,7 +1159,29 @@ function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) {
               </TooltipContent>
             )}
           </Tooltip>
-          <p className="text-muted-foreground mt-1 text-xs">
+          <Collapsible open={runMetaOpen} onOpenChange={setRunMetaOpen} className="mt-2">
+            <CollapsibleTrigger asChild>
+              <Button
+                type="button"
+                variant="ghost"
+                size="sm"
+                className="text-muted-foreground hover:text-foreground h-7 px-2 text-xs"
+              >
+                <ChevronRight
+                  className={cn("size-3.5 transition-transform", runMetaOpen && "rotate-90")}
+                />
+                运行信息
+              </Button>
+            </CollapsibleTrigger>
+            <CollapsibleContent>
+              <div className="mt-2 grid gap-1 rounded-md border bg-muted/30 px-3 py-2 text-xs text-muted-foreground sm:grid-cols-3">
+                <div className="min-w-0 break-all">计划：{formatDate(run.scheduled_for)}</div>
+                <div>开始：{formatDate(run.started_at)}</div>
+                <div>完成：{formatDate(run.finished_at)}</div>
+              </div>
+            </CollapsibleContent>
+          </Collapsible>
+          <p className="hidden">
             开始：{formatDate(run.started_at)} · 完成：{formatDate(run.finished_at)}
           </p>
         </div>
@@ -1122,12 +1333,14 @@ function RunDetailPane({ run, taskId }: { run: PublicRun; taskId: string }) {
                       无内容。点击「编辑」开始添加。
                     </div>
                   ) : (
-                    <MarkdownContent
+                    <ReportMarkdownView
                       className="max-w-none pr-2"
                       content={activeSource}
-                      isLoading={false}
-                      remarkPlugins={reasoningPlugins.remarkPlugins}
-                      rehypePlugins={reasoningPlugins.rehypePlugins}
+                      title={
+                        activeTabMeta?.kind === "file"
+                          ? stripExtension(activeTabMeta.fileName || "")
+                          : run.result?.title
+                      }
                     />
                   )}
                 </ScrollArea>

frontend-web/src/pages/ScheduledTaskRunDetailPage.tsx

@@ -20,6 +20,7 @@ import {
 } from "@/components/workspace/workspace-container";
 import { MarkdownContent } from "@/components/workspace/messages/markdown-content";
 import { HtmlPreviewFrame } from "@/components/html-preview-frame";
+import { ReportMarkdownView } from "@/components/wiki/ReportMarkdownView";
 import {
   downloadScheduledTaskRun,
   downloadScheduledTaskRunFile,
@@ -91,11 +92,12 @@ function RenderPreview({ file }: { file: ScheduledTaskResultFileContent }) {
   }
   if (isMarkdownFile(file)) {
     return (
-      <MarkdownContent
+      <ReportMarkdownView
         className="max-w-none"
         content={file.content}
-        isLoading={false}
-        rehypePlugins={[]}
+        title={file.name.replace(/\.(md|markdown)$/i, "")}
+        badgeLabel="结果文件"
+        compactHeader
       />
     );
   }
@@ -114,11 +116,11 @@ function FullscreenPreviewContent({ preview }: { preview: FullscreenPreview }) {
   }
   if (preview.mode === "markdown") {
     return (
-      <MarkdownContent
+      <ReportMarkdownView
         className="max-w-none"
         content={preview.content || "无正文内容"}
-        isLoading={false}
-        rehypePlugins={[]}
+        title={preview.title}
+        badgeLabel="正文预览"
       />
     );
   }
@@ -653,11 +655,12 @@ export default function ScheduledTaskRunDetailPage() {
                     </Button>
                   </div>
                   {result.format === "markdown" || result.format === "mixed" ? (
-                    <MarkdownContent
+                    <ReportMarkdownView
                       className="max-w-none"
                       content={result.content || "无正文内容"}
-                      isLoading={false}
-                      rehypePlugins={[]}
+                      title={result.title || "正文预览"}
+                      subtitle={`计划时间：${formatDate(run.scheduled_for)}`}
+                      badgeLabel="任务报告"
                     />
                   ) : (
                     <pre className="whitespace-pre-wrap break-words text-sm">

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

@@ -853,8 +853,9 @@ async def _concurrency_sampler_loop(app: FastAPI) -> None:
         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)
+    store = getattr(app.state, "concurrency_sample_store", None)
+    run_mgr = getattr(app.state, "run_manager", None)
+    concurrency_gate = getattr(app.state, "concurrency_gate", None)
     if store is None or run_mgr is None:
         logger.info("Concurrency sampler not started (no store / run manager — memory backend?)")
         return
@@ -875,11 +876,16 @@ async def _concurrency_sampler_loop(app: FastAPI) -> None:
             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))
+                from deerflow.config.system_settings import get_concurrency_monitor_settings
+
+                if get_concurrency_monitor_settings().enabled:
+                    active_count = None
+                    if concurrency_gate is not None:
+                        active_count = await concurrency_gate.count_active()
+                    if active_count is None:
+                        active = await run_mgr.list_active()
+                        active_count = len(active)
+                    await store.record(active_count)
                     ticks += 1
                     if ticks % purge_every_ticks == 0:
                         from datetime import UTC, datetime, timedelta

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

@@ -367,8 +367,11 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
         run_events_config = getattr(config, "run_events", None)
         app.state.run_event_store = make_run_event_store(run_events_config)
 
-        # RunManager with store backing for persistence
-        app.state.run_manager = RunManager(store=app.state.run_store)
+        # RunManager with store backing for persistence
+        app.state.run_manager = RunManager(store=app.state.run_store)
+        from deerflow.runtime import make_redis_concurrency_gate
+
+        app.state.concurrency_gate = make_redis_concurrency_gate(config)
 
         try:
             yield
@@ -382,7 +385,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
                 await queue.stop()
                 set_auto_ingest_queue(None)
-            await close_engine()
+            await close_engine()
+            gate = getattr(app.state, "concurrency_gate", None)
+            if gate is not None:
+                await gate.aclose()
 
 
 # ---------------------------------------------------------------------------
@@ -408,7 +414,12 @@ get_run_manager: Callable[[Request], RunManager] = _require("run_manager", "Run
 get_checkpointer: Callable[[Request], Checkpointer] = _require("checkpointer", "Checkpointer")
 get_run_event_store: Callable[[Request], RunEventStore] = _require("run_event_store", "Run event store")
 get_feedback_repo: Callable[[Request], FeedbackRepository] = _require("feedback_repo", "Feedback")
-get_run_store: Callable[[Request], RunStore] = _require("run_store", "Run store")
+get_run_store: Callable[[Request], RunStore] = _require("run_store", "Run store")
+
+
+def get_concurrency_gate(request: Request):
+    """Return the optional cross-worker concurrency gate."""
+    return getattr(request.app.state, "concurrency_gate", None)
 
 
 def get_store(request: Request):

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

@@ -32,6 +32,7 @@ from app.gateway.roundtable_model_fallback import (
 )
 from app.gateway.roundtable_run_policy import roundtable_run_policy
 from app.gateway.roundtable_seat_skills import append_seat_skill_directive
+from app.gateway.services import RUN_SCOPE_BACKGROUND_CHILD
 from app.gateway.routers._roundtable_seed import (
     COORDINATOR_AGENT_ID,
     DASHBOARD_AGENT_ID,
@@ -346,6 +347,7 @@ async def _run_agent_turn(
     body = RunCreateRequest(
         assistant_id="lead_agent",
         input={"messages": [{"type": "human", "content": [{"type": "text", "text": message}], "additional_kwargs": {}}]},
+        metadata={"run_scope": RUN_SCOPE_BACKGROUND_CHILD},
         config={"recursion_limit": 1000},
         context={
             "agent_name": agent_name,
@@ -374,7 +376,12 @@ async def _run_agent_turn(
     # 落 checkpoint）。额度满时本轮在此 await 排队，等同用户某个在跑轮次释放额度后再开。
     async with _user_background_semaphore(user_id):
         try:
-            record = await start_run(body, thread_id, request)  # type: ignore[arg-type]
+            record = await start_run(  # type: ignore[arg-type]
+                body,
+                thread_id,
+                request,
+                run_scope=RUN_SCOPE_BACKGROUND_CHILD,
+            )
         except Exception as exc:  # noqa: BLE001 — start_run 报错（409 抢占 / DB 抖动等）也要落诊断
             if diag is not None:
                 await diag.record(

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

@@ -22,7 +22,7 @@ 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.deps import get_concurrency_gate, 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
@@ -149,6 +149,15 @@ def _elapsed_seconds(created_at: str, now: datetime) -> int:
     return max(0, int(delta))
 
 
+def _elapsed_seconds_from_epoch(created_at: float, now: datetime) -> int:
+    try:
+        if created_at <= 0:
+            return 0
+        return max(0, int(now.timestamp() - float(created_at)))
+    except Exception:
+        return 0
+
+
 def _kind_for(metadata: dict | None) -> str:
     """Coarse human label for the run's origin, from its thread metadata."""
     meta = metadata or {}
@@ -218,14 +227,46 @@ 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()
+    gate = get_concurrency_gate(request)
+    leases = await gate.list_active() if gate is not None else None
+    records = [] if leases is not None else await run_mgr.list_active()
     now = datetime.now(UTC)
-    maps = await _enrich(records)
+    enrich_source = leases if leases is not None else records
+    maps = await _enrich(enrich_source)
     thread_meta = maps["thread_meta"]
     user_emails = maps["user_emails"]
     agent_names = maps["agent_names"]
 
     items: list[ActiveRunItem] = []
+    if leases is not None:
+        for r in leases:
+            meta_row = thread_meta.get(r.thread_id)
+            user_id = r.user_id or (meta_row.user_id if meta_row else "") or ""
+            metadata = (meta_row.metadata_json if meta_row else None) 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="running",
+                    created_at=datetime.fromtimestamp(r.created_at, tz=UTC).isoformat() if r.created_at else "",
+                    elapsed_seconds=_elapsed_seconds_from_epoch(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,
+                    model_name=r.model_name,
+                    thread_title=(meta_row.display_name if meta_row else "") or "",
+                    kind=_kind_for(metadata),
+                    multitask_strategy="reject",
+                )
+            )
+        return ActiveRunsResponse(count=len(items), server_time=now.isoformat(), runs=items)
+
     for r in records:
         meta_row = thread_meta.get(r.thread_id)
         user_id = (meta_row.user_id if meta_row else "") or ""

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

@@ -65,6 +65,7 @@ from langgraph.checkpoint.base import empty_checkpoint
 from app.gateway.roundtable_diag import record_foreground_diag
 from app.gateway.roundtable_model_fallback import fallback_model_chain, is_llm_error_message, reason_text
 from app.gateway.roundtable_run_policy import merge_skill_stop_names, roundtable_run_policy
+from app.gateway.services import RUN_SCOPE_ORCHESTRATION_CHILD, internal_run_scope_headers
 from deerflow.agents.roundtable_orchestrator.extraction_spec import build_summary_directive
 
 from app.gateway.roundtable_seat_skills import append_seat_skill_directive
@@ -325,7 +326,11 @@ def _auth_headers(request: Request) -> dict[str, str]:
     token = request.headers.get("authorization")
     if not token:
         raise HTTPException(status_code=401, detail="Missing Authorization header")
-    return {"Authorization": token, "Content-Type": "application/json"}
+    return {
+        "Authorization": token,
+        "Content-Type": "application/json",
+        **internal_run_scope_headers(),
+    }
 
 
 async def _create_thread(
@@ -694,6 +699,7 @@ def _run_payload(
                 }
             ]
         },
+        "metadata": {"run_scope": RUN_SCOPE_ORCHESTRATION_CHILD},
         "config": {"recursion_limit": 1000},
         "context": {
             "agent_name": agent_name,

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

@@ -21,7 +21,6 @@ from deerflow.config.system_settings import (
     PublishSquare,
     PublishSquaresSettings,
     SkillCompressionSettings,
-    SystemSettings,
     get_module_tag_ids,
     list_publish_squares,
     load_system_settings,
@@ -223,12 +222,16 @@ class ModelConcurrencyResponse(BaseModel):
     enabled: bool = Field(..., description="Whether per-model concurrency limiting is active")
     default_limit: int = Field(..., description="Fallback max concurrent runs for models without an explicit override")
     model_limits: dict[str, int] = Field(default_factory=dict, description="Per-model overrides: model_name -> max concurrent runs")
+    user_default_limit: int = Field(..., description="Fallback max concurrent runs per user and model. 0 disables the per-user fallback")
+    user_model_limits: dict[str, int] = Field(default_factory=dict, description="Per-user per-model overrides: model_name -> max concurrent runs for one user")
 
 
 class ModelConcurrencyUpdateRequest(BaseModel):
     enabled: bool | None = None
     default_limit: int | None = Field(default=None, ge=1, le=10_000)
     model_limits: dict[str, int] | None = None
+    user_default_limit: int | None = Field(default=None, ge=0, le=10_000)
+    user_model_limits: dict[str, int] | None = None
 
 
 def _skill_compression_to_response(settings: SkillCompressionSettings) -> SkillCompressionResponse:
@@ -252,10 +255,24 @@ def _model_concurrency_to_response(settings: ModelConcurrencySettings) -> ModelC
         if normalized_limit < 1:
             continue
         cleaned[model_name] = min(normalized_limit, 10_000)
+    cleaned_user: dict[str, int] = {}
+    for name, limit in (settings.user_model_limits or {}).items():
+        model_name = str(name).strip()
+        if not model_name:
+            continue
+        try:
+            normalized_limit = int(limit)
+        except (TypeError, ValueError):
+            continue
+        if normalized_limit < 1:
+            continue
+        cleaned_user[model_name] = min(normalized_limit, 10_000)
     return ModelConcurrencyResponse(
         enabled=settings.enabled,
         default_limit=max(1, min(int(settings.default_limit), 10_000)),
         model_limits=cleaned,
+        user_default_limit=max(0, min(int(getattr(settings, "user_default_limit", 0) or 0), 10_000)),
+        user_model_limits=cleaned_user,
     )
 
 
@@ -333,6 +350,22 @@ async def update_model_concurrency(
                 continue
             cleaned[model_name] = min(normalized_limit, 10_000)
         update["model_limits"] = cleaned
+    if "user_default_limit" in fields and fields["user_default_limit"] is not None:
+        update["user_default_limit"] = max(0, min(int(fields["user_default_limit"]), 10_000))
+    if "user_model_limits" in fields and fields["user_model_limits"] is not None:
+        cleaned_user: dict[str, int] = {}
+        for name, limit in (body.user_model_limits or {}).items():
+            model_name = str(name).strip()
+            if not model_name:
+                continue
+            try:
+                normalized_limit = int(limit)
+            except (TypeError, ValueError):
+                continue
+            if normalized_limit < 1:
+                continue
+            cleaned_user[model_name] = min(normalized_limit, 10_000)
+        update["user_model_limits"] = cleaned_user
     settings = current.model_concurrency.model_copy(update=update)
     saved = save_system_settings(current.model_copy(update={"model_concurrency": settings}))
     return _model_concurrency_to_response(saved.model_concurrency)

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

@@ -10,15 +10,17 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+import os
 import re
 import time
+import uuid
 from collections.abc import Mapping
 from typing import Any
 
 from fastapi import HTTPException, Request
 from langchain_core.messages import HumanMessage
 
-from app.gateway.deps import get_run_context, get_run_manager, get_stream_bridge
+from app.gateway.deps import get_concurrency_gate, get_run_context, get_run_manager, get_stream_bridge
 from app.gateway.utils import sanitize_log_param
 from deerflow.config import get_app_config
 from deerflow.config.agents_config import load_agent_config, validate_agent_id
@@ -36,9 +38,105 @@ from deerflow.runtime import (
     UnsupportedStrategyError,
     run_agent,
 )
+from deerflow.runtime.user_context import get_effective_user_id
 
 logger = logging.getLogger(__name__)
 
+RUN_SCOPE_EXTERNAL_ROOT = "external_root"
+RUN_SCOPE_ORCHESTRATION_CHILD = "orchestration_child"
+RUN_SCOPE_BACKGROUND_CHILD = "background_child"
+_INTERNAL_RUN_SCOPES = {RUN_SCOPE_ORCHESTRATION_CHILD, RUN_SCOPE_BACKGROUND_CHILD}
+_ALL_RUN_SCOPES = {RUN_SCOPE_EXTERNAL_ROOT, *_INTERNAL_RUN_SCOPES}
+_INTERNAL_RUN_SCOPE_HEADER = "x-deerflow-internal-run-scope-token"
+_INTERNAL_RUN_SCOPE_TOKEN = (
+    os.getenv("DEERFLOW_INTERNAL_RUN_SCOPE_TOKEN", "").strip()
+    or uuid.uuid4().hex
+)
+
+
+def internal_run_scope_headers() -> dict[str, str]:
+    """Headers used by same-process loopback calls to mark internal child runs."""
+
+    return {_INTERNAL_RUN_SCOPE_HEADER: _INTERNAL_RUN_SCOPE_TOKEN}
+
+
+def _metadata_run_scope(body: Any) -> str | None:
+    metadata = getattr(body, "metadata", None) or {}
+    if not isinstance(metadata, dict):
+        return None
+    raw = str(metadata.get("run_scope") or "").strip()
+    return raw or None
+
+
+def _request_is_loopback(request: Request) -> bool:
+    client = getattr(request, "client", None)
+    host = str(getattr(client, "host", "") or "").strip()
+    return host in {"127.0.0.1", "::1", "localhost"}
+
+
+def _has_internal_run_scope_header(request: Request) -> bool:
+    headers = getattr(request, "headers", None)
+    token = headers.get(_INTERNAL_RUN_SCOPE_HEADER) if headers is not None else None
+    return str(token or "") == _INTERNAL_RUN_SCOPE_TOKEN
+
+
+def _resolve_run_scope(body: Any, request: Request, run_scope: str | None) -> str:
+    """Return the trusted run scope used for concurrency accounting."""
+
+    if run_scope:
+        scope = str(run_scope).strip()
+        return scope if scope in _ALL_RUN_SCOPES else RUN_SCOPE_EXTERNAL_ROOT
+
+    scope = _metadata_run_scope(body)
+    if (
+        scope in _INTERNAL_RUN_SCOPES
+        and _request_is_loopback(request)
+        and _has_internal_run_scope_header(request)
+    ):
+        return scope
+    if scope == RUN_SCOPE_EXTERNAL_ROOT:
+        return RUN_SCOPE_EXTERNAL_ROOT
+    return RUN_SCOPE_EXTERNAL_ROOT
+
+
+async def _run_agent_with_concurrency_lease(
+    *,
+    concurrency_gate: Any | None,
+    lease_acquired: bool,
+    lease_metadata: dict[str, Any],
+    heartbeat_interval_seconds: int,
+    run_agent_kwargs: dict[str, Any],
+) -> None:
+    """Run the agent while refreshing and finally releasing an optional lease."""
+
+    heartbeat_task: asyncio.Task | None = None
+
+    async def _heartbeat_loop() -> None:
+        if concurrency_gate is None or not lease_acquired:
+            return
+        interval = max(5, int(heartbeat_interval_seconds or 20))
+        while True:
+            await asyncio.sleep(interval)
+            await concurrency_gate.heartbeat(**lease_metadata)
+
+    if concurrency_gate is not None and lease_acquired:
+        heartbeat_task = asyncio.create_task(
+            _heartbeat_loop(),
+            name=f"concurrency-heartbeat:{lease_metadata.get('run_id')}",
+        )
+
+    try:
+        await run_agent(**run_agent_kwargs)
+    finally:
+        if heartbeat_task is not None:
+            heartbeat_task.cancel()
+            try:
+                await heartbeat_task
+            except asyncio.CancelledError:
+                pass
+        if concurrency_gate is not None and lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
+
 
 # ---------------------------------------------------------------------------
 # SSE formatting
@@ -498,6 +596,49 @@ def _resolve_effective_run_model_name(body: Any) -> str:
     return default_model_name
 
 
+def _resolve_effective_model_name(body: Any, app_config: Any | None = None) -> str:
+    """Compatibility wrapper for resolving the effective run model."""
+    if app_config is None:
+        return _resolve_effective_run_model_name(body)
+
+    default_model_name = app_config.models[0].name if app_config.models else None
+    if not default_model_name:
+        raise ValueError("No chat models are configured. Please configure at least one model in config.yaml.")
+
+    requested_model_name = _extract_requested_model_name(body)
+    if requested_model_name and app_config.get_model_config(requested_model_name):
+        return requested_model_name
+
+    agent_id = _extract_effective_agent_id(body)
+    if agent_id:
+        try:
+            agent_config = load_agent_config(agent_id)
+        except Exception:
+            agent_config = None
+        agent_model_name = str(getattr(agent_config, "model", "") or "").strip()
+        if agent_model_name and app_config.get_model_config(agent_model_name):
+            return agent_model_name
+
+    return default_model_name
+
+
+def _resolve_model_concurrency_limit(model_name: str, app_config: Any) -> int | None:
+    """Read legacy per-model max-concurrency values from model config extras."""
+    model_config = app_config.get_model_config(model_name)
+    extras = getattr(model_config, "model_extra", None) or {}
+    for key in ("max_concurrency", "model_concurrency_limit"):
+        raw = extras.get(key)
+        if raw is None:
+            continue
+        try:
+            limit = int(raw)
+        except (TypeError, ValueError):
+            continue
+        if limit > 0:
+            return limit
+    return None
+
+
 def _get_model_concurrency_limit(model_name: str) -> int | None:
     """Return the configured limit for a model, or ``None`` when limiting is off."""
     settings = load_system_settings().model_concurrency
@@ -512,6 +653,25 @@ def _get_model_concurrency_limit(model_name: str) -> int | None:
     return max(1, limit)
 
 
+def _get_user_model_concurrency_limit(model_name: str) -> int | None:
+    """Return the configured per-user limit for a model, or ``None`` when off."""
+    settings = load_system_settings().model_concurrency
+    if not settings.enabled:
+        return None
+
+    raw_limit = settings.user_model_limits.get(
+        model_name,
+        getattr(settings, "user_default_limit", 0) or 0,
+    )
+    try:
+        limit = int(raw_limit)
+    except (TypeError, ValueError):
+        limit = int(getattr(settings, "user_default_limit", 0) or 0)
+    if limit < 1:
+        return None
+    return max(1, limit)
+
+
 async def _sync_legacy_agent_records(store) -> None:
     """Import existing folder-backed agents as built-in records.
 
@@ -647,6 +807,7 @@ async def start_run(
     request: Request,
     *,
     enforce_access: bool = True,
+    run_scope: str | None = None,
 ) -> RunRecord:
     """Create a RunRecord and launch the background agent task.
 
@@ -674,6 +835,7 @@ async def start_run(
     bridge = get_stream_bridge(request)
     run_mgr = get_run_manager(request)
     run_ctx = get_run_context(request)
+    concurrency_gate = get_concurrency_gate(request)
 
     disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_
 
@@ -688,10 +850,56 @@ async def start_run(
     t_post_enforce = time.perf_counter()
 
     effective_model_name = _resolve_effective_run_model_name(body)
+    effective_user_id = get_effective_user_id()
+    effective_run_scope = _resolve_run_scope(body, request, run_scope)
     model_concurrency_limit = _get_model_concurrency_limit(effective_model_name)
+    user_model_concurrency_limit = (
+        _get_user_model_concurrency_limit(effective_model_name)
+        if effective_run_scope == RUN_SCOPE_EXTERNAL_ROOT
+        else None
+    )
+    run_id = str(uuid.uuid4())
 
     run_metadata = dict(body.metadata or {})
     run_metadata["model_name"] = effective_model_name
+    run_metadata["user_id"] = effective_user_id
+    run_metadata["run_scope"] = effective_run_scope
+
+    redis_lease_acquired = False
+    redis_gate_degraded = True
+    user_model_accounted = user_model_concurrency_limit is not None
+    lease_metadata = {
+        "run_id": run_id,
+        "thread_id": thread_id,
+        "model_name": effective_model_name,
+        "user_id": effective_user_id,
+        "user_model_accounted": user_model_accounted,
+    }
+    if concurrency_gate is not None:
+        acquire = await concurrency_gate.try_acquire(
+            run_id=run_id,
+            thread_id=thread_id,
+            model_name=effective_model_name,
+            user_id=effective_user_id,
+            assistant_id=body.assistant_id or "",
+            model_limit=model_concurrency_limit,
+            user_model_limit=user_model_concurrency_limit,
+            reject_thread=body.multitask_strategy == "reject",
+        )
+        redis_lease_acquired = acquire.acquired
+        redis_gate_degraded = acquire.degraded
+        if acquire.rejected:
+            if acquire.reason == "thread":
+                raise HTTPException(
+                    status_code=409,
+                    detail="Current thread already has an active run. Please wait or stop it before sending another message.",
+                )
+            if acquire.reason == "user_model":
+                raise HTTPException(
+                    status_code=429,
+                    detail="This account has reached the concurrency limit for the selected model. Please try again later.",
+                )
+            raise HTTPException(status_code=429, detail="The selected model is busy. Please try again later.")
 
     try:
         record = await run_mgr.create_or_reject(
@@ -701,19 +909,35 @@ async def start_run(
             metadata=run_metadata,
             kwargs={"input": body.input, "config": body.config},
             multitask_strategy=body.multitask_strategy,
-            model_concurrency_limit=(effective_model_name, model_concurrency_limit) if model_concurrency_limit is not None else None,
+            model_concurrency_limit=(effective_model_name, model_concurrency_limit) if redis_gate_degraded and model_concurrency_limit is not None else None,
+            user_model_concurrency_limit=(effective_model_name, effective_user_id, user_model_concurrency_limit) if redis_gate_degraded and user_model_concurrency_limit is not None else None,
+            run_id=run_id,
         )
     except ModelConcurrencyLimitError as exc:
-        raise HTTPException(status_code=429, detail="当前使用人数较多，请稍等一下再试。") from exc
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
+        if getattr(exc, "scope", "") == "user_model":
+            detail = "This account has reached the concurrency limit for the selected model. Please try again later."
+        else:
+            detail = "The selected model is busy. Please try again later."
+        raise HTTPException(status_code=429, detail=detail) from exc
     except ConflictError as exc:
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
         # 同一会话上一轮回答还没结束，又提交了新问题 → 拒绝。给用户中文提示，
         # 而不是把内部英文「Thread ... already has an active run」直接抛到前端。
         raise HTTPException(
             status_code=409,
-            detail="当前对话还在生成回答，请等待本轮回复结束，或先点「停止」再发送新的问题。",
+            detail="Current thread already has an active run. Please wait or stop it before sending another message.",
         ) from exc
     except UnsupportedStrategyError as exc:
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
         raise HTTPException(status_code=501, detail=str(exc)) from exc
+    except Exception:
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
+        raise
     t_post_create = time.perf_counter()
 
     # Upsert thread metadata so the thread appears in /threads/search,
@@ -751,7 +975,12 @@ async def start_run(
         logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
     t_post_thread_upsert = time.perf_counter()
 
-    agent_factory = resolve_agent_factory(body.assistant_id)
+    try:
+        agent_factory = resolve_agent_factory(body.assistant_id)
+    except Exception:
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
+        raise
     graph_input = normalize_input(body.input)
 
     # Admin-configurable prompt prefix: when the caller opted in via
@@ -789,24 +1018,38 @@ async def start_run(
     stream_modes = normalize_stream_modes(body.stream_mode)
 
     t_pre_task = time.perf_counter()
-    task = asyncio.create_task(
-        run_agent(
-            bridge,
-            run_mgr,
-            record,
-            ctx=run_ctx,
-            agent_factory=agent_factory,
-            graph_input=graph_input,
-            config=config,
-            stream_modes=stream_modes,
-            stream_subgraphs=body.stream_subgraphs,
-            interrupt_before=body.interrupt_before,
-            interrupt_after=body.interrupt_after,
-            # 透传 SDK ``Command``：resume / update / goto。Pregel 会据此从最近
-            # checkpoint 处恢复，而不是把 graph_input 当新一轮输入跑。
-            command=getattr(body, "command", None),
+    heartbeat_interval_seconds = 20
+    if concurrency_gate is not None:
+        heartbeat_interval_seconds = int(
+            getattr(getattr(concurrency_gate, "config", None), "heartbeat_interval_seconds", 20) or 20
         )
-    )
+    try:
+        task = asyncio.create_task(
+            _run_agent_with_concurrency_lease(
+                concurrency_gate=concurrency_gate,
+                lease_acquired=redis_lease_acquired,
+                lease_metadata=lease_metadata,
+                heartbeat_interval_seconds=heartbeat_interval_seconds,
+                run_agent_kwargs={
+                    "bridge": bridge,
+                    "run_manager": run_mgr,
+                    "record": record,
+                    "ctx": run_ctx,
+                    "agent_factory": agent_factory,
+                    "graph_input": graph_input,
+                    "config": config,
+                    "stream_modes": stream_modes,
+                    "stream_subgraphs": body.stream_subgraphs,
+                    "interrupt_before": body.interrupt_before,
+                    "interrupt_after": body.interrupt_after,
+                    "command": getattr(body, "command", None),
+                },
+            )
+        )
+    except Exception:
+        if concurrency_gate is not None and redis_lease_acquired:
+            await concurrency_gate.release(**lease_metadata)
+        raise
     record.task = task
 
     # Title sync is handled by worker.py's finally block which reads the

offline-backend-20260512/backend/config.yaml

@@ -9,6 +9,29 @@ log_level: info
 gateway:
   port: 8001
 
+# Optional Redis-backed global concurrency coordination.
+# Leave disabled to keep the existing in-process behavior.
+# When enabled, Redis operations are short-timeout and fail-open by default:
+# if Redis is slow or unavailable, user Q&A continues with local-only limits.
+concurrency:
+  redis:
+    enabled: false
+    url: redis://127.0.0.1:6379/0
+    password: null
+    # Prefix with the deployment/system id because Redis may be shared by
+    # multiple projects. Keep this unique per system to avoid cross-project
+    # leases affecting each other.
+    namespace: cmzs_deerflow:concurrency
+    socket_connect_timeout_ms: 80
+    socket_timeout_ms: 80
+    acquire_timeout_ms: 120
+    operation_timeout_ms: 80
+    circuit_breaker_failures: 3
+    circuit_breaker_cooldown_seconds: 30
+    lease_ttl_seconds: 90
+    heartbeat_interval_seconds: 20
+    fail_open: true
+
 models:
   - name: deepseek-chat
     display_name: DeepSeek Chat

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

@@ -11,8 +11,9 @@ from pydantic import BaseModel, ConfigDict, Field
 from deerflow.config.acp_config import ACPAgentConfig, load_acp_config_from_dict
 from deerflow.config.agents_api_config import AgentsApiConfig, load_agents_api_config_from_dict
 from deerflow.config.browser_context_config import BrowserContextConfig
-from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
-from deerflow.config.database_config import DatabaseConfig
+from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
+from deerflow.config.concurrency_config import ConcurrencyConfig
+from deerflow.config.database_config import DatabaseConfig
 from deerflow.config.extensions_config import ExtensionsConfig
 from deerflow.config.gateway_config import GatewayConfig
 from deerflow.config.guardrails_config import GuardrailsConfig, load_guardrails_config_from_dict
@@ -137,8 +138,9 @@ class AppConfig(BaseModel):
     gateway: GatewayConfig = Field(default_factory=GatewayConfig, description="Gateway FastAPI listening port (used by intra-process loopback)")
     run_events: RunEventsConfig = Field(default_factory=RunEventsConfig, description="Run event storage configuration")
     checkpointer: CheckpointerConfig | None = Field(default=None, description="Checkpointer configuration")
-    stream_bridge: StreamBridgeConfig | None = Field(default=None, description="Stream bridge configuration")
-    browser_context: BrowserContextConfig = Field(default_factory=BrowserContextConfig, description="Optional browser page capture integration")
+    stream_bridge: StreamBridgeConfig | None = Field(default=None, description="Stream bridge configuration")
+    concurrency: ConcurrencyConfig = Field(default_factory=ConcurrencyConfig, description="Optional cross-worker concurrency coordination")
+    browser_context: BrowserContextConfig = Field(default_factory=BrowserContextConfig, description="Optional browser page capture integration")
     knowledge: KnowledgeConfig = Field(default_factory=KnowledgeConfig, description="Product knowledge base (obsidian-wiki compatible vault) configuration")
     knowledge_ingest: KnowledgeIngestConfig = Field(default_factory=KnowledgeIngestConfig, description="External knowledge-ingest endpoint configuration")
     task_deeplink: TaskDeeplinkConfig = Field(default_factory=TaskDeeplinkConfig, description="任务深链(rwfx)目的树闸门与跳转地址配置(前端经 /api/public/config/task-deeplink 读取)")

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

@@ -0,0 +1,38 @@
+"""Runtime concurrency coordination configuration."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class RedisConcurrencyConfig(BaseModel):
+    """Optional Redis-backed cross-worker concurrency coordination."""
+
+    enabled: bool = Field(default=False, description="Enable Redis-backed global run concurrency coordination.")
+    url: str | None = Field(default=None, description="Redis URL, for example redis://127.0.0.1:6379/0.")
+    password: str | None = Field(
+        default=None,
+        description="Optional Redis password. You may also include credentials directly in url.",
+    )
+    namespace: str = Field(
+        default="cmzs_deerflow:concurrency",
+        description="Redis key prefix for concurrency leases. Use a deployment/system prefix when Redis is shared.",
+    )
+    socket_connect_timeout_ms: int = Field(default=80, ge=1, le=10_000)
+    socket_timeout_ms: int = Field(default=80, ge=1, le=10_000)
+    acquire_timeout_ms: int = Field(default=120, ge=1, le=10_000)
+    operation_timeout_ms: int = Field(default=80, ge=1, le=10_000)
+    circuit_breaker_failures: int = Field(default=3, ge=1, le=100)
+    circuit_breaker_cooldown_seconds: int = Field(default=30, ge=1, le=3600)
+    lease_ttl_seconds: int = Field(default=90, ge=10, le=3600)
+    heartbeat_interval_seconds: int = Field(default=20, ge=5, le=600)
+    fail_open: bool = Field(
+        default=True,
+        description="Allow runs to proceed with local-only limits when Redis is unavailable or slow.",
+    )
+
+
+class ConcurrencyConfig(BaseModel):
+    """Top-level concurrency coordination config."""
+
+    redis: RedisConcurrencyConfig = Field(default_factory=RedisConcurrencyConfig)

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

@@ -286,6 +286,16 @@ class ModelConcurrencySettings(BaseModel):
         default_factory=dict,
         description="Optional per-model overrides: model_name -> max concurrent runs.",
     )
+    user_default_limit: int = Field(
+        default=0,
+        ge=0,
+        le=10_000,
+        description="Fallback max concurrent runs per user and model. 0 disables the per-user fallback.",
+    )
+    user_model_limits: dict[str, int] = Field(
+        default_factory=dict,
+        description="Optional per-user per-model overrides: model_name -> max concurrent runs for one user.",
+    )
     model_config = ConfigDict(extra="ignore")
 
 

offline-backend-20260512/backend/packages/harness/deerflow/runtime/__init__.py

@@ -7,14 +7,19 @@ directly from ``deerflow.runtime``.
 
 from .checkpointer import checkpointer_context, get_checkpointer, make_checkpointer, reset_checkpointer
 from .runs import (
+    AcquireDecision,
+    AcquireResult,
+    ActiveLease,
     ConflictError,
     DisconnectMode,
     ModelConcurrencyLimitError,
+    RedisConcurrencyGate,
     RunContext,
     RunManager,
     RunRecord,
     RunStatus,
     UnsupportedStrategyError,
+    make_redis_concurrency_gate,
     run_agent,
 )
 from .serialization import serialize, serialize_channel_values, serialize_lc_object, serialize_messages_tuple
@@ -28,14 +33,19 @@ __all__ = [
     "make_checkpointer",
     "reset_checkpointer",
     # runs
+    "AcquireDecision",
+    "AcquireResult",
+    "ActiveLease",
     "ConflictError",
     "DisconnectMode",
     "ModelConcurrencyLimitError",
+    "RedisConcurrencyGate",
     "RunContext",
     "RunManager",
     "RunRecord",
     "RunStatus",
     "UnsupportedStrategyError",
+    "make_redis_concurrency_gate",
     "run_agent",
     # serialization
     "serialize",

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

@@ -2,16 +2,22 @@
 
 from .manager import ConflictError, ModelConcurrencyLimitError, RunManager, RunRecord, UnsupportedStrategyError
 from .schemas import DisconnectMode, RunStatus
+from .concurrency_gate import AcquireDecision, AcquireResult, ActiveLease, RedisConcurrencyGate, make_redis_concurrency_gate
 from .worker import RunContext, run_agent
 
 __all__ = [
+    "AcquireDecision",
+    "AcquireResult",
+    "ActiveLease",
     "ConflictError",
     "DisconnectMode",
     "ModelConcurrencyLimitError",
+    "RedisConcurrencyGate",
     "RunContext",
     "RunManager",
     "RunRecord",
     "RunStatus",
     "UnsupportedStrategyError",
+    "make_redis_concurrency_gate",
     "run_agent",
 ]

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

@@ -0,0 +1,379 @@
+"""Optional cross-worker run concurrency gates.
+
+The Redis gate is deliberately fail-open by default: Redis coordinates global
+limits when it is fast and healthy, but it must not block the chat path when it
+is slow or unavailable.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+from deerflow.config.concurrency_config import RedisConcurrencyConfig
+
+logger = logging.getLogger(__name__)
+
+
+class AcquireDecision(str, Enum):
+    ACQUIRED = "acquired"
+    REJECTED = "rejected"
+    DEGRADED = "degraded"
+
+
+@dataclass(frozen=True)
+class AcquireResult:
+    decision: AcquireDecision
+    reason: str = ""
+    source: str = "redis"
+
+    @property
+    def acquired(self) -> bool:
+        return self.decision == AcquireDecision.ACQUIRED
+
+    @property
+    def rejected(self) -> bool:
+        return self.decision == AcquireDecision.REJECTED
+
+    @property
+    def degraded(self) -> bool:
+        return self.decision == AcquireDecision.DEGRADED
+
+
+@dataclass(frozen=True)
+class ActiveLease:
+    run_id: str
+    thread_id: str = ""
+    model_name: str = ""
+    user_id: str = ""
+    assistant_id: str = ""
+    created_at: float = 0.0
+    expires_at: float = 0.0
+
+
+class CircuitBreaker:
+    """Tiny in-process circuit breaker for Redis failures."""
+
+    def __init__(self, *, failure_threshold: int, cooldown_seconds: int) -> None:
+        self._failure_threshold = max(1, int(failure_threshold))
+        self._cooldown_seconds = max(1, int(cooldown_seconds))
+        self._failures = 0
+        self._opened_until = 0.0
+
+    def allow(self) -> bool:
+        return time.monotonic() >= self._opened_until
+
+    def record_success(self) -> None:
+        self._failures = 0
+        self._opened_until = 0.0
+
+    def record_failure(self) -> None:
+        self._failures += 1
+        if self._failures >= self._failure_threshold:
+            self._opened_until = time.monotonic() + self._cooldown_seconds
+
+    @property
+    def open(self) -> bool:
+        return not self.allow()
+
+
+class RedisConcurrencyGate:
+    """Redis lease-based global concurrency limiter."""
+
+    _ACQUIRE_SCRIPT = """
+local now = tonumber(ARGV[1])
+local expires_at = tonumber(ARGV[2])
+local run_id = ARGV[3]
+local model_limit = tonumber(ARGV[4])
+local user_model_limit = tonumber(ARGV[5])
+local reject_thread = tonumber(ARGV[6])
+local payload = ARGV[7]
+local ttl_ms = tonumber(ARGV[8])
+
+for i = 1, 4 do
+  redis.call('ZREMRANGEBYSCORE', KEYS[i], '-inf', now)
+end
+
+if reject_thread == 1 and redis.call('ZCARD', KEYS[4]) > 0 then
+  return {'rejected', 'thread'}
+end
+if model_limit > 0 and redis.call('ZCARD', KEYS[2]) >= model_limit then
+  return {'rejected', 'model'}
+end
+if user_model_limit > 0 and redis.call('ZCARD', KEYS[3]) >= user_model_limit then
+  return {'rejected', 'user_model'}
+end
+
+redis.call('ZADD', KEYS[1], expires_at, run_id)
+redis.call('ZADD', KEYS[2], expires_at, run_id)
+if user_model_limit > 0 then
+  redis.call('ZADD', KEYS[3], expires_at, run_id)
+end
+redis.call('ZADD', KEYS[4], expires_at, run_id)
+redis.call('SET', KEYS[5], payload, 'PX', ttl_ms)
+return {'acquired', ''}
+"""
+
+    def __init__(self, config: RedisConcurrencyConfig) -> None:
+        self.config = config
+        self._client: Any | None = None
+        self._breaker = CircuitBreaker(
+            failure_threshold=config.circuit_breaker_failures,
+            cooldown_seconds=config.circuit_breaker_cooldown_seconds,
+        )
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self.config.enabled and self.config.url)
+
+    def health(self) -> dict[str, Any]:
+        return {
+            "enabled": self.enabled,
+            "source": "redis",
+            "circuit_open": self._breaker.open,
+            "fail_open": self.config.fail_open,
+        }
+
+    async def aclose(self) -> None:
+        client = self._client
+        self._client = None
+        if client is not None:
+            try:
+                await client.aclose()
+            except Exception:
+                logger.debug("Failed to close Redis concurrency client", exc_info=True)
+
+    def _ns(self) -> str:
+        return self.config.namespace.strip().rstrip(":") or "cmzs_deerflow:concurrency"
+
+    def _keys(self, *, run_id: str, model_name: str, user_id: str, thread_id: str) -> list[str]:
+        ns = self._ns()
+        return [
+            f"{ns}:all",
+            f"{ns}:model:{model_name}",
+            f"{ns}:user_model:{model_name}:{user_id}",
+            f"{ns}:thread:{thread_id}",
+            f"{ns}:run:{run_id}",
+        ]
+
+    async def _redis(self) -> Any:
+        if self._client is not None:
+            return self._client
+        try:
+            from redis.asyncio import Redis
+        except Exception as exc:  # pragma: no cover - exercised when dependency is absent
+            raise RuntimeError("redis package is not installed") from exc
+
+        timeout = max(0.001, self.config.socket_timeout_ms / 1000)
+        connect_timeout = max(0.001, self.config.socket_connect_timeout_ms / 1000)
+        self._client = Redis.from_url(
+            self.config.url,
+            password=self.config.password,
+            socket_timeout=timeout,
+            socket_connect_timeout=connect_timeout,
+            decode_responses=True,
+        )
+        return self._client
+
+    async def _with_timeout(self, coro, timeout_ms: int) -> Any:
+        return await asyncio.wait_for(coro, timeout=max(0.001, timeout_ms / 1000))
+
+    def _degraded(self, reason: str) -> AcquireResult:
+        return AcquireResult(AcquireDecision.DEGRADED, reason=reason)
+
+    def _on_failure(self, action: str, exc: BaseException) -> None:
+        self._breaker.record_failure()
+        logger.warning(
+            "Redis concurrency %s failed; fail_open=%s circuit_open=%s: %s",
+            action,
+            self.config.fail_open,
+            self._breaker.open,
+            exc,
+        )
+
+    async def try_acquire(
+        self,
+        *,
+        run_id: str,
+        thread_id: str,
+        model_name: str,
+        user_id: str,
+        assistant_id: str = "",
+        model_limit: int | None = None,
+        user_model_limit: int | None = None,
+        reject_thread: bool = True,
+    ) -> AcquireResult:
+        if not self.enabled:
+            return self._degraded("disabled")
+        if not self._breaker.allow():
+            return self._degraded("circuit_open")
+
+        now = time.time()
+        expires_at = now + self.config.lease_ttl_seconds
+        payload = {
+            "run_id": run_id,
+            "thread_id": thread_id,
+            "model_name": model_name,
+            "user_id": user_id,
+            "assistant_id": assistant_id,
+            "created_at": now,
+            "expires_at": expires_at,
+        }
+        import json
+
+        keys = self._keys(run_id=run_id, model_name=model_name, user_id=user_id, thread_id=thread_id)
+        ttl_ms = int(self.config.lease_ttl_seconds * 1000)
+        try:
+            client = await self._redis()
+            raw = await self._with_timeout(
+                client.eval(
+                    self._ACQUIRE_SCRIPT,
+                    len(keys),
+                    *keys,
+                    str(now),
+                    str(expires_at),
+                    run_id,
+                    str(max(0, int(model_limit or 0))),
+                    str(max(0, int(user_model_limit or 0))),
+                    "1" if reject_thread else "0",
+                    json.dumps(payload, ensure_ascii=False),
+                    str(ttl_ms),
+                ),
+                self.config.acquire_timeout_ms,
+            )
+            self._breaker.record_success()
+        except Exception as exc:
+            self._on_failure("acquire", exc)
+            if self.config.fail_open:
+                return self._degraded(type(exc).__name__)
+            raise
+
+        status = raw[0] if isinstance(raw, list) and raw else ""
+        reason = raw[1] if isinstance(raw, list) and len(raw) > 1 else ""
+        if status == "acquired":
+            return AcquireResult(AcquireDecision.ACQUIRED)
+        return AcquireResult(AcquireDecision.REJECTED, reason=str(reason or "limit"))
+
+    async def heartbeat(
+        self,
+        *,
+        run_id: str,
+        thread_id: str,
+        model_name: str,
+        user_id: str,
+        user_model_accounted: bool = True,
+    ) -> None:
+        if not self.enabled or not self._breaker.allow():
+            return
+        now = time.time()
+        expires_at = now + self.config.lease_ttl_seconds
+        keys = self._keys(run_id=run_id, model_name=model_name, user_id=user_id, thread_id=thread_id)
+        try:
+            client = await self._redis()
+            pipe = client.pipeline(transaction=False)
+            refresh_keys = [keys[0], keys[1], keys[3]]
+            if user_model_accounted:
+                refresh_keys.append(keys[2])
+            for key in refresh_keys:
+                pipe.zadd(key, {run_id: expires_at})
+            pipe.expire(keys[4], self.config.lease_ttl_seconds)
+            await self._with_timeout(pipe.execute(), self.config.operation_timeout_ms)
+            self._breaker.record_success()
+        except Exception as exc:
+            self._on_failure("heartbeat", exc)
+
+    async def release(
+        self,
+        *,
+        run_id: str,
+        thread_id: str,
+        model_name: str,
+        user_id: str,
+        user_model_accounted: bool = True,
+    ) -> None:
+        if not self.enabled or not self._breaker.allow():
+            return
+        keys = self._keys(run_id=run_id, model_name=model_name, user_id=user_id, thread_id=thread_id)
+        try:
+            client = await self._redis()
+            pipe = client.pipeline(transaction=False)
+            release_keys = [keys[0], keys[1], keys[3]]
+            if user_model_accounted:
+                release_keys.append(keys[2])
+            for key in release_keys:
+                pipe.zrem(key, run_id)
+            pipe.delete(keys[4])
+            await self._with_timeout(pipe.execute(), self.config.operation_timeout_ms)
+            self._breaker.record_success()
+        except Exception as exc:
+            self._on_failure("release", exc)
+
+    async def count_active(self) -> int | None:
+        if not self.enabled or not self._breaker.allow():
+            return None
+        key = f"{self._ns()}:all"
+        now = time.time()
+        try:
+            client = await self._redis()
+            pipe = client.pipeline(transaction=False)
+            pipe.zremrangebyscore(key, "-inf", now)
+            pipe.zcard(key)
+            result = await self._with_timeout(pipe.execute(), self.config.operation_timeout_ms)
+            self._breaker.record_success()
+            return int(result[-1] or 0)
+        except Exception as exc:
+            self._on_failure("count", exc)
+            return None
+
+    async def list_active(self, *, limit: int = 1000) -> list[ActiveLease] | None:
+        if not self.enabled or not self._breaker.allow():
+            return None
+        import json
+
+        all_key = f"{self._ns()}:all"
+        now = time.time()
+        try:
+            client = await self._redis()
+            await self._with_timeout(client.zremrangebyscore(all_key, "-inf", now), self.config.operation_timeout_ms)
+            run_ids = await self._with_timeout(client.zrange(all_key, 0, max(0, limit - 1)), self.config.operation_timeout_ms)
+            if not run_ids:
+                self._breaker.record_success()
+                return []
+            keys = [f"{self._ns()}:run:{run_id}" for run_id in run_ids]
+            payloads = await self._with_timeout(client.mget(keys), self.config.operation_timeout_ms)
+            leases: list[ActiveLease] = []
+            for run_id, payload in zip(run_ids, payloads, strict=False):
+                if not payload:
+                    continue
+                try:
+                    data = json.loads(payload)
+                except Exception:
+                    continue
+                leases.append(
+                    ActiveLease(
+                        run_id=str(data.get("run_id") or run_id),
+                        thread_id=str(data.get("thread_id") or ""),
+                        model_name=str(data.get("model_name") or ""),
+                        user_id=str(data.get("user_id") or ""),
+                        assistant_id=str(data.get("assistant_id") or ""),
+                        created_at=float(data.get("created_at") or 0),
+                        expires_at=float(data.get("expires_at") or 0),
+                    )
+                )
+            self._breaker.record_success()
+            return leases
+        except Exception as exc:
+            self._on_failure("list", exc)
+            return None
+
+
+def make_redis_concurrency_gate(config: Any) -> RedisConcurrencyGate | None:
+    concurrency = getattr(config, "concurrency", None)
+    redis_config = getattr(concurrency, "redis", None)
+    if redis_config is None or not getattr(redis_config, "enabled", False) or not getattr(redis_config, "url", None):
+        return None
+    return RedisConcurrencyGate(redis_config)

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

@@ -218,6 +218,8 @@ class RunManager:
         kwargs: dict | None = None,
         multitask_strategy: str = "reject",
         model_concurrency_limit: tuple[str, int] | None = None,
+        user_model_concurrency_limit: tuple[str, str, int] | None = None,
+        run_id: str | None = None,
     ) -> RunRecord:
         """Atomically check for inflight runs and create a new one.
 
@@ -228,7 +230,7 @@ class RunManager:
         This method holds the lock across both the check and the insert,
         eliminating the TOCTOU race in separate ``has_inflight`` + ``create``.
         """
-        run_id = str(uuid.uuid4())
+        run_id = run_id or str(uuid.uuid4())
         now = _now_iso()
 
         _supported_strategies = ("reject", "interrupt", "rollback")
@@ -261,7 +263,20 @@ class RunManager:
                 model_name, limit = model_concurrency_limit
                 active_for_model = sum(1 for r in self._runs.values() if r.status in (RunStatus.pending, RunStatus.running) and str((r.metadata or {}).get("model_name") or "").strip() == model_name)
                 if active_for_model >= max(1, int(limit)):
-                    raise ModelConcurrencyLimitError(model_name, limit)
+                    raise ModelConcurrencyLimitError(model_name, limit, scope="model")
+
+            if user_model_concurrency_limit is not None:
+                model_name, user_id, limit = user_model_concurrency_limit
+                active_for_user_model = sum(
+                    1
+                    for r in self._runs.values()
+                    if r.status in (RunStatus.pending, RunStatus.running)
+                    and str((r.metadata or {}).get("model_name") or "").strip() == model_name
+                    and str((r.metadata or {}).get("user_id") or "").strip() == user_id
+                    and str((r.metadata or {}).get("run_scope") or "external_root").strip() == "external_root"
+                )
+                if active_for_user_model >= max(1, int(limit)):
+                    raise ModelConcurrencyLimitError(model_name, limit, scope="user_model", user_id=user_id)
 
             record = RunRecord(
                 run_id=run_id,
@@ -316,9 +331,11 @@ class ConflictError(Exception):
 class ModelConcurrencyLimitError(Exception):
     """Raised when a model-specific in-flight run limit has been reached."""
 
-    def __init__(self, model_name: str, limit: int) -> None:
+    def __init__(self, model_name: str, limit: int, *, scope: str = "model", user_id: str | None = None) -> None:
         self.model_name = model_name
         self.limit = limit
+        self.scope = scope
+        self.user_id = user_id
         super().__init__(f"Model {model_name} has reached its concurrency limit")
 
 

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

@@ -0,0 +1,39 @@
+"""Small in-memory gate used by legacy model-concurrency tests."""
+
+from __future__ import annotations
+
+import asyncio
+
+
+class ModelConcurrencyGate:
+    """Track per-model active counts with atomic try-acquire semantics."""
+
+    def __init__(self) -> None:
+        self._lock = asyncio.Lock()
+        self._counts: dict[str, int] = {}
+
+    async def try_acquire(self, model_name: str, limit: int) -> bool:
+        key = str(model_name).strip()
+        if not key:
+            return True
+        async with self._lock:
+            current = self._counts.get(key, 0)
+            if current >= max(1, int(limit)):
+                return False
+            self._counts[key] = current + 1
+            return True
+
+    async def release(self, model_name: str) -> None:
+        key = str(model_name).strip()
+        if not key:
+            return
+        async with self._lock:
+            current = self._counts.get(key, 0)
+            if current <= 1:
+                self._counts.pop(key, None)
+            else:
+                self._counts[key] = current - 1
+
+    async def snapshot(self) -> dict[str, int]:
+        async with self._lock:
+            return dict(self._counts)

offline-backend-20260512/backend/packages/harness/pyproject.toml

@@ -23,9 +23,10 @@ dependencies = [
     "markdownify>=1.2.2",
     "markitdown[all,xlsx]>=0.0.1a2",
     "pydantic>=2.12.5",
-    "pyyaml>=6.0.3",
-    "readabilipy>=0.3.0",
-    "tavily-python>=0.7.17",
+    "pyyaml>=6.0.3",
+    "readabilipy>=0.3.0",
+    "redis>=5.0.0",
+    "tavily-python>=0.7.17",
     "firecrawl-py>=1.15.0",
     "tiktoken>=0.8.0",
     "ddgs>=9.10.0",

offline-backend-20260512/backend/tests/test_model_concurrency.py

@@ -10,10 +10,17 @@ import pytest
 
 from app.gateway.routers import system_settings as ss_router
 from app.gateway.services import (
+    RUN_SCOPE_EXTERNAL_ROOT,
+    RUN_SCOPE_ORCHESTRATION_CHILD,
     _get_model_concurrency_limit,
+    _get_user_model_concurrency_limit,
+    _resolve_run_scope,
     _resolve_effective_run_model_name,
+    internal_run_scope_headers,
 )
 from deerflow.config import system_settings as ss
+from deerflow.config.concurrency_config import RedisConcurrencyConfig
+from deerflow.runtime.runs.concurrency_gate import RedisConcurrencyGate
 from deerflow.runtime.runs.manager import ModelConcurrencyLimitError, RunManager
 
 
@@ -52,23 +59,31 @@ async def test_model_concurrency_settings_round_trip(
             enabled=True,
             default_limit=8,
             model_limits={"glm5": 10, "deepseek": 4},
+            user_default_limit=2,
+            user_model_limits={"glm5": 1},
         ),
         req,
     )
     assert saved.enabled is True
     assert saved.default_limit == 8
     assert saved.model_limits == {"glm5": 10, "deepseek": 4}
+    assert saved.user_default_limit == 2
+    assert saved.user_model_limits == {"glm5": 1}
 
     file_path = tmp_runtime_home / "system_settings.json"
     payload = json.loads(file_path.read_text(encoding="utf-8"))
     assert payload["model_concurrency"]["enabled"] is True
     assert payload["model_concurrency"]["default_limit"] == 8
     assert payload["model_concurrency"]["model_limits"]["glm5"] == 10
+    assert payload["model_concurrency"]["user_default_limit"] == 2
+    assert payload["model_concurrency"]["user_model_limits"]["glm5"] == 1
 
     loaded = await ss_router.get_model_concurrency(req)
     assert loaded.enabled is True
     assert loaded.default_limit == 8
     assert loaded.model_limits["deepseek"] == 4
+    assert loaded.user_default_limit == 2
+    assert loaded.user_model_limits["glm5"] == 1
 
 
 def test_resolve_effective_run_model_name_prefers_requested_model(
@@ -183,6 +198,53 @@ async def test_run_manager_rejects_model_limit_atomically(
         )
 
 
+@pytest.mark.asyncio
+async def test_run_manager_rejects_user_model_limit_atomically(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setattr(
+        "app.gateway.services.load_system_settings",
+        lambda: SimpleNamespace(
+            model_concurrency=SimpleNamespace(
+                enabled=True,
+                default_limit=10,
+                model_limits={},
+                user_default_limit=3,
+                user_model_limits={"glm5": 1},
+            )
+        ),
+    )
+    assert _get_user_model_concurrency_limit("glm5") == 1
+    assert _get_user_model_concurrency_limit("other") == 3
+
+    run_mgr = RunManager()
+    await run_mgr.create_or_reject(
+        "t-child",
+        metadata={
+            "model_name": "glm5",
+            "user_id": "u1",
+            "run_scope": RUN_SCOPE_ORCHESTRATION_CHILD,
+        },
+        user_model_concurrency_limit=("glm5", "u1", 1),
+    )
+    await run_mgr.create_or_reject("t1", metadata={"model_name": "glm5", "user_id": "u1"})
+    await run_mgr.create_or_reject(
+        "t2",
+        metadata={"model_name": "glm5", "user_id": "u2"},
+        user_model_concurrency_limit=("glm5", "u2", 1),
+    )
+
+    with pytest.raises(ModelConcurrencyLimitError) as exc_info:
+        await run_mgr.create_or_reject(
+            "t3",
+            metadata={"model_name": "glm5", "user_id": "u1"},
+            user_model_concurrency_limit=("glm5", "u1", 1),
+        )
+
+    assert exc_info.value.scope == "user_model"
+    assert exc_info.value.user_id == "u1"
+
+
 def test_get_model_concurrency_limit_skips_when_disabled(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
@@ -193,7 +255,51 @@ def test_get_model_concurrency_limit_skips_when_disabled(
                 enabled=False,
                 default_limit=1,
                 model_limits={"glm5": 1},
+                user_default_limit=1,
+                user_model_limits={"glm5": 1},
             )
         ),
     )
     assert _get_model_concurrency_limit("glm5") is None
+    assert _get_user_model_concurrency_limit("glm5") is None
+
+
+def test_internal_run_scope_is_only_trusted_for_loopback_or_server_argument() -> None:
+    body = SimpleNamespace(metadata={"run_scope": RUN_SCOPE_ORCHESTRATION_CHILD})
+
+    trusted_loopback_request = SimpleNamespace(
+        client=SimpleNamespace(host="127.0.0.1"),
+        headers=internal_run_scope_headers(),
+    )
+    untrusted_loopback_request = SimpleNamespace(
+        client=SimpleNamespace(host="127.0.0.1"),
+        headers={},
+    )
+    public_request = SimpleNamespace(
+        client=SimpleNamespace(host="10.0.0.8"),
+        headers=internal_run_scope_headers(),
+    )
+
+    assert (
+        _resolve_run_scope(body, trusted_loopback_request, None)
+        == RUN_SCOPE_ORCHESTRATION_CHILD
+    )
+    assert (
+        _resolve_run_scope(body, untrusted_loopback_request, None)
+        == RUN_SCOPE_EXTERNAL_ROOT
+    )
+    assert _resolve_run_scope(body, public_request, None) == RUN_SCOPE_EXTERNAL_ROOT
+    assert (
+        _resolve_run_scope(
+            SimpleNamespace(metadata={}),
+            public_request,
+            RUN_SCOPE_ORCHESTRATION_CHILD,
+        )
+        == RUN_SCOPE_ORCHESTRATION_CHILD
+    )
+
+
+def test_redis_concurrency_namespace_has_system_prefix() -> None:
+    assert RedisConcurrencyConfig().namespace == "cmzs_deerflow:concurrency"
+    gate = RedisConcurrencyGate(RedisConcurrencyConfig(namespace=""))
+    assert gate._ns() == "cmzs_deerflow:concurrency"