feat: 新增岗位管理、提示词模板与资源标签,完善用户提示词与推荐问题接口

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.8 committed 2026-06-16 10:49:36 +08:00
1 parent 2f414c6e08
commit 3855c30edd
52 files changed
+3143 -57

No files matched your search

@@ -0,0 +1,479 @@
"use client";
import {
CheckIcon,
Loader2,
PencilIcon,
PlusIcon,
RefreshCwIcon,
Trash2Icon,
UserPlusIcon,
UsersIcon,
} from "lucide-react";
import { useEffect, useMemo, useState } from "react";
import { toast } from "sonner";
import { Button } from "@/components/ui/button";
import {
Dialog,
DialogContent,
DialogFooter,
DialogHeader,
DialogTitle,
} from "@/components/ui/dialog";
import { Input } from "@/components/ui/input";
import { Textarea } from "@/components/ui/textarea";
import { resolveTagColor } from "@/components/workspace/tags/tag-colors";
import {
POSITION_RESOURCE_HINT,
POSITION_RESOURCE_LABELS,
POSITION_RESOURCE_TYPES,
resourceTypeToTagType,
useAddMembers,
useCreatePosition,
useDeletePosition,
usePositionMembers,
usePositionTagBindings,
usePositionUsers,
usePositions,
useRemoveMembers,
useResyncPosition,
useSetTagBindings,
useUpdatePosition,
type Position,
type PositionResourceType,
type TagBindings,
} from "@/core/positions";
import { useTags } from "@/core/tags";
import { cn } from "@/lib/utils";
function errMsg(e: unknown): string {
return String(e instanceof Error ? e.message : e);
}
/** Inline multi-select chip row for the tags of one resource type. */
function TagMultiSelect({
resourceType,
value,
onChange,
}: {
resourceType: PositionResourceType;
value: string[];
onChange: (next: string[]) => void;
}) {
const { data: tags, isLoading } = useTags(resourceTypeToTagType(resourceType));
const toggle = (id: string) =>
onChange(value.includes(id) ? value.filter((x) => x !== id) : [...value, id]);
if (isLoading) {
return <Loader2 className="text-muted-foreground size-4 animate-spin" />;
}
if (!tags || tags.length === 0) {
return (
<span className="text-muted-foreground text-xs">
暂无{POSITION_RESOURCE_LABELS[resourceType]}标签,请先到「标签管理」创建。
</span>
);
}
return (
<div className="flex flex-wrap items-center gap-1.5">
{tags.map((tag) => {
const checked = value.includes(tag.id);
const color = resolveTagColor(tag.color);
return (
<button
key={tag.id}
type="button"
onClick={() => toggle(tag.id)}
className={cn(
"inline-flex items-center gap-1 rounded-full border px-3 py-1 text-xs font-medium transition-all",
color.chip,
checked ? "ring-foreground/40 ring-2 ring-offset-1" : "opacity-45 hover:opacity-90",
)}
>
{checked && <CheckIcon className="size-3" />}
{tag.name}
</button>
);
})}
</div>
);
}
function PositionFormDialog({
open,
onOpenChange,
editing,
}: {
open: boolean;
onOpenChange: (open: boolean) => void;
editing: Position | null;
}) {
const createPosition = useCreatePosition();
const updatePosition = useUpdatePosition();
const [name, setName] = useState("");
const [description, setDescription] = useState("");
const [isDefault, setIsDefault] = useState(false);
const [syncedFor, setSyncedFor] = useState<string | null>(null);
const syncKey = `${open}:${editing?.id ?? "new"}`;
if (open && syncedFor !== syncKey) {
setName(editing?.name ?? "");
setDescription(editing?.description ?? "");
setIsDefault(editing?.is_default ?? false);
setSyncedFor(syncKey);
}
if (!open && syncedFor !== null) setSyncedFor(null);
const pending = createPosition.isPending || updatePosition.isPending;
const handleSave = async () => {
const trimmed = name.trim();
if (!trimmed) return;
try {
if (editing) {
await updatePosition.mutateAsync({ id: editing.id, input: { name: trimmed, description, is_default: isDefault } });
} else {
await createPosition.mutateAsync({ name: trimmed, description, is_default: isDefault });
}
onOpenChange(false);
} catch (e) {
toast.error(errMsg(e));
}
};
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent>
<DialogHeader>
<DialogTitle>{editing ? "编辑岗位" : "新增岗位"}</DialogTitle>
</DialogHeader>
<div className="flex flex-col gap-4">
<div className="flex flex-col gap-1.5">
<label className="text-sm font-medium">岗位名称</label>
<Input value={name} onChange={(e) => setName(e.target.value)} placeholder="如:理赔专员" maxLength={191} />
</div>
<div className="flex flex-col gap-1.5">
<label className="text-sm font-medium">描述</label>
<Textarea value={description} onChange={(e) => setDescription(e.target.value)} rows={2} maxLength={1024} />
</div>
<label className="flex cursor-pointer items-center gap-2 text-sm">
<input type="checkbox" checked={isDefault} onChange={(e) => setIsDefault(e.target.checked)} />
设为默认岗位(未分配岗位的用户继承此岗位的配置)
</label>
</div>
<DialogFooter>
<Button variant="outline" onClick={() => onOpenChange(false)}>取消</Button>
<Button onClick={handleSave} disabled={pending || !name.trim()}>
{pending && <Loader2 className="size-4 animate-spin" />}保存
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
);
}
function AddMembersDialog({
open,
onOpenChange,
positionId,
}: {
open: boolean;
onOpenChange: (open: boolean) => void;
positionId: string;
}) {
const { data: users, isLoading } = usePositionUsers();
const addMembers = useAddMembers();
const [selected, setSelected] = useState<Set<string>>(new Set());
const [search, setSearch] = useState("");
useEffect(() => {
if (open) setSelected(new Set());
}, [open]);
const filtered = useMemo(() => {
const list = users ?? [];
const q = search.trim().toLowerCase();
return q ? list.filter((u) => u.email.toLowerCase().includes(q)) : list;
}, [users, search]);
const toggle = (id: string) => {
setSelected((prev) => {
const next = new Set(prev);
if (next.has(id)) next.delete(id);
else next.add(id);
return next;
});
};
const handleSave = async () => {
if (selected.size === 0) return;
try {
await addMembers.mutateAsync({ positionId, userIds: [...selected] });
toast.success(`已为 ${selected.size} 名用户配置岗位`);
onOpenChange(false);
} catch (e) {
toast.error(errMsg(e));
}
};
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="max-w-lg">
<DialogHeader>
<DialogTitle>批量添加成员</DialogTitle>
</DialogHeader>
<Input value={search} onChange={(e) => setSearch(e.target.value)} placeholder="搜索邮箱…" />
<div className="max-h-72 overflow-y-auto rounded-md border">
{isLoading && (
<div className="text-muted-foreground flex items-center justify-center gap-2 py-6">
<Loader2 className="size-4 animate-spin" />加载用户…
</div>
)}
{!isLoading && filtered.length === 0 && (
<div className="text-muted-foreground py-6 text-center text-sm">无匹配用户</div>
)}
{filtered.map((u) => {
const checked = selected.has(u.id);
return (
<button
key={u.id}
type="button"
onClick={() => toggle(u.id)}
className={cn(
"flex w-full items-center justify-between gap-2 border-b px-3 py-2 text-left text-sm last:border-b-0 hover:bg-muted/50",
checked && "bg-muted/40",
)}
>
<span className="truncate">
{u.email}
{u.system_role === "admin" && <span className="text-muted-foreground ml-2 text-xs">(管理员)</span>}
</span>
<span className="flex items-center gap-2">
{u.position_id && u.position_id !== positionId && (
<span className="text-muted-foreground text-xs">已属其他岗位</span>
)}
<span className={cn("flex size-4 items-center justify-center rounded border", checked ? "bg-foreground text-background border-foreground" : "border-muted-foreground/40")}>
{checked && <CheckIcon className="size-3" />}
</span>
</span>
</button>
);
})}
</div>
<DialogFooter>
<Button variant="outline" onClick={() => onOpenChange(false)}>取消</Button>
<Button onClick={handleSave} disabled={addMembers.isPending || selected.size === 0}>
{addMembers.isPending && <Loader2 className="size-4 animate-spin" />}添加 {selected.size > 0 ? `(${selected.size})` : ""}
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
);
}
function PositionDetail({ position }: { position: Position }) {
const { data: serverBindings } = usePositionTagBindings(position.id);
const { data: members } = usePositionMembers(position.id);
const setBindings = useSetTagBindings();
const removeMembers = useRemoveMembers();
const resync = useResyncPosition();
const [draft, setDraft] = useState<TagBindings>({});
const [addOpen, setAddOpen] = useState(false);
// Reload the draft whenever the selected position or its server bindings change.
const [loadedFor, setLoadedFor] = useState<string | null>(null);
const loadKey = `${position.id}:${serverBindings ? "ready" : "loading"}`;
if (serverBindings && loadedFor !== loadKey) {
setDraft({ ...serverBindings });
setLoadedFor(loadKey);
}
const setType = (rt: PositionResourceType, ids: string[]) =>
setDraft((prev) => ({ ...prev, [rt]: ids }));
const handleSaveBindings = async () => {
try {
await setBindings.mutateAsync({ positionId: position.id, bindings: draft });
toast.success("标签配置已保存,已同步岗位成员");
} catch (e) {
toast.error(errMsg(e));
}
};
const handleResync = async () => {
try {
const n = await resync.mutateAsync(position.id);
toast.success(`已重新同步 ${n} 名成员`);
} catch (e) {
toast.error(errMsg(e));
}
};
const handleRemove = async (userId: string) => {
try {
await removeMembers.mutateAsync({ positionId: position.id, userIds: [userId] });
} catch (e) {
toast.error(errMsg(e));
}
};
return (
<div className="flex flex-col gap-6">
<div className="rounded-lg border p-4">
<div className="mb-3 flex items-center justify-between">
<h3 className="text-sm font-semibold">默认资源标签配置</h3>
<Button size="sm" onClick={handleSaveBindings} disabled={setBindings.isPending}>
{setBindings.isPending && <Loader2 className="size-4 animate-spin" />}保存标签配置
</Button>
</div>
<div className="flex flex-col divide-y">
{POSITION_RESOURCE_TYPES.map((rt) => (
<div key={rt} className="flex flex-col gap-2 py-3 first:pt-0 last:pb-0">
<div className="flex items-baseline gap-2">
<span className="text-sm font-medium">{POSITION_RESOURCE_LABELS[rt]}</span>
<span className="text-muted-foreground text-xs">{POSITION_RESOURCE_HINT[rt]}</span>
</div>
<TagMultiSelect resourceType={rt} value={draft[rt] ?? []} onChange={(ids) => setType(rt, ids)} />
</div>
))}
</div>
</div>
<div className="rounded-lg border p-4">
<div className="mb-3 flex items-center justify-between">
<h3 className="flex items-center gap-1.5 text-sm font-semibold">
<UsersIcon className="size-4" />岗位成员 ({members?.length ?? 0})
</h3>
<div className="flex gap-2">
<Button size="sm" variant="outline" onClick={handleResync} disabled={resync.isPending}>
{resync.isPending ? <Loader2 className="size-4 animate-spin" /> : <RefreshCwIcon className="size-4" />}重新同步
</Button>
<Button size="sm" onClick={() => setAddOpen(true)}>
<UserPlusIcon className="size-4" />添加成员
</Button>
</div>
</div>
{members && members.length > 0 ? (
<div className="flex flex-col divide-y">
{members.map((m) => (
<div key={m.id} className="flex items-center justify-between py-2 text-sm">
<span className="truncate">{m.email}</span>
<Button variant="ghost" size="icon" onClick={() => handleRemove(m.id)} aria-label="移除">
<Trash2Icon className="text-destructive size-4" />
</Button>
</div>
))}
</div>
) : (
<div className="text-muted-foreground py-6 text-center text-sm">暂无成员</div>
)}
</div>
<AddMembersDialog open={addOpen} onOpenChange={setAddOpen} positionId={position.id} />
</div>
);
}
export function PositionManagementPage() {
const { data: positions, isLoading, error } = usePositions();
const deletePosition = useDeletePosition();
const [selectedId, setSelectedId] = useState<string | null>(null);
const [dialogOpen, setDialogOpen] = useState(false);
const [editing, setEditing] = useState<Position | null>(null);
const selected = useMemo(
() => positions?.find((p) => p.id === selectedId) ?? null,
[positions, selectedId],
);
// Auto-select the first position once loaded.
useEffect(() => {
if (!selectedId && positions && positions.length > 0) setSelectedId(positions[0].id);
}, [positions, selectedId]);
const handleDelete = async (p: Position) => {
if (!window.confirm(`确认删除岗位「${p.name}」?其成员将回退到默认岗位。`)) return;
try {
await deletePosition.mutateAsync(p.id);
if (selectedId === p.id) setSelectedId(null);
} catch (e) {
toast.error(errMsg(e));
}
};
return (
<div className="flex flex-col gap-4 p-6">
<div className="flex items-center justify-between">
<div>
<h1 className="text-lg font-semibold">岗位管理</h1>
<p className="text-muted-foreground text-sm">
为岗位配置默认智能体/技能/定时任务/常用提示词/推荐问题的标签,并批量分配给用户。
</p>
</div>
<Button size="sm" onClick={() => { setEditing(null); setDialogOpen(true); }}>
<PlusIcon className="size-4" />新增岗位
</Button>
</div>
{isLoading && (
<div className="text-muted-foreground flex items-center justify-center gap-2 py-12">
<Loader2 className="size-4 animate-spin" />加载中…
</div>
)}
{error && <div className="text-destructive text-sm">{errMsg(error)}</div>}
{!isLoading && !error && (
<div className="grid grid-cols-1 gap-6 lg:grid-cols-[280px_1fr]">
<div className="flex flex-col gap-2">
{positions && positions.length === 0 && (
<div className="text-muted-foreground rounded-lg border border-dashed py-8 text-center text-sm">
还没有岗位,点击「新增岗位」创建。
</div>
)}
{positions?.map((p) => (
<div
key={p.id}
onClick={() => setSelectedId(p.id)}
className={cn(
"group cursor-pointer rounded-lg border p-3 transition-colors",
selectedId === p.id ? "border-foreground bg-muted/40" : "hover:bg-muted/30",
)}
>
<div className="flex items-center justify-between">
<div className="flex items-center gap-2">
<span className="text-sm font-medium">{p.name}</span>
{p.is_default && (
<span className="rounded bg-blue-100 px-1.5 py-0.5 text-[10px] text-blue-700 dark:bg-blue-950 dark:text-blue-300">默认</span>
)}
</div>
<div className="flex opacity-0 transition-opacity group-hover:opacity-100">
<Button variant="ghost" size="icon" onClick={(e) => { e.stopPropagation(); setEditing(p); setDialogOpen(true); }} aria-label="编辑">
<PencilIcon className="size-3.5" />
</Button>
<Button variant="ghost" size="icon" onClick={(e) => { e.stopPropagation(); void handleDelete(p); }} aria-label="删除">
<Trash2Icon className="text-destructive size-3.5" />
</Button>
</div>
</div>
<div className="text-muted-foreground mt-1 flex items-center gap-2 text-xs">
<UsersIcon className="size-3" />{p.member_count} 名成员
</div>
</div>
))}
</div>
<div>
{selected ? (
<PositionDetail key={selected.id} position={selected} />
) : (
<div className="text-muted-foreground rounded-lg border border-dashed py-16 text-center text-sm">
选择左侧一个岗位以配置标签与成员
</div>
)}
</div>
</div>
)}
<PositionFormDialog open={dialogOpen} onOpenChange={setDialogOpen} editing={editing} />
</div>
);
}
@@ -0,0 +1,258 @@
"use client";
import { CheckIcon, Loader2, PencilIcon, PlusIcon, TagIcon, Trash2Icon } from "lucide-react";
import { useState } from "react";
import { toast } from "sonner";
import { Button } from "@/components/ui/button";
import {
Dialog,
DialogContent,
DialogFooter,
DialogHeader,
DialogTitle,
} from "@/components/ui/dialog";
import { Input } from "@/components/ui/input";
import { Textarea } from "@/components/ui/textarea";
import { resolveTagColor } from "@/components/workspace/tags/tag-colors";
import {
usePromptTemplates,
useCreatePromptTemplate,
useDeletePromptTemplate,
useUpdatePromptTemplate,
type PromptTemplate,
} from "@/core/custom-prompts";
import { useResourceTags, useSetResourceTags, useTags } from "@/core/tags";
import { cn } from "@/lib/utils";
function errMsg(e: unknown): string {
return String(e instanceof Error ? e.message : e);
}
function TemplateFormDialog({
open,
onOpenChange,
editing,
}: {
open: boolean;
onOpenChange: (open: boolean) => void;
editing: PromptTemplate | null;
}) {
const createT = useCreatePromptTemplate();
const updateT = useUpdatePromptTemplate();
const [title, setTitle] = useState("");
const [content, setContent] = useState("");
const [syncedFor, setSyncedFor] = useState<string | null>(null);
const syncKey = `${open}:${editing?.id ?? "new"}`;
if (open && syncedFor !== syncKey) {
setTitle(editing?.title ?? "");
setContent(editing?.content ?? "");
setSyncedFor(syncKey);
}
if (!open && syncedFor !== null) setSyncedFor(null);
const pending = createT.isPending || updateT.isPending;
const handleSave = async () => {
if (!content.trim()) return;
try {
if (editing) {
await updateT.mutateAsync({ id: editing.id, input: { title: title.trim(), content } });
} else {
await createT.mutateAsync({ title: title.trim(), content });
}
onOpenChange(false);
} catch (e) {
toast.error(errMsg(e));
}
};
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent>
<DialogHeader>
<DialogTitle>{editing ? "编辑提示词模板" : "新增提示词模板"}</DialogTitle>
</DialogHeader>
<div className="flex flex-col gap-4">
<div className="flex flex-col gap-1.5">
<label className="text-sm font-medium">标题</label>
<Input value={title} onChange={(e) => setTitle(e.target.value)} placeholder="如:理赔标准话术" maxLength={255} />
</div>
<div className="flex flex-col gap-1.5">
<label className="text-sm font-medium">内容</label>
<Textarea value={content} onChange={(e) => setContent(e.target.value)} rows={5} placeholder="提示词正文…" />
</div>
</div>
<DialogFooter>
<Button variant="outline" onClick={() => onOpenChange(false)}>取消</Button>
<Button onClick={handleSave} disabled={pending || !content.trim()}>
{pending && <Loader2 className="size-4 animate-spin" />}保存
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
);
}
function TemplateTagsDialog({
open,
onOpenChange,
template,
}: {
open: boolean;
onOpenChange: (open: boolean) => void;
template: PromptTemplate;
}) {
const { data: tags } = useTags("user_prompt");
const { data: current } = useResourceTags("user_prompt", template.id, open);
const setTags = useSetResourceTags();
const [selected, setSelected] = useState<string[]>([]);
const [syncedFor, setSyncedFor] = useState<string | null>(null);
const syncKey = `${open}:${current ? "ready" : "loading"}`;
if (open && current && syncedFor !== syncKey) {
setSelected(current.map((t) => t.id));
setSyncedFor(syncKey);
}
if (!open && syncedFor !== null) setSyncedFor(null);
const toggle = (id: string) =>
setSelected((prev) => (prev.includes(id) ? prev.filter((x) => x !== id) : [...prev, id]));
const handleSave = async () => {
try {
await setTags.mutateAsync({ targetType: "user_prompt", targetId: template.id, tagIds: selected });
onOpenChange(false);
} catch (e) {
toast.error(errMsg(e));
}
};
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent>
<DialogHeader>
<DialogTitle>为模板打标签</DialogTitle>
</DialogHeader>
{!tags || tags.length === 0 ? (
<p className="text-muted-foreground text-sm">
暂无「常用提示词」标签,请先到「标签管理」创建。
</p>
) : (
<div className="flex flex-wrap gap-1.5">
{tags.map((tag) => {
const checked = selected.includes(tag.id);
const color = resolveTagColor(tag.color);
return (
<button
key={tag.id}
type="button"
onClick={() => toggle(tag.id)}
className={cn(
"inline-flex items-center gap-1 rounded-full border px-3 py-1 text-xs font-medium transition-all",
color.chip,
checked ? "ring-foreground/40 ring-2 ring-offset-1" : "opacity-45 hover:opacity-90",
)}
>
{checked && <CheckIcon className="size-3" />}
{tag.name}
</button>
);
})}
</div>
)}
<DialogFooter>
<Button variant="outline" onClick={() => onOpenChange(false)}>取消</Button>
<Button onClick={handleSave} disabled={setTags.isPending}>
{setTags.isPending && <Loader2 className="size-4 animate-spin" />}保存
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
);
}
export function PromptTemplatesPage() {
const { data: templates, isLoading, error } = usePromptTemplates();
const deleteT = useDeletePromptTemplate();
const [formOpen, setFormOpen] = useState(false);
const [tagsOpen, setTagsOpen] = useState(false);
const [editing, setEditing] = useState<PromptTemplate | null>(null);
const [tagging, setTagging] = useState<PromptTemplate | null>(null);
const handleDelete = async (tpl: PromptTemplate) => {
if (!window.confirm(`确认删除模板「${tpl.title || tpl.content.slice(0, 20)}」?已分发给用户的克隆将在下次同步时移除。`)) return;
try {
await deleteT.mutateAsync(tpl.id);
} catch (e) {
toast.error(errMsg(e));
}
};
return (
<div className="flex flex-col gap-4 p-6">
<div className="flex items-center justify-between">
<div>
<h1 className="text-lg font-semibold">常用提示词模板</h1>
<p className="text-muted-foreground text-sm">
管理员维护提示词模板并打标签;岗位配置了对应标签的用户,会自动加载这些模板到「常用提示词」。
</p>
</div>
<Button size="sm" onClick={() => { setEditing(null); setFormOpen(true); }}>
<PlusIcon className="size-4" />新增模板
</Button>
</div>
{isLoading && (
<div className="text-muted-foreground flex items-center justify-center gap-2 py-12">
<Loader2 className="size-4 animate-spin" />加载中…
</div>
)}
{error && <div className="text-destructive text-sm">{errMsg(error)}</div>}
{!isLoading && !error && templates && templates.length === 0 && (
<div className="text-muted-foreground rounded-lg border border-dashed py-12 text-center text-sm">
还没有模板,点击「新增模板」创建。
</div>
)}
{!isLoading && !error && templates && templates.length > 0 && (
<div className="overflow-hidden rounded-md border">
<table className="w-full text-sm">
<thead className="bg-muted/50">
<tr>
<th className="text-muted-foreground px-4 py-3 text-left font-medium">标题</th>
<th className="text-muted-foreground px-4 py-3 text-left font-medium">内容</th>
<th className="text-muted-foreground px-4 py-3 text-right font-medium">操作</th>
</tr>
</thead>
<tbody>
{templates.map((tpl, i) => (
<tr key={tpl.id} className={i % 2 === 0 ? "bg-background" : "bg-muted/20"}>
<td className="px-4 py-3 font-medium">{tpl.title || "—"}</td>
<td className="text-muted-foreground max-w-md truncate px-4 py-3">{tpl.content}</td>
<td className="px-4 py-3 text-right">
<div className="flex justify-end gap-1">
<Button variant="ghost" size="icon" onClick={() => { setTagging(tpl); setTagsOpen(true); }} aria-label="打标签">
<TagIcon className="size-4" />
</Button>
<Button variant="ghost" size="icon" onClick={() => { setEditing(tpl); setFormOpen(true); }} aria-label="编辑">
<PencilIcon className="size-4" />
</Button>
<Button variant="ghost" size="icon" onClick={() => handleDelete(tpl)} aria-label="删除">
<Trash2Icon className="text-destructive size-4" />
</Button>
</div>
</td>
</tr>
))}
</tbody>
</table>
</div>
)}
<TemplateFormDialog open={formOpen} onOpenChange={setFormOpen} editing={editing} />
{tagging && <TemplateTagsDialog open={tagsOpen} onOpenChange={setTagsOpen} template={tagging} />}
</div>
);
}
@@ -27,6 +27,7 @@ import {
type Tag,
type TagType,
} from "@/core/tags";
import { TAG_TYPES, TAG_TYPE_LABELS } from "@/core/tags/types";
import { cn } from "@/lib/utils";
function TagFormDialog({
@@ -191,11 +192,11 @@ export function TagManagementPage() {
onValueChange={(v) => setActiveType(v as TagType)}
>
<TabsList variant="line">
<TabsTrigger value="agent">{t.adminTags.typeAgent}</TabsTrigger>
<TabsTrigger value="skill">{t.adminTags.typeSkill}</TabsTrigger>
<TabsTrigger value="scheduled_task">
{t.adminTags.typeScheduledTask}
</TabsTrigger>
{TAG_TYPES.map((tt) => (
<TabsTrigger key={tt} value={tt}>
{TAG_TYPE_LABELS[tt]}
</TabsTrigger>
))}
</TabsList>
</Tabs>
<Button size="sm" onClick={openCreate}>
@@ -521,25 +521,6 @@ export function InputBox({
</div>
)}
{isNewThread && searchParams.get("mode") !== "skill" && showAdminRecommendedQuestions && (
<div className="flex flex-wrap justify-center gap-2.5 px-4 mb-[10px]">
{recommendedQuestions.map((q) => (
<button
key={q.id}
type="button"
className="rounded-full border border-[#e4e4e7] bg-white px-4 py-2 text-sm text-[#18181b] shadow-none transition-colors hover:bg-[#f4f4f5] cursor-pointer dark:border-slate-600 dark:bg-slate-700/50 dark:text-slate-300 dark:hover:bg-slate-700"
onClick={() => {
textInput.setInput(q.content);
setFollowupsHidden(true);
setTimeout(() => requestFormSubmit(), 0);
}}
>
{desensitize(q.title)}
</button>
))}
</div>
)}
{showFollowups && (
<div className="flex items-center justify-center pb-2">
<div className="flex items-center gap-2">
@@ -1060,6 +1041,35 @@ export function InputBox({
</ModelSelectorList>
</ModelSelectorContent>
</ModelSelector>
{isNewThread && searchParams.get("mode") !== "skill" && showAdminRecommendedQuestions && (
<DropdownMenu>
<DropdownMenuTrigger asChild>
<button
type="button"
title="推荐问题"
className="inline-flex items-center gap-1.5 rounded-full border border-[#e4e4e7] bg-background px-3 py-1.5 text-xs text-[#18181b] shadow-sm transition-colors hover:bg-[#f4f4f5] cursor-pointer dark:border-slate-600 dark:bg-slate-700/60 dark:text-slate-200 dark:hover:bg-slate-700"
>
<LightbulbIcon className="size-4 text-amber-500" />
推荐问题
</button>
</DropdownMenuTrigger>
<DropdownMenuContent side="top" align="end" className="max-h-80 w-72 overflow-y-auto">
{recommendedQuestions.map((q) => (
<DropdownMenuItem
key={q.id}
className="cursor-pointer whitespace-normal"
onSelect={() => {
textInput.setInput(q.content);
setFollowupsHidden(true);
setTimeout(() => requestFormSubmit(), 0);
}}
>
{desensitize(q.title)}
</DropdownMenuItem>
))}
</DropdownMenuContent>
</DropdownMenu>
)}
<PromptInputSubmit
className="rounded-full border-border bg-background shadow-sm hover:bg-muted dark:bg-blue-500 dark:border-blue-500 dark:text-white dark:hover:bg-blue-400 dark:shadow-none"
disabled={disabled}
@@ -176,13 +176,20 @@ function SkillSettingsList({ onClose }: { onClose?: () => void }) {
const fillNamesMutation = useGenerateMissingSkillNameZh();
// Optional model override for the admin "一键补充中文名" action (null = default).
const [fillModel, setFillModel] = useState<string | null>(null);
// Name + tag filtering is resolved server-side via the /api/skills query.
const { skills, isLoading, error } = useSkills({ search, tagIds, sort });
const storedAuth = useMemo(() => getStoredAuth(), []);
const currentUserId = storedAuth?.user_id ?? null;
const isAdmin = storedAuth?.system_role === "admin";
// Name + tag filtering is resolved server-side via the /api/skills query.
// Non-admins get their 岗位 (position) visibility applied (scope=mine): a
// position that binds skill tags restricts which skills they can see.
const { skills, isLoading, error } = useSkills({
search,
tagIds,
sort,
scope: isAdmin ? undefined : "mine",
});
// 把可见技能分类成 UI 关心的几个桶。
// "All" = 内置 + 已发布的技能(仅 active 状态);自己的技能也要自己发布了才出现。
// "Mine" = 严格属于我自己的自定义技能(仅 active 状态)。
@@ -0,0 +1,98 @@
"use client";
import { CheckIcon, Loader2 } from "lucide-react";
import { useState } from "react";
import { toast } from "sonner";
import { Button } from "@/components/ui/button";
import {
Dialog,
DialogContent,
DialogFooter,
DialogHeader,
DialogTitle,
} from "@/components/ui/dialog";
import { resolveTagColor } from "@/components/workspace/tags/tag-colors";
import { useResourceTags, useSetResourceTags, useTags, type TagType } from "@/core/tags";
import { cn } from "@/lib/utils";
/** Generic admin dialog to assign a resource's tags (multi-select chips). */
export function ResourceTagDialog({
open,
onOpenChange,
targetType,
targetId,
title = "打标签",
}: {
open: boolean;
onOpenChange: (open: boolean) => void;
targetType: TagType;
targetId: string;
title?: string;
}) {
const { data: tags } = useTags(targetType);
const { data: current } = useResourceTags(targetType, targetId, open);
const setTags = useSetResourceTags();
const [selected, setSelected] = useState<string[]>([]);
const [syncedFor, setSyncedFor] = useState<string | null>(null);
const syncKey = `${open}:${targetId}:${current ? "ready" : "loading"}`;
if (open && current && syncedFor !== syncKey) {
setSelected(current.map((t) => t.id));
setSyncedFor(syncKey);
}
if (!open && syncedFor !== null) setSyncedFor(null);
const toggle = (id: string) =>
setSelected((prev) => (prev.includes(id) ? prev.filter((x) => x !== id) : [...prev, id]));
const handleSave = async () => {
try {
await setTags.mutateAsync({ targetType, targetId, tagIds: selected });
onOpenChange(false);
} catch (e) {
toast.error(String(e instanceof Error ? e.message : e));
}
};
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent>
<DialogHeader>
<DialogTitle>{title}</DialogTitle>
</DialogHeader>
{!tags || tags.length === 0 ? (
<p className="text-muted-foreground text-sm">暂无可用标签,请先到「标签管理」创建对应类型的标签。</p>
) : (
<div className="flex flex-wrap gap-1.5">
{tags.map((tag) => {
const checked = selected.includes(tag.id);
const color = resolveTagColor(tag.color);
return (
<button
key={tag.id}
type="button"
onClick={() => toggle(tag.id)}
className={cn(
"inline-flex items-center gap-1 rounded-full border px-3 py-1 text-xs font-medium transition-all",
color.chip,
checked ? "ring-foreground/40 ring-2 ring-offset-1" : "opacity-45 hover:opacity-90",
)}
>
{checked && <CheckIcon className="size-3" />}
{tag.name}
</button>
);
})}
</div>
)}
<DialogFooter>
<Button variant="outline" onClick={() => onOpenChange(false)}>取消</Button>
<Button onClick={handleSave} disabled={setTags.isPending}>
{setTags.isPending && <Loader2 className="size-4 animate-spin" />}保存
</Button>
</DialogFooter>
</DialogContent>
</Dialog>
);
}
@@ -15,11 +15,14 @@ export interface CustomPrompt {
* selected.
*/
agent_id?: string | null;
/** ``"user"`` (the user's own) or ``"position"`` (granted by their 岗位). */
origin?: string;
created_at?: string | null;
updated_at?: string | null;
}
const QUERY_KEY = ["user-prompts"] as const;
const TEMPLATES_KEY = ["user-prompts", "templates"] as const;
const EMPTY_PROMPTS: CustomPrompt[] = [];
const baseUrl = () => `${getBackendBaseURL()}/api/user-prompts`;
@@ -117,3 +120,91 @@ export function useDeleteCustomPrompt() {
},
});
}
// ----------------------------------------------------- admin prompt templates
//
// Templates are admin-maintained prompts that carry ``user_prompt`` tags. The
// 岗位 (position) sync engine clones the templates a user's position grants
// into that user's own prompts. Tagging itself reuses the generic tags API.
export interface PromptTemplate {
id: string;
title: string;
content: string;
sort_order?: number;
agent_id?: string | null;
created_at?: string | null;
updated_at?: string | null;
}
const templatesUrl = () => `${getBackendBaseURL()}/api/user-prompts/templates`;
async function listTemplates(): Promise<PromptTemplate[]> {
const res = await apiFetch(templatesUrl());
if (!res.ok) throw new Error(`Failed to load templates: ${res.statusText}`);
const data = (await res.json()) as { templates: PromptTemplate[] };
return data.templates;
}
async function createTemplate(input: { title: string; content: string; agent_id?: string | null }): Promise<PromptTemplate> {
const res = await apiFetch(templatesUrl(), {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(input),
});
if (!res.ok) {
const err = (await res.json().catch(() => ({}))) as { detail?: string };
throw new Error(err.detail ?? `Failed to create template: ${res.statusText}`);
}
return res.json() as Promise<PromptTemplate>;
}
async function updateTemplate(id: string, input: { title?: string; content?: string; agent_id?: string | null }): Promise<PromptTemplate> {
const res = await apiFetch(`${templatesUrl()}/${encodeURIComponent(id)}`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(input),
});
if (!res.ok) {
const err = (await res.json().catch(() => ({}))) as { detail?: string };
throw new Error(err.detail ?? `Failed to update template: ${res.statusText}`);
}
return res.json() as Promise<PromptTemplate>;
}
async function deleteTemplate(id: string): Promise<void> {
const res = await apiFetch(`${templatesUrl()}/${encodeURIComponent(id)}`, { method: "DELETE" });
if (!res.ok) {
const err = (await res.json().catch(() => ({}))) as { detail?: string };
throw new Error(err.detail ?? `Failed to delete template: ${res.statusText}`);
}
}
export function usePromptTemplates() {
return useQuery({ queryKey: TEMPLATES_KEY, queryFn: listTemplates });
}
export function useCreatePromptTemplate() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: createTemplate,
onSuccess: () => void queryClient.invalidateQueries({ queryKey: TEMPLATES_KEY }),
});
}
export function useUpdatePromptTemplate() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: ({ id, input }: { id: string; input: { title?: string; content?: string; agent_id?: string | null } }) =>
updateTemplate(id, input),
onSuccess: () => void queryClient.invalidateQueries({ queryKey: TEMPLATES_KEY }),
});
}
export function useDeletePromptTemplate() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: deleteTemplate,
onSuccess: () => void queryClient.invalidateQueries({ queryKey: TEMPLATES_KEY }),
});
}
@@ -12,6 +12,7 @@ import {
BookOpen,
Bot,
Brain,
Briefcase,
CalendarClock,
CircleHelp,
Database,
@@ -246,6 +247,20 @@ export const sidebarMenuItems: MenuItem[] = [
path: "/page/strategy/admin/users",
adminOnly: true,
},
{
id: "mgmt-positions",
label: "岗位管理",
icon: Briefcase,
path: "/page/strategy/admin/positions",
adminOnly: true,
},
{
id: "mgmt-prompt-templates",
label: "提示词模板",
icon: MessageSquareCode,
path: "/page/strategy/admin/prompt-templates",
adminOnly: true,
},
{
id: "mgmt-user-leaderboard",
label: "?????",
+123
View File
@@ -0,0 +1,123 @@
import { apiFetch } from "@/core/api";
import { getBackendBaseURL } from "@/core/config";
import type {
Position,
PositionCreateInput,
PositionListResponse,
PositionMember,
PositionUpdateInput,
PositionUser,
TagBindings,
TagBindingsResponse,
} from "./types";
async function parseDetail(res: Response, fallback: string): Promise<Error> {
const data = (await res.json().catch(() => ({}))) as { detail?: unknown };
const detail = typeof data.detail === "string" && data.detail ? data.detail : null;
return new Error(detail ?? `${fallback} (HTTP ${res.status})`);
}
const base = () => `${getBackendBaseURL()}/api/positions`;
export async function listPositions(): Promise<Position[]> {
const res = await apiFetch(base());
if (!res.ok) throw await parseDetail(res, "Failed to list positions");
const data = (await res.json()) as PositionListResponse;
return data.positions;
}
export async function createPosition(input: PositionCreateInput): Promise<Position> {
const res = await apiFetch(base(), {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(input),
});
if (!res.ok) throw await parseDetail(res, "Failed to create position");
return (await res.json()) as Position;
}
export async function updatePosition(id: string, input: PositionUpdateInput): Promise<Position> {
const res = await apiFetch(`${base()}/${encodeURIComponent(id)}`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(input),
});
if (!res.ok) throw await parseDetail(res, "Failed to update position");
return (await res.json()) as Position;
}
export async function deletePosition(id: string): Promise<void> {
const res = await apiFetch(`${base()}/${encodeURIComponent(id)}`, { method: "DELETE" });
if (!res.ok) throw await parseDetail(res, "Failed to delete position");
}
export async function getTagBindings(positionId: string): Promise<TagBindings> {
const res = await apiFetch(`${base()}/${encodeURIComponent(positionId)}/tag-bindings`);
if (!res.ok) throw await parseDetail(res, "Failed to load tag bindings");
const data = (await res.json()) as TagBindingsResponse;
return data.bindings;
}
export async function setTagBindings(positionId: string, bindings: TagBindings): Promise<TagBindings> {
const res = await apiFetch(`${base()}/${encodeURIComponent(positionId)}/tag-bindings`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ bindings }),
});
if (!res.ok) throw await parseDetail(res, "Failed to save tag bindings");
const data = (await res.json()) as TagBindingsResponse;
return data.bindings;
}
export async function listMembers(positionId: string): Promise<PositionMember[]> {
const res = await apiFetch(`${base()}/${encodeURIComponent(positionId)}/members`);
if (!res.ok) throw await parseDetail(res, "Failed to list members");
const data = (await res.json()) as { members: PositionMember[] };
return data.members;
}
export async function addMembers(positionId: string, userIds: string[]): Promise<PositionMember[]> {
const res = await apiFetch(`${base()}/${encodeURIComponent(positionId)}/members`, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ user_ids: userIds }),
});
if (!res.ok) throw await parseDetail(res, "Failed to add members");
const data = (await res.json()) as { members: PositionMember[] };
return data.members;
}
export async function removeMembers(positionId: string, userIds: string[]): Promise<PositionMember[]> {
const res = await apiFetch(`${base()}/${encodeURIComponent(positionId)}/members`, {
method: "DELETE",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ user_ids: userIds }),
});
if (!res.ok) throw await parseDetail(res, "Failed to remove members");
const data = (await res.json()) as { members: PositionMember[] };
return data.members;
}
export async function resyncPosition(positionId: string): Promise<number> {
const res = await apiFetch(`${base()}/${encodeURIComponent(positionId)}/resync`, { method: "POST" });
if (!res.ok) throw await parseDetail(res, "Failed to resync position");
const data = (await res.json()) as { synced: number };
return data.synced;
}
export async function listPositionUsers(): Promise<PositionUser[]> {
const res = await apiFetch(`${base()}/users`);
if (!res.ok) throw await parseDetail(res, "Failed to list users");
const data = (await res.json()) as { users: PositionUser[] };
return data.users;
}
export async function setUserPosition(userId: string, positionId: string | null): Promise<void> {
const res = await apiFetch(`${base()}/users/${encodeURIComponent(userId)}`, {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ position_id: positionId }),
});
if (!res.ok) throw await parseDetail(res, "Failed to set user position");
}
+124
View File
@@ -0,0 +1,124 @@
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import {
addMembers,
createPosition,
deletePosition,
getTagBindings,
listMembers,
listPositionUsers,
listPositions,
removeMembers,
resyncPosition,
setTagBindings,
setUserPosition,
updatePosition,
} from "./api";
import type {
PositionCreateInput,
PositionUpdateInput,
TagBindings,
} from "./types";
const positionKeys = {
root: ["positions"] as const,
list: ["positions", "list"] as const,
bindings: (id: string) => ["positions", "bindings", id] as const,
members: (id: string) => ["positions", "members", id] as const,
users: ["positions", "users"] as const,
};
export function usePositions() {
return useQuery({ queryKey: positionKeys.list, queryFn: listPositions });
}
export function usePositionUsers() {
return useQuery({ queryKey: positionKeys.users, queryFn: listPositionUsers });
}
export function usePositionTagBindings(positionId: string | null) {
return useQuery({
queryKey: positionKeys.bindings(positionId ?? ""),
queryFn: () => getTagBindings(positionId as string),
enabled: !!positionId,
});
}
export function usePositionMembers(positionId: string | null) {
return useQuery({
queryKey: positionKeys.members(positionId ?? ""),
queryFn: () => listMembers(positionId as string),
enabled: !!positionId,
});
}
function useInvalidate() {
const qc = useQueryClient();
return () => void qc.invalidateQueries({ queryKey: positionKeys.root });
}
export function useCreatePosition() {
const invalidate = useInvalidate();
return useMutation({
mutationFn: (input: PositionCreateInput) => createPosition(input),
onSuccess: invalidate,
});
}
export function useUpdatePosition() {
const invalidate = useInvalidate();
return useMutation({
mutationFn: ({ id, input }: { id: string; input: PositionUpdateInput }) => updatePosition(id, input),
onSuccess: invalidate,
});
}
export function useDeletePosition() {
const invalidate = useInvalidate();
return useMutation({
mutationFn: (id: string) => deletePosition(id),
onSuccess: invalidate,
});
}
export function useSetTagBindings() {
const qc = useQueryClient();
return useMutation({
mutationFn: ({ positionId, bindings }: { positionId: string; bindings: TagBindings }) =>
setTagBindings(positionId, bindings),
onSuccess: (_data, { positionId }) => {
void qc.invalidateQueries({ queryKey: positionKeys.bindings(positionId) });
},
});
}
export function useAddMembers() {
const qc = useQueryClient();
return useMutation({
mutationFn: ({ positionId, userIds }: { positionId: string; userIds: string[] }) =>
addMembers(positionId, userIds),
onSuccess: () => void qc.invalidateQueries({ queryKey: positionKeys.root }),
});
}
export function useRemoveMembers() {
const qc = useQueryClient();
return useMutation({
mutationFn: ({ positionId, userIds }: { positionId: string; userIds: string[] }) =>
removeMembers(positionId, userIds),
onSuccess: () => void qc.invalidateQueries({ queryKey: positionKeys.root }),
});
}
export function useResyncPosition() {
return useMutation({ mutationFn: (positionId: string) => resyncPosition(positionId) });
}
export function useSetUserPosition() {
const qc = useQueryClient();
return useMutation({
mutationFn: ({ userId, positionId }: { userId: string; positionId: string | null }) =>
setUserPosition(userId, positionId),
onSuccess: () => void qc.invalidateQueries({ queryKey: positionKeys.root }),
});
}
+3
View File
@@ -0,0 +1,3 @@
export * from "./types";
export * from "./api";
export * from "./hooks";
+86
View File
@@ -0,0 +1,86 @@
import type { TagType } from "@/core/tags/types";
/** Resource types a 岗位 (position) can grant, by tag. */
export type PositionResourceType =
| "agent"
| "skill"
| "scheduled_task"
| "user_prompt"
| "recommended_question";
export const POSITION_RESOURCE_TYPES: PositionResourceType[] = [
"agent",
"skill",
"scheduled_task",
"user_prompt",
"recommended_question",
];
// A position's resource types are exactly the tag types — reuse the labels.
export const POSITION_RESOURCE_LABELS: Record<PositionResourceType, string> = {
agent: "智能体",
skill: "技能",
scheduled_task: "定时任务",
user_prompt: "常用提示词",
recommended_question: "推荐问题",
};
export const POSITION_RESOURCE_HINT: Record<PositionResourceType, string> = {
agent: "不配置标签 = 默认加载全部内置智能体;配置后只加载所选标签下的智能体。",
skill: "不配置标签 = 可见全部技能;配置后只可见所选标签下的技能。",
scheduled_task: "不配置标签 = 不订阅任何定时任务;配置后订阅所选标签下的定时任务。",
user_prompt: "不配置标签 = 加载未打标签的全局模板;配置后加载所选标签下的提示词模板。",
recommended_question: "不配置标签 = 展示全部推荐问题;配置后只展示所选标签下的推荐问题。",
};
export interface Position {
id: string;
name: string;
description: string;
is_default: boolean;
member_count: number;
created_at: string | null;
updated_at: string | null;
}
export interface PositionListResponse {
positions: Position[];
resource_types: PositionResourceType[];
}
export interface PositionCreateInput {
name: string;
description?: string;
is_default?: boolean;
}
export interface PositionUpdateInput {
name?: string;
description?: string;
is_default?: boolean;
}
/** resource_type -> [tag_id] */
export type TagBindings = Partial<Record<PositionResourceType, string[]>>;
export interface TagBindingsResponse {
position_id: string;
bindings: TagBindings;
}
export interface PositionMember {
id: string;
email: string;
}
export interface PositionUser {
id: string;
email: string;
system_role: string;
position_id: string | null;
}
// The resource types double as tag types for the binding pickers.
export function resourceTypeToTagType(rt: PositionResourceType): TagType {
return rt as TagType;
}
@@ -9,8 +9,10 @@ import type {
const BASE = () => `${getBackendBaseURL()}/api/recommended-questions`;
export async function listRecommendedQuestions(): Promise<RecommendedQuestion[]> {
const res = await apiFetch(BASE());
export async function listRecommendedQuestions(all = false): Promise<RecommendedQuestion[]> {
// ``all=true`` is the admin management view: every question, unfiltered by the
// caller's 岗位 (position). The chat input uses the default (filtered) view.
const res = await apiFetch(`${BASE()}${all ? "?all=1" : ""}`);
if (!res.ok) throw new Error(`Failed to load questions: ${res.statusText}`);
const data = (await res.json()) as { questions: RecommendedQuestion[] };
return data.questions;
@@ -10,11 +10,21 @@ import {
import type { CreateQuestionRequest, UpdateQuestionRequest } from "./index";
const QK = ["recommended-questions"] as const;
const ADMIN_QK = ["recommended-questions", "admin"] as const;
export function useRecommendedQuestions() {
const { data, isLoading, error } = useQuery({
queryKey: QK,
queryFn: listRecommendedQuestions,
queryFn: () => listRecommendedQuestions(false),
});
return { questions: data ?? [], isLoading, error };
}
/** Admin management view — every question, unfiltered by 岗位. */
export function useAdminRecommendedQuestions() {
const { data, isLoading, error } = useQuery({
queryKey: ADMIN_QK,
queryFn: () => listRecommendedQuestions(true),
});
return { questions: data ?? [], isLoading, error };
}
+3
View File
@@ -9,6 +9,8 @@ export interface ListSkillsParams {
sort?: "time_desc" | "time_asc" | "activity_desc" | "activity_asc" | "usage_desc" | "usage_asc" | "name_asc" | "name_desc";
/** Keep only skills filed under this publish square (发布广场). */
square?: string;
/** ``"mine"`` restricts to the skills the caller's 岗位 (position) makes visible. */
scope?: "mine";
}
export async function loadSkills(params: ListSkillsParams = {}) {
@@ -17,6 +19,7 @@ export async function loadSkills(params: ListSkillsParams = {}) {
for (const id of params.tagIds ?? []) qs.append("tag_ids", id);
if (params.sort) qs.set("sort", params.sort);
if (params.square) qs.set("square", params.square);
if (params.scope) qs.set("scope", params.scope);
const query = qs.toString();
const skills = await apiFetch(
`${getBackendBaseURL()}/api/skills${query ? `?${query}` : ""}`,
+3 -2
View File
@@ -42,9 +42,10 @@ export function useSkills(params: ListSkillsParams = {}) {
const search = params.search ?? "";
const tagIds = params.tagIds ?? [];
const sort = params.sort ?? "time_desc";
const scope = params.scope;
const { data, isLoading, error } = useQuery({
queryKey: ["skills", search, [...tagIds].sort(), sort],
queryFn: () => loadSkills({ search, tagIds, sort }),
queryKey: ["skills", search, [...tagIds].sort(), sort, scope ?? "all"],
queryFn: () => loadSkills({ search, tagIds, sort, scope }),
placeholderData: keepPreviousData,
});
return { skills: data ?? EMPTY_SKILLS, isLoading, error };
+22 -2
View File
@@ -1,6 +1,26 @@
export type TagType = "agent" | "skill" | "scheduled_task";
export type TagType =
| "agent"
| "skill"
| "scheduled_task"
| "user_prompt"
| "recommended_question";
export const TAG_TYPES: TagType[] = ["agent", "skill", "scheduled_task"];
export const TAG_TYPES: TagType[] = [
"agent",
"skill",
"scheduled_task",
"user_prompt",
"recommended_question",
];
// Human labels for the tag types (used by the tag management + 岗位 pages).
export const TAG_TYPE_LABELS: Record<TagType, string> = {
agent: "智能体",
skill: "技能",
scheduled_task: "定时任务",
user_prompt: "常用提示词",
recommended_question: "推荐问题",
};
export interface Tag {
id: string;
+17
View File
@@ -343,6 +343,23 @@ export function useThreadStream({
) {
const e = event as { type: "llm_retry"; message: string };
toast(e.message);
return;
}
// Context compression about to run a (blocking) summarization LLM call —
// surface a transient hint so the wait before the answer streams isn't a
// silent dead air. Emitted by DeerFlowSummarizationMiddleware.
if (
typeof event === "object" &&
event !== null &&
"type" in event &&
event.type === "context_compacting" &&
"message" in event &&
typeof event.message === "string" &&
event.message.trim()
) {
const e = event as { type: "context_compacting"; message: string };
toast(e.message);
}
},
onError(error) {
+5
View File
@@ -0,0 +1,5 @@
import { PositionManagementPage } from "@/components/workspace/admin/position-management-page";
export default function PositionsPage() {
return <PositionManagementPage />;
}
@@ -0,0 +1,5 @@
import { PromptTemplatesPage as PromptTemplatesManagement } from "@/components/workspace/admin/prompt-templates-page";
export default function PromptTemplatesPage() {
return <PromptTemplatesManagement />;
}
@@ -8,11 +8,13 @@ import {
Loader2Icon,
PencilIcon,
PlusIcon,
TagIcon,
Trash2Icon,
} from "lucide-react";
import { toast } from "sonner";
import { Button } from "@/components/ui/button";
import { ResourceTagDialog } from "@/components/workspace/tags/resource-tag-dialog";
import { useTheme } from "@/contexts/ThemeContext";
import {
Dialog,
@@ -24,9 +26,9 @@ import {
import { Input } from "@/components/ui/input";
import { Textarea } from "@/components/ui/textarea";
import {
useAdminRecommendedQuestions,
useCreateRecommendedQuestion,
useDeleteRecommendedQuestion,
useRecommendedQuestions,
useReorderRecommendedQuestions,
useUpdateRecommendedQuestion,
} from "@/core/recommended-questions/hooks";
@@ -65,7 +67,7 @@ export default function RecommendedQuestionsPage({
embedded?: boolean;
} = {}) {
const { theme } = useTheme();
const { questions, isLoading, error } = useRecommendedQuestions();
const { questions, isLoading, error } = useAdminRecommendedQuestions();
const createMutation = useCreateRecommendedQuestion();
const updateMutation = useUpdateRecommendedQuestion();
const deleteMutation = useDeleteRecommendedQuestion();
@@ -75,6 +77,7 @@ export default function RecommendedQuestionsPage({
const [formTitle, setFormTitle] = useState("");
const [formContent, setFormContent] = useState("");
const [deleteTarget, setDeleteTarget] = useState<RecommendedQuestion | null>(null);
const [tagTarget, setTagTarget] = useState<RecommendedQuestion | null>(null);
// drag state
const [dragIndex, setDragIndex] = useState<number | null>(null);
@@ -245,6 +248,16 @@ export default function RecommendedQuestionsPage({
</div>
</div>
<div className="flex shrink-0 items-center gap-1">
<Button
variant="ghost"
size="icon"
className="size-7"
onClick={() => setTagTarget(q)}
disabled={isSaving}
title="打标签(按岗位分发)"
>
<TagIcon className="size-3.5" />
</Button>
<Button
variant="ghost"
size="icon"
@@ -308,6 +321,17 @@ export default function RecommendedQuestionsPage({
</DialogContent>
</Dialog>
{/* 打标签弹窗(按岗位分发推荐问题) */}
{tagTarget && (
<ResourceTagDialog
open={tagTarget !== null}
onOpenChange={(o) => !o && setTagTarget(null)}
targetType="recommended_question"
targetId={tagTarget.id}
title={`为「${tagTarget.title}」打标签`}
/>
)}
{/* 删除确认弹窗 */}
<Dialog
open={deleteTarget !== null}
+22
View File
@@ -40,6 +40,8 @@ import ScheduledTaskRunDetailPage from "./ScheduledTaskRunDetailPage";
import ScheduledTasksPage from "./ScheduledTasksPage";
import SkillEvolutionPage from "./SkillEvolutionPage";
import SkillsPage from "./SkillsPage";
import PositionsPage from "./PositionsPage";
import PromptTemplatesPage from "./PromptTemplatesPage";
import TagsPage from "./TagsPage";
import UsersPage from "./UsersPage";
import LlmMetricsPage from "./LlmMetricsPage";
@@ -287,6 +289,26 @@ export default function StrategyRoutes() {
path="admin/tags"
element={<TagsPage />}
/>
<Route
path="admin/positions"
element={
getStoredAuth()?.system_role === "admin" ? (
<PositionsPage />
) : (
<Navigate to="/page/workspace/chats/new" replace />
)
}
/>
<Route
path="admin/prompt-templates"
element={
getStoredAuth()?.system_role === "admin" ? (
<PromptTemplatesPage />
) : (
<Navigate to="/page/workspace/chats/new" replace />
)
}
/>
</Routes>
</div>
</SidebarProvider>
@@ -39,6 +39,8 @@ import AdminUserActivityPage from './AdminUserActivityPage';
import AdminMemoryPage from './AdminMemoryPage';
import AdminUserHubPage from './AdminUserHubPage';
import TagsPage from './TagsPage';
import PositionsPage from './PositionsPage';
import PromptTemplatesPage from './PromptTemplatesPage';
import RecommendedQuestionsPage from './RecommendedQuestionsPage';
import AIWritingPage from '@/open-canvas/pages/AIWritingPage';
import StudioNotebooksPage from './StudioNotebooksPage';
@@ -471,6 +473,30 @@ export default function WorkspaceRoutes({
</WorkspaceLayout>
}
/>
<Route
path="admin/positions"
element={
getStoredAuth()?.system_role === 'admin' ? (
<WorkspaceLayout {...layoutProps}>
<PositionsPage />
</WorkspaceLayout>
) : (
<Navigate to="/page/workspace/chats/new" replace />
)
}
/>
<Route
path="admin/prompt-templates"
element={
getStoredAuth()?.system_role === 'admin' ? (
<WorkspaceLayout {...layoutProps}>
<PromptTemplatesPage />
</WorkspaceLayout>
) : (
<Navigate to="/page/workspace/chats/new" replace />
)
}
/>
<Route
path="settings/recommended-questions"
element={
+57 -2
View File
@@ -163,7 +163,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har
6. **GuardrailMiddleware** - Pre-tool-call authorization via pluggable `GuardrailProvider` protocol (optional, if `guardrails.enabled` in config). Evaluates each tool call and returns error ToolMessage on deny. Three provider options: built-in `AllowlistProvider` (zero deps), OAP policy providers (e.g. `aport-agent-guardrails`), or custom providers. See [docs/GUARDRAILS.md](docs/GUARDRAILS.md) for setup, usage, and how to implement a provider.
7. **SandboxAuditMiddleware** - Audits sandboxed shell/file operations for security logging before tool execution continues
8. **ToolErrorHandlingMiddleware** - Converts tool exceptions into error `ToolMessage`s so the run can continue instead of aborting
9. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` ? the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached per-thread (bounded, sticky boundary) to avoid an LLM call on every model invocation. Tests: `tests/test_summarization_middleware.py`
9. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` ? the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached **per-thread in a process-wide, thread-safe LRU** (`_GLOBAL_SUMMARY_CACHE`, bounded to 512 threads, sticky boundary). The cache is module-level — **not** on the middleware instance — because `make_lead_agent` rebuilds a fresh middleware on every run; an instance cache would be cold each turn and force a blocking summary LLM call on every turn of a long conversation. Keyed by `thread_id` (resolved like `ThreadDataMiddleware`), so a summary computed on one turn is reused on the next, paying for an LLM call only on the first threshold crossing or a genuine re-compaction. **Threshold-first gating**: `_plan_compression` returns passthrough (no compression at all) whenever the *full* current conversation is still under the trigger — the cache/summary substitution only engages once the real context crosses the configured trigger. When the genuine (blocking) summarize path runs, it emits a `{"type":"context_compacting","message":"正在压缩上下文…"}` event on LangGraph's `custom` stream channel (the cheap reuse path stays silent); the chat frontend's `onCustomEvent` renders it as a transient toast. Tests: `tests/test_summarization_middleware.py`
10. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode)
11. **TokenUsageMiddleware** - Records token usage metrics when token tracking is enabled (optional)
12. **TitleMiddleware** - Two-phase auto-title (never shows "untitled"): `before_model` sets an instant question-derived *provisional* title the moment the first user message arrives (no LLM, zero latency), then `after_model` asynchronously generates the final LLM title and replaces it (tracked via the `title_provisional` state flag). Normalizes structured message content before prompting the title model
@@ -175,7 +175,7 @@ Lead-agent middlewares are assembled in strict append order across `packages/har
18. **ClarificationMiddleware** - Intercepts `ask_clarification` tool calls, interrupts via `Command(goto=END)` (must be last)
8a. **ToolMetricsMiddleware** - Appended immediately after `ToolErrorHandlingMiddleware` in `_build_runtime_middlewares`, so it sits *deepest* in the `wrap_tool_call` chain. Records one `tool_call_metrics` row per tool/skill/MCP invocation ? timing duration and bucketing failures into a coarse `error_category` (rate_limited / ip_blocked / auth_error / timeout / network_error / not_found / empty_result / tool_error). It catches both raised exceptions (before #8 converts them) and *self-handled* errors returned as error `ToolMessage`s (search tools that swallow HTTP 429/403). It also captures the targeted Agent Skill into `tool_call_metrics.skill_name`: the lead agent invokes a skill by `read_file`-ing its `/mnt/skills/<category>/<name>/SKILL.md` path (per the `<skill_system>` prompt block ? there is no dedicated "use skill" tool), so `skill_name` is recovered by scanning every tool call's string arguments for the `skills/{public,custom}/{name}/` path shape (`_SKILL_PATH_RE`), plus the explicit `name` argument of `skill_view`/`skill_manage`/`skill_archive`. `skill_view`-specific success detection handles its swallowed failures (`Skill 'x' not found.` returned as a plain string rather than raised). Regression coverage: `tests/test_tool_metrics_middleware.py`. Writes are best-effort fire-and-forget. **Per-run skill merging**: `SqlToolMetricsStore.record` upserts skill rows by `(run_id, skill_name)` instead of inserting one row per call ? a skill touched by several tool calls / retries inside one run collapses to a single row. Any success in the run wins (a skill that failed then finally succeeded is recorded as one `success`); a skill that only ever failed yields one `error` row whose `error_message` accumulates every attempt's message; durations are summed. Non-skill tool calls, and skill calls with no `run_id`, keep one row each. The read?merge?write is guarded by a per-event-loop `asyncio.Lock`. Surfaced by the admin `GET /api/admin/tool-metrics` router (which has a `kind=skill` filter and a per-skill `/summary` breakdown).
9. **PromptPrefixMiddleware** - Injects the admin-configured prompt prefix into the model request only. The prefix is tagged onto the first human message's `additional_kwargs['prompt_prefix']` by the Gateway (`app.gateway.services`) and prepended to content here at call time, so the persisted/displayed message stays prefix-free
10. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` ? the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached per-thread (bounded, sticky boundary) to avoid an LLM call on every model invocation. Tests: `tests/test_summarization_middleware.py`
10. **SummarizationMiddleware** (`DeerFlowSummarizationMiddleware`) - Context reduction when approaching token limits (optional, if enabled). **Non-destructive**: overrides the stock `before_model` (which persists a `RemoveMessage(REMOVE_ALL_MESSAGES)`) to a no-op and instead compresses **transiently** in `wrap_model_call` ? the model request is replaced with `[summary, *recent]` while the full original conversation stays in state (so the frontend keeps rendering the user's real Q&A). The `summary` message is `hide_from_ui` and never persisted; summaries are cached **per-thread in a process-wide, thread-safe LRU** (`_GLOBAL_SUMMARY_CACHE`, bounded to 512 threads, sticky boundary). The cache is module-level — **not** on the middleware instance — because `make_lead_agent` rebuilds a fresh middleware on every run; an instance cache would be cold each turn and force a blocking summary LLM call on every turn of a long conversation. Keyed by `thread_id` (resolved like `ThreadDataMiddleware`), so a summary computed on one turn is reused on the next, paying for an LLM call only on the first threshold crossing or a genuine re-compaction. **Threshold-first gating**: `_plan_compression` returns passthrough (no compression at all) whenever the *full* current conversation is still under the trigger — the cache/summary substitution only engages once the real context crosses the configured trigger. When the genuine (blocking) summarize path runs, it emits a `{"type":"context_compacting","message":"正在压缩上下文…"}` event on LangGraph's `custom` stream channel (the cheap reuse path stays silent); the chat frontend's `onCustomEvent` renders it as a transient toast. Tests: `tests/test_summarization_middleware.py`
11. **TodoListMiddleware** - Task tracking with `write_todos` tool (optional, if plan_mode)
12. **TokenUsageMiddleware** - Records token usage metrics when token tracking is enabled (optional)
13. **TitleMiddleware** - Two-phase auto-title (never shows "untitled"): `before_model` sets an instant question-derived *provisional* title the moment the first user message arrives (no LLM, zero latency), then `after_model` asynchronously generates the final LLM title and replaces it (tracked via the `title_provisional` state flag). Normalizes structured message content before prompting the title model
@@ -328,6 +328,61 @@ Admin-configurable, **access-control-free** categories that published agents/ski
Each agent/skill carries a single `square_id` column (`AgentRow` / `SkillRow`, `String(64)` NOT NULL default `''`, migration `20260611_01`, idempotent SQL in `column_additions_mysql.sql`). One column = at most one square, so an item can never be "published twice". `''` = unassigned ? resolved to the default square at read/filter time. Writes normalize via `normalize_square_id`: agents on create/update (`square_id` field), skills on publish (`PUT /api/skills/custom/{name}/published` accepts optional `square_id`). Listing filters: `GET /api/agents?square=` and `GET /api/skills?square=` keep only that square's items; the **default** square additionally sweeps in legacy `''` rows (`square_is_default`). Tests: `tests/test_publish_squares.py`. Frontend: `core/system-settings` (`usePublishSquares`/`useUpdatePublishSquares`), admin `PublishSquaresSettingsPage` (embedded in the appearance/settings hub), square picker in the agent edit dialog + per-skill card, and a ??-tab square sub-filter in both galleries.
### 岗位 (Position) Subscription System
An admin-managed aggregation layer that decides, per user, which agents / skills
/ scheduled tasks / custom prompts / recommended questions they are subscribed
to or can see. **A user has at most one position** (`users.position_id`, 1:1);
unassigned users inherit the single `is_default` position.
**Persistence** (`deerflow/persistence/positions/`, wired as
`app.state.position_store`; tables auto-created by `create_all`, migration
`20260616_01`):
- `positions` (`PositionRow`): `id`, `name` (unique), `description`,
`is_default` (only one at a time — the store clears the flag on others).
- `position_tag_bindings` (`PositionTagBindingRow`): one row per
`(position, resource_type, tag)`. **No binding for a resource_type = "use the
default set".** `POSITION_RESOURCE_TYPES = (agent, skill, scheduled_task,
user_prompt, recommended_question)` — the same set drives the tag types
(`TAG_TYPES` was extended with `user_prompt` + `recommended_question`).
**Sync engine** (`deerflow/services/position_sync.py`, wired as
`app.state.position_sync`). `PositionSyncService.resolve_visible(user, type)`
returns `set | None` (None = unrestricted) for the **query-time** visibility
types (`skill`, `recommended_question`). `sync_user(user)` **materializes** the
other three into the existing "我的" tables, idempotently:
- agents → `agent_favorites` rows with `origin='position'`
- scheduled tasks → `scheduled_task_subscriptions` rows with `origin='position'`
- prompts → clones of admin **templates** into `user_custom_prompts`
(`origin='position'`, `template_id` set)
Sync only ever adds/removes `origin='position'` rows, so a user's own
favorites/subscriptions/prompts are never clobbered. Default sets: agents =
built-ins (`user_id IS NULL`); scheduled tasks = empty; prompts = untagged
templates; skills/questions = unrestricted. `origin`/`position_id` columns were
added to `agent_favorites`, `scheduled_task_subscriptions`, and
`user_custom_prompts` (+ `is_template`/`template_id`) — see migration
`20260616_01` + `scripts/sql/column_additions_mysql.sql`. The running app also
auto-adds these via `_ensure_orm_columns_sync` on startup.
**Triggers**: assigning/removing members, changing tag bindings (re-syncs all
members), becoming the default position (`sync_default_members`), and login
(`auth.py::_sync_user_position`, best-effort so a brand-new user inheriting the
default gets their grants materialized).
**API** (`app/gateway/routers/positions.py`, `/api/positions`, **admin only**):
position CRUD; `GET/PUT /{id}/tag-bindings`; `GET/POST/DELETE /{id}/members`
(batch assign/unassign → triggers sync); `POST /{id}/resync`;
`GET /users` (all users + their position); `PUT /users/{user_id}` (set one
user's position). Related changes: prompt **templates** CRUD under
`/api/user-prompts/templates` (admin; `is_template` rows excluded from a user's
own `GET /api/user-prompts`); `GET /api/recommended-questions` is now filtered
by the caller's position (`?all=1` bypasses for admin management);
`GET /api/skills?scope=mine` applies skill visibility. Frontend: `core/positions`
+ admin `PositionManagementPage` / `PromptTemplatesPage` (侧边栏「综合管理」),
recommended-questions moved to a bottom-right Popover in the chat input. Tests:
`tests/test_positions.py`.
### Portable Column Types (`deerflow/persistence/types.py`)
Two `TypeDecorator`s keep ORM columns correct across the `sqlite` / `postgres` / `mysql` backends. **All ORM timestamp/JSON columns must use these, not the raw SQLAlchemy types.**
@@ -50,6 +50,7 @@ from app.gateway.routers import (
notifications,
parallel_agent_tasks,
parallel_agents,
positions,
public_embed,
public_leaderboard,
public_scheduled_tasks,
@@ -1397,6 +1398,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# Tags API is mounted at /api/tags
app.include_router(tags.router)
# 岗位 (position) management API is mounted at /api/positions
app.include_router(positions.router)
# Thread Runs API (LangGraph Platform-compatible runs lifecycle)
app.include_router(thread_runs.router)
@@ -133,6 +133,14 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
app.state.tag_store = make_tag_store(sf)
from deerflow.persistence.positions import make_position_store
app.state.position_store = make_position_store(sf)
from deerflow.services.position_sync import PositionSyncService
app.state.position_sync = PositionSyncService(sf, app.state.position_store, app.state.tag_store)
from deerflow.persistence.thread_shares import make_thread_share_store
app.state.thread_share_store = make_thread_share_store(sf)
@@ -371,6 +379,22 @@ def get_tag_store(request: Request) -> TagStore:
return val
def get_position_store(request: Request):
"""Return the 岗位 (position) store."""
val = getattr(request.app.state, "position_store", None)
if val is None:
raise HTTPException(status_code=503, detail="Position store not available")
return val
def get_position_sync(request: Request):
"""Return the position → subscription sync service."""
val = getattr(request.app.state, "position_sync", None)
if val is None:
raise HTTPException(status_code=503, detail="Position sync service not available")
return val
def get_thread_share_store(request: Request):
"""Return the conversation share-link store."""
val = getattr(request.app.state, "thread_share_store", None)
@@ -202,6 +202,21 @@ def _username_to_email(username: str) -> str:
return f"{local}@{domain}"
async def _sync_user_position(request: Request, user_id: str) -> None:
"""Best-effort: materialize the user's 岗位 (position) grants on login.
Ensures a user who has never been synced — e.g. a brand-new account that
inherits the ``is_default`` position — gets the default agents / tasks /
prompts loaded into their "我的" tables. Failures never block login.
"""
try:
sync = getattr(request.app.state, "position_sync", None)
if sync is not None:
await sync.sync_user(str(user_id))
except Exception:
logger.debug("position sync on login failed for user %s", user_id, exc_info=True)
async def _get_or_create_user_by_username(username: str):
"""Resolve a username to a local user, auto-registering when absent.
@@ -476,6 +491,7 @@ async def login_local(
_record_login_success(client_ip)
token = create_access_token(str(user.id), token_version=user.token_version)
_set_session_cookie(response, token, request)
await _sync_user_position(request, user.id)
return LoginResponse(
expires_in=get_auth_config().token_expiry_days * 24 * 3600,
@@ -520,6 +536,7 @@ async def login_by_username(
token = create_access_token(str(user.id), token_version=user.token_version)
_set_session_cookie(response, token, request)
await _sync_user_position(request, user.id)
return UsernameLoginResponse(
access_token=token,
@@ -570,6 +587,7 @@ async def login_by_token(
access_token = create_access_token(str(user.id), token_version=user.token_version)
_set_session_cookie(response, access_token, request)
await _sync_user_position(request, user.id)
return UsernameLoginResponse(
access_token=access_token,
@@ -600,6 +618,7 @@ async def register(request: Request, response: Response, body: RegisterRequest):
token = create_access_token(str(user.id), token_version=user.token_version)
_set_session_cookie(response, token, request)
await _sync_user_position(request, user.id)
return UserResponse(id=str(user.id), email=user.email, system_role=user.system_role)
@@ -0,0 +1,311 @@
"""岗位 (position) management API — admin only.
A position aggregates, per resource type, the tags whose resources are granted
to its members. Assigning a user to a position (1:1) materializes those grants
into the user's "我的" tables via the sync engine; recommended-question and
skill visibility are filtered live elsewhere.
All endpoints require ``system_role == "admin"`` (auth-disabled mode lets calls
through, matching the other admin routers).
"""
from __future__ import annotations
import logging
import uuid
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, Field
from app.gateway.deps import get_optional_user_from_request, get_position_store, get_position_sync, get_tag_store
from deerflow.persistence.positions import POSITION_RESOURCE_TYPES, PositionStore
from deerflow.persistence.tags import TagStore
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/positions", tags=["positions"])
async def _require_admin(request: Request) -> None:
user = await get_optional_user_from_request(request)
if user is None:
return
if getattr(user, "system_role", None) != "admin":
raise HTTPException(status_code=403, detail="岗位管理仅限管理员")
def _validate_bindings(bindings: dict[str, list[str]]) -> dict[str, list[str]]:
out: dict[str, list[str]] = {}
for resource_type, tag_ids in bindings.items():
if resource_type not in POSITION_RESOURCE_TYPES:
raise HTTPException(status_code=422, detail=f"Invalid resource type '{resource_type}'. Allowed: {', '.join(POSITION_RESOURCE_TYPES)}")
out[resource_type] = list(tag_ids or [])
return out
# ------------------------------------------------------------------ schemas
class PositionResponse(BaseModel):
id: str
name: str
description: str = ""
is_default: bool = False
member_count: int = 0
created_at: str | None = None
updated_at: str | None = None
class PositionListResponse(BaseModel):
positions: list[PositionResponse]
resource_types: list[str] = Field(default_factory=lambda: list(POSITION_RESOURCE_TYPES))
class PositionCreateRequest(BaseModel):
name: str = Field(..., min_length=1, max_length=191)
description: str = Field(default="", max_length=1024)
is_default: bool = False
class PositionUpdateRequest(BaseModel):
name: str | None = Field(default=None, min_length=1, max_length=191)
description: str | None = Field(default=None, max_length=1024)
is_default: bool | None = None
class TagBindingsResponse(BaseModel):
position_id: str
bindings: dict[str, list[str]]
class TagBindingsRequest(BaseModel):
bindings: dict[str, list[str]] = Field(default_factory=dict, description="resource_type -> [tag_id]; omitted/empty type = default set")
class MemberResponse(BaseModel):
id: str
email: str = ""
class MembersResponse(BaseModel):
members: list[MemberResponse]
class MembersMutationRequest(BaseModel):
user_ids: list[str] = Field(..., min_length=1)
class UserPositionRequest(BaseModel):
position_id: str | None = Field(default=None, description="null clears the user's position")
class PositionUserResponse(BaseModel):
id: str
email: str = ""
system_role: str = "user"
position_id: str | None = None
class PositionUsersResponse(BaseModel):
users: list[PositionUserResponse]
# ----------------------------------------------------------------- position CRUD
@router.get("", response_model=PositionListResponse, summary="List Positions")
async def list_positions(request: Request, store: PositionStore = Depends(get_position_store)) -> PositionListResponse:
await _require_admin(request)
records = await store.list_positions()
return PositionListResponse(positions=[PositionResponse(**r) for r in records])
@router.get("/users", response_model=PositionUsersResponse, summary="List All Users + Their Position")
async def list_position_users(request: Request, store: PositionStore = Depends(get_position_store)) -> PositionUsersResponse:
await _require_admin(request)
users = await store.list_all_users()
return PositionUsersResponse(users=[PositionUserResponse(**u) for u in users])
@router.post("", response_model=PositionResponse, status_code=201, summary="Create Position")
async def create_position(request: Request, body: PositionCreateRequest, store: PositionStore = Depends(get_position_store)) -> PositionResponse:
await _require_admin(request)
try:
record = await store.create_position({"id": uuid.uuid4().hex, "name": body.name.strip(), "description": body.description, "is_default": body.is_default})
return PositionResponse(**record)
except ValueError as e:
raise HTTPException(status_code=409, detail=str(e))
@router.put("/{position_id}", response_model=PositionResponse, summary="Update Position")
async def update_position(
request: Request,
position_id: str,
body: PositionUpdateRequest,
store: PositionStore = Depends(get_position_store),
sync=Depends(get_position_sync),
) -> PositionResponse:
await _require_admin(request)
update_data = body.model_dump(exclude_unset=True)
if update_data.get("name"):
update_data["name"] = update_data["name"].strip()
try:
record = await store.update_position(position_id, update_data)
except ValueError as e:
raise HTTPException(status_code=409, detail=str(e))
if record is None:
raise HTTPException(status_code=404, detail="Position not found")
# Becoming the default changes what unassigned users inherit.
if update_data.get("is_default"):
await sync.sync_default_members()
return PositionResponse(**record)
@router.delete("/{position_id}", status_code=204, summary="Delete Position")
async def delete_position(
request: Request,
position_id: str,
store: PositionStore = Depends(get_position_store),
sync=Depends(get_position_sync),
) -> None:
await _require_admin(request)
member_ids = await store.list_member_ids(position_id)
deleted = await store.delete_position(position_id)
if not deleted:
raise HTTPException(status_code=404, detail="Position not found")
# Former members now fall back to the default position; re-sync them.
for uid in member_ids:
await sync.sync_user(uid)
# ----------------------------------------------------------------- tag bindings
@router.get("/{position_id}/tag-bindings", response_model=TagBindingsResponse, summary="Get Tag Bindings")
async def get_tag_bindings(request: Request, position_id: str, store: PositionStore = Depends(get_position_store)) -> TagBindingsResponse:
await _require_admin(request)
if await store.get_position(position_id) is None:
raise HTTPException(status_code=404, detail="Position not found")
bindings = await store.get_tag_bindings(position_id)
return TagBindingsResponse(position_id=position_id, bindings=bindings)
@router.put("/{position_id}/tag-bindings", response_model=TagBindingsResponse, summary="Set Tag Bindings")
async def set_tag_bindings(
request: Request,
position_id: str,
body: TagBindingsRequest,
store: PositionStore = Depends(get_position_store),
tag_store: TagStore = Depends(get_tag_store),
sync=Depends(get_position_sync),
) -> TagBindingsResponse:
await _require_admin(request)
if await store.get_position(position_id) is None:
raise HTTPException(status_code=404, detail="Position not found")
cleaned = _validate_bindings(body.bindings)
# Validate every tag id exists and matches its resource type.
for resource_type, tag_ids in cleaned.items():
if not tag_ids:
continue
known = {t["id"] for t in await tag_store.list_tags(resource_type)}
for tid in tag_ids:
if tid not in known:
raise HTTPException(status_code=422, detail=f"Tag '{tid}' is not a '{resource_type}' tag")
bindings = await store.set_tag_bindings(position_id, cleaned)
await sync.sync_position(position_id)
return TagBindingsResponse(position_id=position_id, bindings=bindings)
# --------------------------------------------------------------------- members
@router.get("/{position_id}/members", response_model=MembersResponse, summary="List Position Members")
async def list_members(request: Request, position_id: str, store: PositionStore = Depends(get_position_store)) -> MembersResponse:
await _require_admin(request)
if await store.get_position(position_id) is None:
raise HTTPException(status_code=404, detail="Position not found")
members = await store.list_members(position_id)
return MembersResponse(members=[MemberResponse(**m) for m in members])
@router.post("/{position_id}/members", response_model=MembersResponse, summary="Add Position Members")
async def add_members(
request: Request,
position_id: str,
body: MembersMutationRequest,
store: PositionStore = Depends(get_position_store),
sync=Depends(get_position_sync),
) -> MembersResponse:
await _require_admin(request)
if await store.get_position(position_id) is None:
raise HTTPException(status_code=404, detail="Position not found")
# Re-sync users leaving their previous position too, so its grants are dropped.
prev_positions = {uid: await store.get_user_position_id(uid) for uid in body.user_ids}
await store.assign_users(position_id, body.user_ids)
for uid in body.user_ids:
await sync.sync_user(uid)
prev = prev_positions.get(uid)
if prev and prev != position_id:
logger.debug("user %s moved from position %s to %s", uid, prev, position_id)
members = await store.list_members(position_id)
return MembersResponse(members=[MemberResponse(**m) for m in members])
@router.delete("/{position_id}/members", response_model=MembersResponse, summary="Remove Position Members")
async def remove_members(
request: Request,
position_id: str,
body: MembersMutationRequest,
store: PositionStore = Depends(get_position_store),
sync=Depends(get_position_sync),
) -> MembersResponse:
await _require_admin(request)
if await store.get_position(position_id) is None:
raise HTTPException(status_code=404, detail="Position not found")
# Only unassign users that are actually in this position.
member_ids = set(await store.list_member_ids(position_id))
victims = [uid for uid in body.user_ids if uid in member_ids]
await store.unassign_users(victims)
for uid in victims:
await sync.sync_user(uid)
members = await store.list_members(position_id)
return MembersResponse(members=[MemberResponse(**m) for m in members])
@router.post("/{position_id}/resync", summary="Re-sync Position Members")
async def resync_position(
request: Request,
position_id: str,
store: PositionStore = Depends(get_position_store),
sync=Depends(get_position_sync),
) -> dict:
await _require_admin(request)
if await store.get_position(position_id) is None:
raise HTTPException(status_code=404, detail="Position not found")
count = await sync.sync_position(position_id)
return {"synced": count}
# --------------------------------------------------- per-user position (1:1)
@router.put("/users/{user_id}", response_model=PositionResponse | None, summary="Set User Position")
async def set_user_position(
request: Request,
user_id: str,
body: UserPositionRequest,
store: PositionStore = Depends(get_position_store),
sync=Depends(get_position_sync),
) -> PositionResponse | None:
await _require_admin(request)
if body.position_id:
if await store.get_position(body.position_id) is None:
raise HTTPException(status_code=404, detail="Position not found")
await store.assign_users(body.position_id, [user_id])
else:
await store.unassign_users([user_id])
await sync.sync_user(user_id)
if not body.position_id:
return None
record = await store.get_position(body.position_id)
return PositionResponse(**record) if record else None
@@ -6,13 +6,22 @@ import logging
from datetime import datetime
from uuid import uuid4
from fastapi import APIRouter, HTTPException, Request
from fastapi import APIRouter, HTTPException, Query, Request
from pydantic import BaseModel, Field
from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api", tags=["recommended-questions"])
def _current_user_id(request: Request) -> str:
user = getattr(request.state, "user", None)
if user is not None:
return str(user.id)
return get_effective_user_id()
class RecommendedQuestionResponse(BaseModel):
id: str
title: str
@@ -50,9 +59,25 @@ def _get_store(request: Request):
@router.get("/recommended-questions", response_model=RecommendedQuestionsListResponse)
async def list_recommended_questions(request: Request) -> RecommendedQuestionsListResponse:
async def list_recommended_questions(
request: Request,
all: bool = Query(default=False, description="Admin management view: return every question, unfiltered by 岗位."),
) -> RecommendedQuestionsListResponse:
store = _get_store(request)
rows = await store.list_all()
if not all:
# Filter by the caller's 岗位 (position) visibility. ``None`` =
# unrestricted (no skill/question tag binding → show all default
# questions); a set = show only the questions carrying the bound tags.
sync = getattr(request.app.state, "position_sync", None)
if sync is not None:
try:
visible = await sync.resolve_visible(_current_user_id(request), "recommended_question")
except Exception:
logger.debug("recommended-question visibility resolve failed", exc_info=True)
visible = None
if visible is not None:
rows = [r for r in rows if r["id"] in visible]
return RecommendedQuestionsListResponse(questions=[RecommendedQuestionResponse(**r) for r in rows])
@@ -777,6 +777,7 @@ async def list_skills(
search: str | None = Query(default=None, description="Case-insensitive fuzzy match on skill name or Chinese name (name_zh)"),
tag_ids: list[str] | None = Query(default=None, description="Keep skills carrying any of these tags (OR)"),
square: str | None = Query(default=None, description="Keep only skills filed under this publish square (发布广场). Omit for all squares."),
scope: str | None = Query(default=None, description="``mine`` filters to the skills the caller's 岗位 (position) makes visible."),
sort: str = Query(
default="time_desc",
pattern="^(time_desc|time_asc|activity_desc|activity_asc|usage_desc|usage_asc|name_asc|name_desc)$",
@@ -846,6 +847,19 @@ async def list_skills(
for r in responses
if (r.square_id or "") == square_id or (square_is_default and not (r.square_id or ""))
]
# 岗位 (position) visibility: when the caller asks for "我的技能",
# restrict to the skills their position grants. ``None`` from the sync
# service = unrestricted (no skill tag binding → all skills visible).
if scope == "mine" and not admin:
sync = getattr(request.app.state, "position_sync", None)
if sync is not None:
try:
visible = await sync.resolve_visible(user_id, "skill")
except Exception:
logger.debug("skill visibility resolve failed", exc_info=True)
visible = None
if visible is not None:
responses = [r for r in responses if r.name in visible]
responses = _sort_skill_responses(responses, sort)
return SkillsListResponse(skills=responses)
except HTTPException:
@@ -13,12 +13,21 @@ from uuid import uuid4
from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel, Field
from app.gateway.deps import get_optional_user_from_request
from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/user-prompts", tags=["user-prompts"])
async def _require_admin(request: Request) -> None:
user = await get_optional_user_from_request(request)
if user is None:
return
if getattr(user, "system_role", None) != "admin":
raise HTTPException(status_code=403, detail="提示词模板管理仅限管理员")
class UserPromptResponse(BaseModel):
id: str
title: str = ""
@@ -26,10 +35,40 @@ class UserPromptResponse(BaseModel):
sort_order: int = 0
# Optional single-agent association. NULL = global (shown for every agent).
agent_id: str | None = None
# ``"user"`` (the user's own) or ``"position"`` (granted by their 岗位).
origin: str = "user"
created_at: datetime | str | None = None
updated_at: datetime | str | None = None
class PromptTemplateResponse(BaseModel):
id: str
title: str = ""
content: str
sort_order: int = 0
agent_id: str | None = None
created_at: datetime | str | None = None
updated_at: datetime | str | None = None
class PromptTemplatesListResponse(BaseModel):
templates: list[PromptTemplateResponse]
class PromptTemplateCreateRequest(BaseModel):
title: str = Field(default="", max_length=255)
content: str = Field(..., min_length=1)
sort_order: int = Field(default=0)
agent_id: str | None = Field(default=None, max_length=64)
class PromptTemplateUpdateRequest(BaseModel):
title: str | None = Field(default=None, max_length=255)
content: str | None = Field(default=None, min_length=1)
sort_order: int | None = None
agent_id: str | None = Field(default=None, max_length=64)
class UserPromptsListResponse(BaseModel):
prompts: list[UserPromptResponse]
@@ -118,3 +157,72 @@ async def delete_user_prompt(request: Request, prompt_id: str) -> None:
deleted = await store.delete(prompt_id, user_id)
if not deleted:
raise HTTPException(status_code=404, detail="Prompt not found")
# --------------------------------------------------- admin prompt templates
#
# Templates are admin-maintained prompts that carry tags (target_type
# ``user_prompt`` via /api/tags/assignments). The 岗位 (position) sync engine
# clones the templates whose tags a user's position grants into that user's own
# prompts (origin='position'). Tagging itself reuses the generic tags API.
def _require_template_store(request: Request):
store = _get_store(request)
if not hasattr(store, "list_templates"):
raise HTTPException(status_code=503, detail="Prompt templates require a database backend")
return store
@router.get("/templates", response_model=PromptTemplatesListResponse)
async def list_prompt_templates(request: Request) -> PromptTemplatesListResponse:
await _require_admin(request)
store = _require_template_store(request)
rows = await store.list_templates()
return PromptTemplatesListResponse(templates=[PromptTemplateResponse(**r) for r in rows])
@router.post("/templates", response_model=PromptTemplateResponse, status_code=201)
async def create_prompt_template(request: Request, body: PromptTemplateCreateRequest) -> PromptTemplateResponse:
await _require_admin(request)
store = _require_template_store(request)
admin_id = _current_user_id(request)
row = await store.create_template(
admin_id,
{
"id": uuid4().hex,
"title": (body.title or "").strip(),
"content": body.content,
"sort_order": body.sort_order,
"agent_id": (body.agent_id or "").strip() or None,
},
)
return PromptTemplateResponse(**row)
@router.put("/templates/{template_id}", response_model=PromptTemplateResponse)
async def update_prompt_template(request: Request, template_id: str, body: PromptTemplateUpdateRequest) -> PromptTemplateResponse:
await _require_admin(request)
store = _require_template_store(request)
data: dict = {}
if body.title is not None:
data["title"] = body.title.strip()
if body.content is not None:
data["content"] = body.content
if body.sort_order is not None:
data["sort_order"] = body.sort_order
if "agent_id" in body.model_fields_set:
data["agent_id"] = (body.agent_id or "").strip() or None
row = await store.update_template(template_id, data)
if row is None:
raise HTTPException(status_code=404, detail="Template not found")
return PromptTemplateResponse(**row)
@router.delete("/templates/{template_id}", status_code=204)
async def delete_prompt_template(request: Request, template_id: str) -> None:
await _require_admin(request)
store = _require_template_store(request)
deleted = await store.delete_template(template_id)
if not deleted:
raise HTTPException(status_code=404, detail="Template not found")
@@ -22,6 +22,8 @@ LangChain's stock `SummarizationMiddleware` runs in `before_model` and persists
- Compression runs in `wrap_model_call` / `awrap_model_call`: for a single model call, the message list is replaced with `[summary, *recent]` via `request.override(messages=...)`. **Nothing is written back to state.**
- The full original messages stay in the checkpoint, so the frontend keeps showing the user's real conversation. The summary message is named `summary` and flagged `hide_from_ui`, and because it only ever lives in the transient request it never reaches the frontend at all.
- The generated summary is cached **per-thread** so we don't pay an extra LLM call on every model invocation. The cached summary is reused (no LLM call) until enough *new* messages accumulate to push the compressed context back over the trigger threshold (a sticky boundary mirroring the stock middleware's post-compaction behavior). The cache is bounded (oldest threads evicted first).
- **Threshold-first**: while the full current conversation is still under the trigger, the request is passed through untouched — no cache lookup, no summary substitution. Compression only ever engages once the real context crosses the configured trigger.
- When the genuine (blocking) summarize path actually runs an LLM call, the middleware emits a `{"type":"context_compacting","message":"正在压缩上下文…"}` event on LangGraph's `custom` stream channel so the chat UI can show a transient "正在压缩" toast (the cheap cache-reuse path stays silent). Frontend handler: `onCustomEvent` in `frontend-web/src/core/threads/hooks.ts`.
> **Legacy threads:** conversations compacted by an older (destructive) build already lost their original messages and carry a persisted `summary` message. The frontend hides any `name: "summary"` message, so the stray summary bubble no longer shows — but those already-discarded messages cannot be recovered.
@@ -326,7 +328,7 @@ Summarization runs after ThreadData and Sandbox initialization but before Title
- Configuration is loaded once at startup
- Summarization is **non-destructive**: the summary replaces older messages only in the transient model request (`wrap_model_call`), never in persisted state. The checkpointer therefore keeps the full original conversation
- The per-thread summary cache lives in the middleware instance (in-memory, bounded). It is a cost optimization only — a cold/stale cache simply re-summarizes, which is always correct
- The per-thread summary cache lives in a **process-wide, thread-safe LRU** (`_GLOBAL_SUMMARY_CACHE`, bounded), **not** on the middleware instance. `make_lead_agent` rebuilds a fresh middleware every run, so an instance cache would be cold each turn and re-run a blocking summary LLM call on every turn of a long conversation; the module-level cache keyed by `thread_id` lets a summary survive across turns. It is a cost optimization only — a cold/stale cache simply re-summarizes, which is always correct
## Example Configurations
@@ -18,6 +18,7 @@ post-compaction behavior).
from __future__ import annotations
import logging
import threading
from collections import OrderedDict
from collections.abc import Awaitable, Callable, Collection
from dataclasses import dataclass
@@ -131,6 +132,27 @@ class _SummaryCacheEntry:
reusable: bool
# Process-wide summary cache shared across every middleware instance.
#
# ``make_lead_agent`` rebuilds the agent — and therefore a *fresh*
# ``DeerFlowSummarizationMiddleware`` — on every run. If the cache lived on the
# instance (``self._summary_cache``) it would be empty at the start of each
# user turn, forcing a brand-new (blocking) summarization LLM call on *every*
# turn of a long conversation. Hoisting it to module scope, keyed by
# ``thread_id``, lets a summary computed on one turn be reused on the next,
# so only the first threshold crossing (or a genuine re-compaction) pays for an
# LLM call. A lock guards the non-atomic LRU eviction against concurrent
# threads (subagents / parallel conversations share this process).
_GLOBAL_SUMMARY_CACHE: "OrderedDict[str, _SummaryCacheEntry]" = OrderedDict()
_GLOBAL_SUMMARY_CACHE_LOCK = threading.Lock()
def _clear_global_summary_cache() -> None:
"""Drop every cached summary. Intended for tests/diagnostics only."""
with _GLOBAL_SUMMARY_CACHE_LOCK:
_GLOBAL_SUMMARY_CACHE.clear()
class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
"""Summarization middleware with pre-compression hook dispatch and skill rescue."""
@@ -153,8 +175,9 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
self._preserve_recent_skill_tokens = max(0, preserve_recent_skill_tokens)
self._preserve_recent_skill_tokens_per_skill = max(0, preserve_recent_skill_tokens_per_skill)
# Per-thread cache of the most recent summary: thread_id -> (summarized message ids, summary text).
# Lets us reuse a summary across model calls instead of re-running the LLM every invocation.
self._summary_cache: OrderedDict[str, _SummaryCacheEntry] = OrderedDict()
# Points at the *process-wide* cache (not a fresh per-instance dict) so a
# summary survives across runs/turns — see ``_GLOBAL_SUMMARY_CACHE``.
self._summary_cache: OrderedDict[str, _SummaryCacheEntry] = _GLOBAL_SUMMARY_CACHE
@override
def before_model(self, state: AgentState, runtime: Runtime) -> dict | None:
@@ -208,6 +231,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
return payload # type: ignore[return-value]
# action == "summarize"
messages_to_summarize, preserved, thread_id = payload # type: ignore[misc]
self._emit_compacting_notice()
self._fire_hooks(messages_to_summarize, preserved, getattr(request, "runtime", None))
summary = self._create_summary(messages_to_summarize)
return self._finish_summarize(messages_to_summarize, preserved, summary, thread_id)
@@ -221,10 +245,34 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
if action == "reuse":
return payload # type: ignore[return-value]
messages_to_summarize, preserved, thread_id = payload # type: ignore[misc]
self._emit_compacting_notice()
self._fire_hooks(messages_to_summarize, preserved, getattr(request, "runtime", None))
summary = await self._acreate_summary(messages_to_summarize)
return self._finish_summarize(messages_to_summarize, preserved, summary, thread_id)
@staticmethod
def _emit_compacting_notice() -> None:
"""Best-effort UI hint that a (blocking) summarization LLM call is starting.
Only the genuine summarize path calls this — the cheap cache-reuse path
does no LLM work and stays silent. Emitted on LangGraph's ``custom``
stream channel; the chat frontend renders it as a transient toast. A
no-op when there is no active stream writer (e.g. unit tests, non-stream
runs), so callers never branch.
"""
try:
from langgraph.config import get_stream_writer
writer = get_stream_writer()
except Exception:
return
if writer is None:
return
try:
writer({"type": "context_compacting", "message": "正在压缩上下文以容纳更长对话…"})
except Exception:
logger.debug("summarization: failed to emit context_compacting notice", exc_info=True)
def _plan_compression(
self,
request: ModelRequest,
@@ -243,23 +291,30 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
if not messages:
return None
self._ensure_message_ids(messages)
thread_id = self._resolve_request_thread_id(request)
cached = self._summary_cache.get(thread_id) if thread_id else None
# Threshold is the first gate: while the *full* current conversation is
# still under the configured trigger, there is nothing to compress —
# pass it through untouched. No cache lookup, no summary substitution.
# Compression only ever engages once the real context crosses the
# trigger, which is exactly when it is needed.
if not self._should_summarize(messages, self.token_counter(messages)):
return None
thread_id = self._resolve_request_thread_id(request)
if thread_id is None:
logger.debug("summarization: thread_id unresolved; cross-turn summary cache disabled for this call")
cached = self._get_cache_entry(thread_id) if thread_id else None
present_ids = {m.id for m in messages}
if cached is not None and cached.reusable and cached.summarized_ids <= present_ids:
preserved = [m for m in messages if m.id not in cached.summarized_ids]
effective = [self._summary_message(cached.summary_text), *preserved]
if not self._should_summarize(effective, self.token_counter(effective)):
# Refresh LRU recency and reuse without an LLM call.
self._summary_cache.move_to_end(thread_id)
self._touch_cache_entry(thread_id)
return ("reuse", effective)
# New messages have pushed the compressed context back over the
# threshold — fall through and re-summarize a larger prefix.
if not self._should_summarize(messages, self.token_counter(messages)):
return None
cutoff_index = self._determine_cutoff_index(messages)
if cutoff_index <= 0:
return None
@@ -294,12 +349,24 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware):
return compressed
def _get_cache_entry(self, thread_id: str) -> _SummaryCacheEntry | None:
"""Return the cached entry for ``thread_id`` (thread-safe read)."""
with _GLOBAL_SUMMARY_CACHE_LOCK:
return self._summary_cache.get(thread_id)
def _touch_cache_entry(self, thread_id: str) -> None:
"""Refresh LRU recency for ``thread_id`` if still present (thread-safe)."""
with _GLOBAL_SUMMARY_CACHE_LOCK:
if thread_id in self._summary_cache:
self._summary_cache.move_to_end(thread_id)
def _store_cache_entry(self, thread_id: str, entry: _SummaryCacheEntry) -> None:
"""Insert/refresh a cache entry, evicting the oldest thread if over capacity."""
self._summary_cache[thread_id] = entry
self._summary_cache.move_to_end(thread_id)
while len(self._summary_cache) > _SUMMARY_CACHE_MAX_THREADS:
self._summary_cache.popitem(last=False)
with _GLOBAL_SUMMARY_CACHE_LOCK:
self._summary_cache[thread_id] = entry
self._summary_cache.move_to_end(thread_id)
while len(self._summary_cache) > _SUMMARY_CACHE_MAX_THREADS:
self._summary_cache.popitem(last=False)
@staticmethod
def _resolve_request_thread_id(request: ModelRequest) -> str | None:
@@ -32,7 +32,7 @@ class AgentRow(Base):
# ``is_private``. Empty string = "not assigned" → the listing layer treats
# it as the admin-configured default square. Each agent belongs to at most
# one square (single column), so it can never be published to two squares.
square_id: Mapped[str] = mapped_column(String(64), nullable=False, default="", server_default="", index=True)
square_id: Mapped[str] = mapped_column(String(64), nullable=False, default="", server_default="", index=True)
# NULL = not featured on the homepage selector. Non-NULL = admin-curated
# featured rank (lower sorts first). Only ``published=True`` agents are
# eligible; the router enforces that invariant on writes. BIGINT so the
@@ -57,6 +57,13 @@ class AgentFavoriteRow(Base):
id: Mapped[str] = mapped_column(String(64), primary_key=True)
user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
agent_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
# Where this favorite came from: ``"user"`` = the user bookmarked it
# themselves; ``"position"`` = it was auto-granted by the user's 岗位
# (position) via the subscription sync engine. Sync only ever adds/removes
# ``"position"`` rows, so a user's own bookmarks are never clobbered.
origin: Mapped[str] = mapped_column(String(16), nullable=False, default="user", server_default="user")
# The position that granted this favorite (only set when ``origin='position'``).
position_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
@@ -0,0 +1,108 @@
"""岗位 (position) subscription system.
New tables ``positions`` + ``position_tag_bindings``; new column
``users.position_id``; ``origin`` / ``position_id`` on ``agent_favorites`` and
``scheduled_task_subscriptions``; ``is_template`` / ``origin`` / ``template_id``
/ ``position_id`` on ``user_custom_prompts``.
Note: the running app also auto-adds these via ``_ensure_orm_columns_sync`` and
``create_all`` on startup; this migration keeps the alembic history consistent
per the repo convention (see backend/CLAUDE.md → Schema Migrations).
Revision ID: 20260616_01
Revises: 20260612_01
Create Date: 2026-06-16
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260616_01"
down_revision = "20260612_01"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
return table_name in inspect(op.get_bind()).get_table_names()
def _has_column(table_name: str, column_name: str) -> bool:
if not _has_table(table_name):
return False
return any(col["name"] == column_name for col in inspect(op.get_bind()).get_columns(table_name))
def upgrade() -> None:
if not _has_table("positions"):
op.create_table(
"positions",
sa.Column("id", sa.String(64), primary_key=True),
sa.Column("name", sa.String(191), nullable=False),
sa.Column("description", sa.String(1024), nullable=False, server_default=""),
sa.Column("is_default", sa.Boolean(), nullable=False, server_default=sa.text("0")),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
sa.UniqueConstraint("name", name="uq_positions_name"),
)
op.create_index("ix_positions_is_default", "positions", ["is_default"])
if not _has_table("position_tag_bindings"):
op.create_table(
"position_tag_bindings",
sa.Column("id", sa.String(64), primary_key=True),
sa.Column("position_id", sa.String(64), nullable=False),
sa.Column("resource_type", sa.String(32), nullable=False),
sa.Column("tag_id", sa.String(64), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
sa.UniqueConstraint("position_id", "resource_type", "tag_id", name="uq_position_tag_bindings"),
)
op.create_index("ix_position_tag_bindings_position_id", "position_tag_bindings", ["position_id"])
op.create_index("ix_position_tag_bindings_resource_type", "position_tag_bindings", ["resource_type"])
op.create_index("ix_position_tag_bindings_tag_id", "position_tag_bindings", ["tag_id"])
op.create_index("ix_position_tag_bindings_position_type", "position_tag_bindings", ["position_id", "resource_type"])
if _has_table("users") and not _has_column("users", "position_id"):
op.add_column("users", sa.Column("position_id", sa.String(64), nullable=True))
op.create_index("ix_users_position_id", "users", ["position_id"])
for table in ("agent_favorites", "scheduled_task_subscriptions"):
if _has_table(table) and not _has_column(table, "origin"):
op.add_column(table, sa.Column("origin", sa.String(16), nullable=False, server_default="user"))
if _has_table(table) and not _has_column(table, "position_id"):
op.add_column(table, sa.Column("position_id", sa.String(64), nullable=True))
op.create_index(f"ix_{table}_position_id", table, ["position_id"])
if _has_table("user_custom_prompts"):
if not _has_column("user_custom_prompts", "is_template"):
op.add_column("user_custom_prompts", sa.Column("is_template", sa.Boolean(), nullable=False, server_default=sa.text("0")))
op.create_index("ix_user_custom_prompts_is_template", "user_custom_prompts", ["is_template"])
if not _has_column("user_custom_prompts", "origin"):
op.add_column("user_custom_prompts", sa.Column("origin", sa.String(16), nullable=False, server_default="user"))
if not _has_column("user_custom_prompts", "template_id"):
op.add_column("user_custom_prompts", sa.Column("template_id", sa.String(64), nullable=True))
op.create_index("ix_user_custom_prompts_template_id", "user_custom_prompts", ["template_id"])
if not _has_column("user_custom_prompts", "position_id"):
op.add_column("user_custom_prompts", sa.Column("position_id", sa.String(64), nullable=True))
op.create_index("ix_user_custom_prompts_position_id", "user_custom_prompts", ["position_id"])
def downgrade() -> None:
for table in ("agent_favorites", "scheduled_task_subscriptions"):
if _has_column(table, "position_id"):
op.drop_column(table, "position_id")
if _has_column(table, "origin"):
op.drop_column(table, "origin")
for col in ("position_id", "template_id", "origin", "is_template"):
if _has_column("user_custom_prompts", col):
op.drop_column("user_custom_prompts", col)
if _has_column("users", "position_id"):
op.drop_column("users", "position_id")
if _has_table("position_tag_bindings"):
op.drop_table("position_tag_bindings")
if _has_table("positions"):
op.drop_table("positions")
@@ -41,6 +41,7 @@ from deerflow.persistence.llm_metrics.model import LlmCallMetricRow
from deerflow.persistence.models.run_event import RunEventRow
from deerflow.persistence.notifications.model import NotificationRow
from deerflow.persistence.parallel_agent_tasks.model import ParallelAgentTaskRow
from deerflow.persistence.positions.model import PositionRow, PositionTagBindingRow
from deerflow.persistence.recommended_questions.model import RecommendedQuestionRow
from deerflow.persistence.roundtable_chains.model import RoundtableChainRow
from deerflow.persistence.roundtable_draft_shares.model import RoundtableDraftShareRow
@@ -88,6 +89,8 @@ __all__ = [
"LlmCallMetricRow",
"NotificationRow",
"ParallelAgentTaskRow",
"PositionRow",
"PositionTagBindingRow",
"RecommendedQuestionRow",
"RoundtableChainRow",
"RoundtableDraftRow",
@@ -0,0 +1,26 @@
"""Persistence layer for 岗位 (positions) and per-resource tag bindings."""
from deerflow.persistence.positions.base import PositionStore
from deerflow.persistence.positions.memory import MemoryPositionStore
from deerflow.persistence.positions.model import (
POSITION_RESOURCE_TYPES,
PositionRow,
PositionTagBindingRow,
)
from deerflow.persistence.positions.sql import PositionRepository
__all__ = [
"PositionStore",
"MemoryPositionStore",
"PositionRepository",
"PositionRow",
"PositionTagBindingRow",
"POSITION_RESOURCE_TYPES",
"make_position_store",
]
def make_position_store(session_factory):
if session_factory is None:
return MemoryPositionStore()
return PositionRepository(session_factory)
@@ -0,0 +1,112 @@
"""Abstract persistence API for 岗位 (positions) and their tag bindings."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
class PositionStore(ABC):
# -------------------------------------------------------------- positions
@abstractmethod
async def list_positions(self) -> list[dict[str, Any]]:
"""List all positions. Each record carries a derived ``member_count``."""
raise NotImplementedError
@abstractmethod
async def get_position(self, position_id: str) -> dict[str, Any] | None:
raise NotImplementedError
@abstractmethod
async def create_position(self, data: dict[str, Any]) -> dict[str, Any]:
"""Create a position. ``data`` must carry ``id`` and ``name``.
Raises ``ValueError`` if ``name`` is not unique. When ``is_default`` is
truthy, any other default position is cleared first.
"""
raise NotImplementedError
@abstractmethod
async def update_position(self, position_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
"""Update ``name`` / ``description`` / ``is_default``.
Returns ``None`` if the position does not exist. Raises ``ValueError``
on a name collision. Setting ``is_default=True`` clears the flag on the
other positions.
"""
raise NotImplementedError
@abstractmethod
async def delete_position(self, position_id: str) -> bool:
"""Delete a position, its tag bindings, and unassign its members."""
raise NotImplementedError
@abstractmethod
async def get_default_position_id(self) -> str | None:
"""Return the id of the ``is_default`` position, or ``None``."""
raise NotImplementedError
# ------------------------------------------------------------ tag bindings
@abstractmethod
async def get_tag_bindings(self, position_id: str) -> dict[str, list[str]]:
"""Return ``{resource_type: [tag_id, ...]}`` for one position.
Resource types with no bound tags are omitted (callers treat a missing
key as "use the default set").
"""
raise NotImplementedError
@abstractmethod
async def set_tag_bindings(self, position_id: str, bindings: dict[str, list[str]]) -> dict[str, list[str]]:
"""Replace the full set of tag bindings for a position.
``bindings`` maps each resource_type to its desired tag-id list. A
resource_type omitted from ``bindings`` (or mapped to ``[]``) ends up
with no rows — i.e. it falls back to the default set.
"""
raise NotImplementedError
# --------------------------------------------------------------- members
@abstractmethod
async def get_user_position_id(self, user_id: str) -> str | None:
"""Return the position id explicitly assigned to a user, or ``None``."""
raise NotImplementedError
@abstractmethod
async def list_members(self, position_id: str) -> list[dict[str, Any]]:
"""Return ``[{id, email}, ...]`` of users assigned to this position."""
raise NotImplementedError
@abstractmethod
async def assign_users(self, position_id: str, user_ids: list[str]) -> int:
"""Assign a position to many users (1:1 — overwrites any prior position).
Returns the number of users updated.
"""
raise NotImplementedError
@abstractmethod
async def unassign_users(self, user_ids: list[str]) -> int:
"""Clear the position of many users. Returns the number updated."""
raise NotImplementedError
@abstractmethod
async def list_member_ids(self, position_id: str) -> list[str]:
"""Return just the user ids assigned to this position (sync fan-out)."""
raise NotImplementedError
@abstractmethod
async def list_assigned_user_ids(self) -> list[str]:
"""Return every user id that has an explicit position (for bulk resync)."""
raise NotImplementedError
@abstractmethod
async def list_all_users(self) -> list[dict[str, Any]]:
"""Return ``[{id, email, system_role, position_id}, ...]`` for every user.
Powers the admin member-picker and the per-user position editor.
"""
raise NotImplementedError
@@ -0,0 +1,143 @@
"""In-memory ``PositionStore`` implementation for tests."""
from __future__ import annotations
import copy
from datetime import UTC, datetime
from typing import Any
def _now() -> str:
return datetime.now(UTC).isoformat()
class MemoryPositionStore:
def __init__(self) -> None:
self._positions: dict[str, dict[str, Any]] = {}
# position_id -> {resource_type -> [tag_id]}
self._bindings: dict[str, dict[str, list[str]]] = {}
# user_id -> position_id
self._user_position: dict[str, str] = {}
# user_id -> email (optional; populated via register_user for nicer member lists)
self._user_email: dict[str, str] = {}
# test helper — not part of the abstract API
def register_user(self, user_id: str, email: str = "") -> None:
self._user_email[user_id] = email
def _member_count(self, position_id: str) -> int:
return sum(1 for pid in self._user_position.values() if pid == position_id)
def _to_dict(self, record: dict[str, Any]) -> dict[str, Any]:
out = copy.deepcopy(record)
out["member_count"] = self._member_count(record["id"])
return out
# -------------------------------------------------------------- positions
async def list_positions(self) -> list[dict[str, Any]]:
rows = sorted(self._positions.values(), key=lambda r: r["created_at"])
return [self._to_dict(r) for r in rows]
async def get_position(self, position_id: str) -> dict[str, Any] | None:
record = self._positions.get(position_id)
return self._to_dict(record) if record is not None else None
def _clear_default(self, keep_id: str | None) -> None:
for r in self._positions.values():
if r["id"] != keep_id:
r["is_default"] = False
async def create_position(self, data: dict[str, Any]) -> dict[str, Any]:
name = data["name"]
if any(r["name"] == name for r in self._positions.values()):
raise ValueError(f"Position '{name}' already exists")
is_default = bool(data.get("is_default"))
if is_default:
self._clear_default(keep_id=data["id"])
record = {
"id": data["id"],
"name": name,
"description": data.get("description") or "",
"is_default": is_default,
"created_at": _now(),
"updated_at": _now(),
}
self._positions[data["id"]] = record
return self._to_dict(record)
async def update_position(self, position_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
record = self._positions.get(position_id)
if record is None:
return None
new_name = data.get("name", record["name"])
if new_name is not None and any(r["id"] != position_id and r["name"] == new_name for r in self._positions.values()):
raise ValueError(f"Position '{new_name}' already exists")
if "name" in data and data["name"] is not None:
record["name"] = data["name"]
if "description" in data and data["description"] is not None:
record["description"] = data["description"]
if "is_default" in data and data["is_default"] is not None:
record["is_default"] = bool(data["is_default"])
if record["is_default"]:
self._clear_default(keep_id=position_id)
record["updated_at"] = _now()
return self._to_dict(record)
async def delete_position(self, position_id: str) -> bool:
if position_id not in self._positions:
return False
del self._positions[position_id]
self._bindings.pop(position_id, None)
for uid in [uid for uid, pid in self._user_position.items() if pid == position_id]:
del self._user_position[uid]
return True
async def get_default_position_id(self) -> str | None:
for r in self._positions.values():
if r["is_default"]:
return r["id"]
return None
# ------------------------------------------------------------ tag bindings
async def get_tag_bindings(self, position_id: str) -> dict[str, list[str]]:
return copy.deepcopy(self._bindings.get(position_id, {}))
async def set_tag_bindings(self, position_id: str, bindings: dict[str, list[str]]) -> dict[str, list[str]]:
cleaned = {rt: list(dict.fromkeys(tag_ids)) for rt, tag_ids in bindings.items() if tag_ids}
self._bindings[position_id] = cleaned
return copy.deepcopy(cleaned)
# --------------------------------------------------------------- members
async def get_user_position_id(self, user_id: str) -> str | None:
return self._user_position.get(user_id)
async def list_members(self, position_id: str) -> list[dict[str, Any]]:
return [{"id": uid, "email": self._user_email.get(uid, "")} for uid, pid in self._user_position.items() if pid == position_id]
async def assign_users(self, position_id: str, user_ids: list[str]) -> int:
for uid in user_ids:
self._user_position[uid] = position_id
return len(user_ids)
async def unassign_users(self, user_ids: list[str]) -> int:
count = 0
for uid in user_ids:
if uid in self._user_position:
del self._user_position[uid]
count += 1
return count
async def list_member_ids(self, position_id: str) -> list[str]:
return [uid for uid, pid in self._user_position.items() if pid == position_id]
async def list_assigned_user_ids(self) -> list[str]:
return list(self._user_position.keys())
async def list_all_users(self) -> list[dict[str, Any]]:
return [
{"id": uid, "email": email, "system_role": "user", "position_id": self._user_position.get(uid)}
for uid, email in sorted(self._user_email.items(), key=lambda kv: kv[1])
]
@@ -0,0 +1,73 @@
"""ORM models for 岗位 (positions) and their per-resource tag bindings.
A **position** (岗位) is an admin-managed aggregation layer that decides which
agents / skills / scheduled tasks / custom prompts / recommended questions a
user is subscribed to. Each user has at most one position (the link lives on
``users.position_id``, a 1:1 relation — see ``UserRow``).
Two tables here:
- ``positions`` — the position itself. ``is_default`` marks the fallback
position inherited by users who have not been assigned one explicitly.
- ``position_tag_bindings`` — for each ``(position, resource_type)`` the set of
tags whose resources are granted to the position's members. **No binding for
a resource_type means "use the default set"** (see ``deerflow.services.
position_sync`` for what "default" resolves to per type).
"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import Boolean, Index, String, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
# Resource types a position can grant. The first three reuse the existing
# typed-tag namespace (``deerflow.persistence.tags``); ``user_prompt`` and
# ``recommended_question`` are added by the same change (see TAG_TYPES).
POSITION_RESOURCE_TYPES: tuple[str, ...] = (
"agent",
"skill",
"scheduled_task",
"user_prompt",
"recommended_question",
)
class PositionRow(Base):
__tablename__ = "positions"
__table_args__ = (UniqueConstraint("name", name="uq_positions_name"),)
id: Mapped[str] = mapped_column(String(64), primary_key=True)
name: Mapped[str] = mapped_column(String(191), nullable=False)
description: Mapped[str] = mapped_column(String(1024), nullable=False, default="")
# Exactly one position should carry ``is_default=True`` at a time (the
# router enforces "only one default" by clearing the flag on others). Users
# with no explicit ``users.position_id`` inherit this position's grants.
is_default: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
class PositionTagBindingRow(Base):
"""One bound tag for a ``(position, resource_type)`` pair.
The absence of *any* row for a ``(position, resource_type)`` means the
position uses the **default set** for that resource type. One or more rows
mean the position grants the *union* of the resources carrying those tags.
"""
__tablename__ = "position_tag_bindings"
__table_args__ = (
UniqueConstraint("position_id", "resource_type", "tag_id", name="uq_position_tag_bindings"),
Index("ix_position_tag_bindings_position_type", "position_id", "resource_type"),
)
id: Mapped[str] = mapped_column(String(64), primary_key=True)
position_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
resource_type: Mapped[str] = mapped_column(String(32), nullable=False, index=True)
tag_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
@@ -0,0 +1,196 @@
"""SQLAlchemy-backed position + tag-binding repository."""
from __future__ import annotations
import uuid
from datetime import UTC, datetime
from typing import Any
from sqlalchemy import delete, func, select, update
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.positions.base import PositionStore
from deerflow.persistence.positions.model import PositionRow, PositionTagBindingRow
from deerflow.persistence.user.model import UserRow
def _position_to_dict(row: PositionRow, member_count: int = 0) -> dict[str, Any]:
data = row.to_dict()
for key in ("created_at", "updated_at"):
value = data.get(key)
if isinstance(value, datetime):
data[key] = value.isoformat()
data["member_count"] = member_count
return data
class PositionRepository(PositionStore):
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
@staticmethod
async def _member_counts(session: AsyncSession, position_ids: list[str]) -> dict[str, int]:
if not position_ids:
return {}
stmt = select(UserRow.position_id, func.count()).where(UserRow.position_id.in_(position_ids)).group_by(UserRow.position_id)
result = await session.execute(stmt)
return {pid: count for pid, count in result.all() if pid is not None}
@staticmethod
async def _clear_default(session: AsyncSession, keep_id: str | None) -> None:
stmt = update(PositionRow).where(PositionRow.is_default.is_(True))
if keep_id is not None:
stmt = stmt.where(PositionRow.id != keep_id)
await session.execute(stmt.values(is_default=False))
# -------------------------------------------------------------- positions
async def list_positions(self) -> list[dict[str, Any]]:
async with self._sf() as session:
rows = list((await session.execute(select(PositionRow).order_by(PositionRow.created_at.asc()))).scalars())
counts = await self._member_counts(session, [r.id for r in rows])
return [_position_to_dict(r, counts.get(r.id, 0)) for r in rows]
async def get_position(self, position_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(PositionRow, position_id)
if row is None:
return None
counts = await self._member_counts(session, [position_id])
return _position_to_dict(row, counts.get(position_id, 0))
async def create_position(self, data: dict[str, Any]) -> dict[str, Any]:
now = datetime.now(UTC)
is_default = bool(data.get("is_default"))
row = PositionRow(
id=data["id"],
name=data["name"],
description=data.get("description") or "",
is_default=is_default,
created_at=now,
updated_at=now,
)
async with self._sf() as session:
if is_default:
await self._clear_default(session, keep_id=row.id)
session.add(row)
try:
await session.commit()
except IntegrityError as exc:
await session.rollback()
raise ValueError(f"Position '{data['name']}' already exists") from exc
await session.refresh(row)
return _position_to_dict(row, 0)
async def update_position(self, position_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(PositionRow, position_id)
if row is None:
return None
if "name" in data and data["name"] is not None:
row.name = data["name"]
if "description" in data and data["description"] is not None:
row.description = data["description"]
if "is_default" in data and data["is_default"] is not None:
row.is_default = bool(data["is_default"])
if row.is_default:
await self._clear_default(session, keep_id=position_id)
row.updated_at = datetime.now(UTC)
try:
await session.commit()
except IntegrityError as exc:
await session.rollback()
raise ValueError(f"Position '{data.get('name')}' already exists") from exc
await session.refresh(row)
counts = await self._member_counts(session, [position_id])
return _position_to_dict(row, counts.get(position_id, 0))
async def delete_position(self, position_id: str) -> bool:
async with self._sf() as session:
row = await session.get(PositionRow, position_id)
if row is None:
return False
await session.execute(delete(PositionTagBindingRow).where(PositionTagBindingRow.position_id == position_id))
await session.execute(update(UserRow).where(UserRow.position_id == position_id).values(position_id=None))
await session.delete(row)
await session.commit()
return True
async def get_default_position_id(self) -> str | None:
async with self._sf() as session:
return (await session.execute(select(PositionRow.id).where(PositionRow.is_default.is_(True)).limit(1))).scalar_one_or_none()
# ------------------------------------------------------------ tag bindings
async def get_tag_bindings(self, position_id: str) -> dict[str, list[str]]:
async with self._sf() as session:
rows = list(
(
await session.execute(
select(PositionTagBindingRow).where(PositionTagBindingRow.position_id == position_id).order_by(PositionTagBindingRow.created_at.asc())
)
).scalars()
)
out: dict[str, list[str]] = {}
for r in rows:
out.setdefault(r.resource_type, []).append(r.tag_id)
return out
async def set_tag_bindings(self, position_id: str, bindings: dict[str, list[str]]) -> dict[str, list[str]]:
now = datetime.now(UTC)
async with self._sf() as session:
await session.execute(delete(PositionTagBindingRow).where(PositionTagBindingRow.position_id == position_id))
for resource_type, tag_ids in bindings.items():
for tag_id in dict.fromkeys(tag_ids):
session.add(
PositionTagBindingRow(
id=uuid.uuid4().hex,
position_id=position_id,
resource_type=resource_type,
tag_id=tag_id,
created_at=now,
)
)
await session.commit()
return await self.get_tag_bindings(position_id)
# --------------------------------------------------------------- members
async def get_user_position_id(self, user_id: str) -> str | None:
async with self._sf() as session:
return (await session.execute(select(UserRow.position_id).where(UserRow.id == user_id))).scalar_one_or_none()
async def list_members(self, position_id: str) -> list[dict[str, Any]]:
async with self._sf() as session:
rows = (await session.execute(select(UserRow.id, UserRow.email).where(UserRow.position_id == position_id).order_by(UserRow.email.asc()))).all()
return [{"id": uid, "email": email} for uid, email in rows]
async def assign_users(self, position_id: str, user_ids: list[str]) -> int:
if not user_ids:
return 0
async with self._sf() as session:
result = await session.execute(update(UserRow).where(UserRow.id.in_(user_ids)).values(position_id=position_id))
await session.commit()
return result.rowcount or 0
async def unassign_users(self, user_ids: list[str]) -> int:
if not user_ids:
return 0
async with self._sf() as session:
result = await session.execute(update(UserRow).where(UserRow.id.in_(user_ids)).values(position_id=None))
await session.commit()
return result.rowcount or 0
async def list_member_ids(self, position_id: str) -> list[str]:
async with self._sf() as session:
return [uid for (uid,) in (await session.execute(select(UserRow.id).where(UserRow.position_id == position_id))).all()]
async def list_assigned_user_ids(self) -> list[str]:
async with self._sf() as session:
return [uid for (uid,) in (await session.execute(select(UserRow.id).where(UserRow.position_id.isnot(None)))).all()]
async def list_all_users(self) -> list[dict[str, Any]]:
async with self._sf() as session:
rows = (await session.execute(select(UserRow.id, UserRow.email, UserRow.system_role, UserRow.position_id).order_by(UserRow.email.asc()))).all()
return [{"id": uid, "email": email, "system_role": role, "position_id": pid} for uid, email, role, pid in rows]
@@ -55,6 +55,11 @@ class ScheduledTaskSubscriptionRow(Base):
task_id: Mapped[str] = mapped_column(String(64), index=True)
user_id: Mapped[str] = mapped_column(String(64), index=True)
notify_element: Mapped[bool] = mapped_column(Boolean, default=False)
# Source of this subscription: ``"user"`` (subscribed manually from the
# 广场) or ``"position"`` (auto-granted by the user's 岗位 via the sync
# engine). Sync only ever touches ``"position"`` rows.
origin: Mapped[str] = mapped_column(String(16), nullable=False, default="user", server_default="user")
position_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -71,3 +71,14 @@ class TagStore(ABC):
async def unassign_all(self, target_type: str, target_id: str) -> int:
"""Drop every tag assignment for a resource. Returns the count removed."""
raise NotImplementedError
@abstractmethod
async def list_targets_for_tags(self, target_type: str, tag_ids: list[str]) -> list[str]:
"""Return the ids of ``target_type`` resources carrying ANY of ``tag_ids`` (OR).
Reverse of :meth:`list_assignments`. Powers the 岗位 (position) sync
engine and the per-position visibility filters: "which resources does
this set of tags grant?". Returns a de-duplicated list; empty when
``tag_ids`` is empty.
"""
raise NotImplementedError
@@ -136,3 +136,15 @@ class MemoryTagStore(TagStore):
for aid in victims:
del self._assignments[aid]
return len(victims)
async def list_targets_for_tags(self, target_type: str, tag_ids: list[str]) -> list[str]:
wanted = set(tag_ids)
if not wanted:
return []
out: list[str] = []
seen: set[str] = set()
for a in self._assignments.values():
if a["target_type"] == target_type and a["tag_id"] in wanted and a["target_id"] not in seen:
seen.add(a["target_id"])
out.append(a["target_id"])
return out
@@ -21,7 +21,9 @@ from deerflow.persistence.base import Base
from deerflow.persistence.types import BeijingDateTime
# Allowed tag/target types. Kept here so persistence + API share one source.
TAG_TYPES: tuple[str, ...] = ("agent", "skill", "scheduled_task")
# ``user_prompt`` / ``recommended_question`` were added for the 岗位 (position)
# subscription system so those resources can be tag-targeted like agents/skills.
TAG_TYPES: tuple[str, ...] = ("agent", "skill", "scheduled_task", "user_prompt", "recommended_question")
class TagRow(Base):
@@ -185,3 +185,15 @@ class TagRepository(TagStore):
)
await session.commit()
return result.rowcount or 0
async def list_targets_for_tags(self, target_type: str, tag_ids: list[str]) -> list[str]:
unique_ids = list(dict.fromkeys(tag_ids))
if not unique_ids:
return []
stmt = (
select(TagAssignmentRow.target_id)
.where(TagAssignmentRow.target_type == target_type, TagAssignmentRow.tag_id.in_(unique_ids))
.distinct()
)
async with self._sf() as session:
return [tid for (tid,) in (await session.execute(stmt)).all()]
@@ -49,6 +49,11 @@ class UserRow(Base):
needs_setup: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
token_version: Mapped[int] = mapped_column(nullable=False, default=0)
# 岗位 (position) the user is assigned to (1:1). NULL = no explicit
# position → the user inherits the ``is_default`` position's grants. Drives
# the subscription sync engine (see ``deerflow.services.position_sync``).
position_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
__table_args__ = (
Index(
"idx_users_oauth_identity",
@@ -4,7 +4,7 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import Integer, String, Text
from sqlalchemy import Boolean, Integer, String, Text
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
@@ -22,5 +22,18 @@ class UserCustomPromptRow(Base):
# Optional association to a single agent. NULL = global (shown everywhere);
# otherwise the prompt only surfaces when that agent is selected.
agent_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
# When True, this row is an admin-maintained **template** (not a user's own
# prompt). Templates are tag-targeted (via TagAssignmentRow,
# target_type='user_prompt') and cloned into a user's prompts by the 岗位
# sync engine. ``user_id`` of a template holds the admin's id (audit only).
is_template: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default="0", index=True)
# Source of a non-template row: ``"user"`` (created by the user) or
# ``"position"`` (cloned from a template by the sync engine). Sync only
# touches ``"position"`` rows.
origin: Mapped[str] = mapped_column(String(16), nullable=False, default="user", server_default="user")
# For ``origin='position'`` clones: the template this row was cloned from
# (so re-sync can reconcile by template id) and the granting position.
template_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
position_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
created_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(BeijingDateTime(), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -25,15 +25,76 @@ class UserCustomPromptRepository:
self._sf = session_factory
async def list_for_user(self, user_id: str) -> list[dict[str, Any]]:
# Exclude admin templates (``is_template=True``) — those are not a
# user's own prompts; position-cloned rows (``is_template=False``,
# ``origin='position'``) DO show, so the user sees their 岗位 prompts.
stmt = (
select(UserCustomPromptRow)
.where(UserCustomPromptRow.user_id == user_id)
.where(UserCustomPromptRow.user_id == user_id, UserCustomPromptRow.is_template.is_(False))
.order_by(UserCustomPromptRow.sort_order.asc(), UserCustomPromptRow.created_at.asc())
)
async with self._sf() as session:
result = await session.execute(stmt)
return [_row_to_dict(row) for row in result.scalars()]
# ----------------------------------------------------- admin templates
async def list_templates(self) -> list[dict[str, Any]]:
stmt = select(UserCustomPromptRow).where(UserCustomPromptRow.is_template.is_(True)).order_by(UserCustomPromptRow.sort_order.asc(), UserCustomPromptRow.created_at.asc())
async with self._sf() as session:
return [_row_to_dict(row) for row in (await session.execute(stmt)).scalars()]
async def get_template(self, template_id: str) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(UserCustomPromptRow, template_id)
if row is None or not row.is_template:
return None
return _row_to_dict(row)
async def create_template(self, admin_user_id: str, data: dict[str, Any]) -> dict[str, Any]:
now = datetime.now(UTC)
row = UserCustomPromptRow(
id=data["id"],
user_id=admin_user_id,
title=data.get("title") or "",
content=data["content"],
sort_order=data.get("sort_order", 0),
agent_id=(data.get("agent_id") or None),
is_template=True,
origin="user",
created_at=now,
updated_at=now,
)
async with self._sf() as session:
session.add(row)
await session.commit()
await session.refresh(row)
return _row_to_dict(row)
async def update_template(self, template_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(UserCustomPromptRow, template_id)
if row is None or not row.is_template:
return None
for key in ("title", "content", "sort_order"):
if key in data:
setattr(row, key, data[key])
if "agent_id" in data:
row.agent_id = data["agent_id"] or None
row.updated_at = datetime.now(UTC)
await session.commit()
await session.refresh(row)
return _row_to_dict(row)
async def delete_template(self, template_id: str) -> bool:
async with self._sf() as session:
row = await session.get(UserCustomPromptRow, template_id)
if row is None or not row.is_template:
return False
await session.delete(row)
await session.commit()
return True
async def create(self, user_id: str, data: dict[str, Any]) -> dict[str, Any]:
now = datetime.now(UTC)
row = UserCustomPromptRow(
@@ -0,0 +1 @@
"""Cross-cutting harness services (no ``app.*`` imports)."""
@@ -0,0 +1,220 @@
"""岗位 (position) → resource subscription **sync engine**.
A user's position decides which agents / scheduled tasks / custom prompts they
are subscribed to, and which skills / recommended questions they can see. This
service turns a position's *tag bindings* into concrete, materialized rows in
the existing "我的" tables (idempotent, re-runnable).
Resolution rule (per resource type):
* The user's position is ``users.position_id``; if unset, the ``is_default``
position is used; if there is none, every type falls back to its default set.
* For a resource type **with** bound tags → the union of resources carrying any
of those tags.
* For a resource type **without** bound tags → the *default set*:
- ``agent`` → all built-in agents (``user_id IS NULL``)
- ``scheduled_task`` → empty (no built-in tasks)
- ``user_prompt`` → admin templates that carry no tag
- ``skill`` / ``recommended_question`` → *unrestricted* (handled at query
time, never materialized — see :meth:`resolve_visible`)
Materialization (``sync_user``) only ever touches rows it created itself
(``origin='position'``), so a user's own favorites / subscriptions / prompts are
never clobbered. The skill + recommended-question visibility filters are
computed live by :meth:`resolve_visible` and are *not* materialized.
"""
from __future__ import annotations
import logging
import uuid
from datetime import UTC, datetime
from sqlalchemy import delete, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.agents.model import AgentFavoriteRow, AgentRow
from deerflow.persistence.positions.base import PositionStore
from deerflow.persistence.scheduled_tasks.model import ScheduledTaskSubscriptionRow
from deerflow.persistence.tags.base import TagStore
from deerflow.persistence.user_prompts.model import UserCustomPromptRow
logger = logging.getLogger(__name__)
# Materialized into "我的" tables by sync_user.
_MATERIALIZED_TYPES = ("agent", "scheduled_task", "user_prompt")
# Resolved live at query time (visibility filters), never materialized.
_VISIBILITY_TYPES = ("skill", "recommended_question")
class PositionSyncService:
def __init__(self, session_factory: async_sessionmaker[AsyncSession], position_store: PositionStore, tag_store: TagStore) -> None:
self._sf = session_factory
self._positions = position_store
self._tags = tag_store
# ----------------------------------------------------------- resolution
async def _effective_position_id(self, user_id: str) -> str | None:
pid = await self._positions.get_user_position_id(user_id)
if pid:
return pid
return await self._positions.get_default_position_id()
async def resolve_visible(self, user_id: str, resource_type: str) -> set[str] | None:
"""Return the set of visible ``resource_type`` ids, or ``None`` = unrestricted.
Used by the skills list + recommended-questions filters. ``None`` means
"no position restriction → show everything"; a set means "restrict to
exactly these ids".
"""
pid = await self._effective_position_id(user_id)
if not pid:
return None
bindings = await self._positions.get_tag_bindings(pid)
tag_ids = bindings.get(resource_type) or []
if not tag_ids:
return None
return set(await self._tags.list_targets_for_tags(resource_type, tag_ids))
async def _resolve_materialized(self, user_id: str, resource_type: str, session: AsyncSession) -> set[str]:
pid = await self._effective_position_id(user_id)
bindings = await self._positions.get_tag_bindings(pid) if pid else {}
tag_ids = bindings.get(resource_type) or []
if tag_ids:
return set(await self._tags.list_targets_for_tags(resource_type, tag_ids))
return await self._default_set(resource_type, session)
async def _default_set(self, resource_type: str, session: AsyncSession) -> set[str]:
if resource_type == "agent":
rows = (await session.execute(select(AgentRow.id).where(AgentRow.user_id.is_(None)))).all()
return {r[0] for r in rows}
if resource_type == "scheduled_task":
return set()
if resource_type == "user_prompt":
# Admin templates that carry no tag = global default templates.
template_ids = {r[0] for r in (await session.execute(select(UserCustomPromptRow.id).where(UserCustomPromptRow.is_template.is_(True)))).all()}
if not template_ids:
return set()
tagged = set(await self._tags.list_targets_for_tags("user_prompt", await self._all_user_prompt_tag_ids()))
return template_ids - tagged
return set()
async def _all_user_prompt_tag_ids(self) -> list[str]:
return [t["id"] for t in await self._tags.list_tags("user_prompt")]
# --------------------------------------------------------- materialize
async def sync_user(self, user_id: str) -> None:
"""Reconcile all materialized subscriptions for one user (idempotent)."""
if self._sf is None:
return
pid = await self._effective_position_id(user_id)
async with self._sf() as session:
await self._sync_agents(session, user_id, pid)
await self._sync_tasks(session, user_id, pid)
await self._sync_prompts(session, user_id, pid)
await session.commit()
async def _sync_agents(self, session: AsyncSession, user_id: str, position_id: str | None) -> None:
target = await self._resolve_materialized(user_id, "agent", session)
rows = list((await session.execute(select(AgentFavoriteRow).where(AgentFavoriteRow.user_id == user_id))).scalars())
owned_any = {r.agent_id for r in rows}
position_rows = {r.agent_id: r for r in rows if r.origin == "position"}
now = datetime.now(UTC)
for agent_id in target - owned_any:
session.add(AgentFavoriteRow(id=uuid.uuid4().hex, user_id=user_id, agent_id=agent_id, origin="position", position_id=position_id, created_at=now))
for agent_id, row in position_rows.items():
if agent_id not in target:
await session.delete(row)
async def _sync_tasks(self, session: AsyncSession, user_id: str, position_id: str | None) -> None:
target = await self._resolve_materialized(user_id, "scheduled_task", session)
rows = list((await session.execute(select(ScheduledTaskSubscriptionRow).where(ScheduledTaskSubscriptionRow.user_id == user_id))).scalars())
owned_any = {r.task_id for r in rows}
position_rows = {r.task_id: r for r in rows if r.origin == "position"}
now = datetime.now(UTC)
for task_id in target - owned_any:
session.add(ScheduledTaskSubscriptionRow(id=uuid.uuid4().hex, task_id=task_id, user_id=user_id, notify_element=False, origin="position", position_id=position_id, created_at=now, updated_at=now))
for task_id, row in position_rows.items():
if task_id not in target:
await session.delete(row)
async def _sync_prompts(self, session: AsyncSession, user_id: str, position_id: str | None) -> None:
target_template_ids = await self._resolve_materialized(user_id, "user_prompt", session)
# Load the templates we need to clone, so we can copy their content.
templates: dict[str, UserCustomPromptRow] = {}
if target_template_ids:
for row in (await session.execute(select(UserCustomPromptRow).where(UserCustomPromptRow.id.in_(target_template_ids), UserCustomPromptRow.is_template.is_(True)))).scalars():
templates[row.id] = row
existing = list(
(await session.execute(select(UserCustomPromptRow).where(UserCustomPromptRow.user_id == user_id, UserCustomPromptRow.origin == "position"))).scalars()
)
existing_by_template = {r.template_id: r for r in existing if r.template_id}
now = datetime.now(UTC)
# Remove stale clones (template no longer granted or template deleted).
for tpl_id, row in existing_by_template.items():
if tpl_id not in templates:
await session.delete(row)
# Add / refresh clones for granted templates.
for tpl_id, tpl in templates.items():
row = existing_by_template.get(tpl_id)
if row is None:
session.add(
UserCustomPromptRow(
id=uuid.uuid4().hex,
user_id=user_id,
title=tpl.title,
content=tpl.content,
sort_order=tpl.sort_order,
agent_id=tpl.agent_id,
is_template=False,
origin="position",
template_id=tpl_id,
position_id=position_id,
created_at=now,
updated_at=now,
)
)
else:
# Keep the user's clone in step with the template content.
row.title = tpl.title
row.content = tpl.content
row.agent_id = tpl.agent_id
row.position_id = position_id
row.updated_at = now
# ------------------------------------------------------------- fan-out
async def sync_position(self, position_id: str) -> int:
"""Re-sync every member of a position. Returns the member count."""
member_ids = await self._positions.list_member_ids(position_id)
for uid in member_ids:
try:
await self.sync_user(uid)
except Exception:
logger.exception("position sync failed for user %s (position %s)", uid, position_id)
return len(member_ids)
async def sync_default_members(self) -> int:
"""Re-sync users with NO explicit position (they inherit the default)."""
default_id = await self._positions.get_default_position_id()
if not default_id:
return 0
assigned = set(await self._positions.list_assigned_user_ids())
count = 0
if self._sf is None:
return 0
from deerflow.persistence.user.model import UserRow
async with self._sf() as session:
all_ids = [uid for (uid,) in (await session.execute(select(UserRow.id))).all()]
for uid in all_ids:
if uid in assigned:
continue
try:
await self.sync_user(uid)
count += 1
except Exception:
logger.exception("default position sync failed for user %s", uid)
return count
@@ -82,6 +82,23 @@ CALL _add_column_if_missing('roundtable_drafts', 'task_id', 'VARCHAR(128) NULL D
-- ORM: RoundtableJobRow.orchestration_plan = PortableLongText(LONGTEXT), nullable。
CALL _add_column_if_missing('roundtable_jobs', 'orchestration_plan', 'LONGTEXT NULL DEFAULT NULL');
-- ---- 岗位 (positions) 订阅系统,对应 migration 20260616_01 -------------------
-- positions / position_tag_bindings 是【新表】,由 create_all 自动建,无需补列。
-- 以下是给【已存在的表】补的列:
-- users.position_id: 用户所属岗位(1:1)。ORM: UserRow.position_id = String(64) NULL。
CALL _add_column_if_missing('users', 'position_id', 'VARCHAR(64) NULL DEFAULT NULL');
-- agent_favorites: origin 区分用户自收藏 vs 岗位授予;position_id 记授予岗位。
CALL _add_column_if_missing('agent_favorites', 'origin', "VARCHAR(16) NOT NULL DEFAULT 'user'");
CALL _add_column_if_missing('agent_favorites', 'position_id', 'VARCHAR(64) NULL DEFAULT NULL');
-- scheduled_task_subscriptions: 同上。
CALL _add_column_if_missing('scheduled_task_subscriptions', 'origin', "VARCHAR(16) NOT NULL DEFAULT 'user'");
CALL _add_column_if_missing('scheduled_task_subscriptions', 'position_id', 'VARCHAR(64) NULL DEFAULT NULL');
-- user_custom_prompts: is_template 标记管理员模板;origin/template_id/position_id 记岗位克隆来源。
CALL _add_column_if_missing('user_custom_prompts', 'is_template', 'TINYINT(1) NOT NULL DEFAULT 0');
CALL _add_column_if_missing('user_custom_prompts', 'origin', "VARCHAR(16) NOT NULL DEFAULT 'user'");
CALL _add_column_if_missing('user_custom_prompts', 'template_id', 'VARCHAR(64) NULL DEFAULT NULL');
CALL _add_column_if_missing('user_custom_prompts', 'position_id', 'VARCHAR(64) NULL DEFAULT NULL');
-- ---- 清理临时存储过程 --------------------------------------------------------
DROP PROCEDURE IF EXISTS _add_column_if_missing;