feat: 定时任务管理 + 会话只读分享

定时任务:
- 管理员可查看/编辑/删除/触发全部用户的定时任务(新增 /api/scheduled-tasks/admin/* 接口与管理页)
- 卡死任务恢复:单次执行超 30 分钟超时重试(最多 3 次),耗尽后标记 failed;看门狗回收服务重启遗留的「执行中」任务

会话分享:
- 新增「只读链接」分享模式,生成公开免登录的只读对话快照页(/share/:code)
- 公开接口 /api/public/shares/* 提供对话快照与图片附件

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
吉鑫升andClaude Opus 4.7 committed 2026-05-18 15:39:40 +08:00
1 parent 4761677676
commit e0c1cb6581
45 files changed
+3235 -434

No files matched your search

+3
View File
@@ -13,6 +13,7 @@ import { HashRouter, useLocation } from "react-router-dom";
import LandingPage from "./pages/LandingPage";
import LoginPage from "./pages/LoginPage";
import PageRoutes from "./pages/PageRoutes";
import PublicSharePage from "./pages/PublicSharePage";
import { NotFoundPage } from "./pages/NotFoundPage";
function AppProviders({ children }: { children: React.ReactNode }) {
@@ -47,6 +48,8 @@ export function App() {
<Route path="/login" element={<LoginPage />} />
<Route path="/login/:username" element={<LoginPage />} />
<Route path="/landing" element={<LandingPage />} />
{/* Public read-only conversation viewer — no authentication. */}
<Route path="/share/:shareCode" element={<PublicSharePage />} />
<Route path="/page/*" element={<PageRoutes />} />
<Route path="*" element={<NotFoundPage />} />
</Routes>
@@ -6,6 +6,13 @@ import type { AgentThreadState } from "@/core/threads";
export interface ThreadContextType {
thread: BaseStream<AgentThreadState>;
isMock?: boolean;
/**
* When set, the conversation is rendered from a public read-only share.
* Artifact/image URLs then resolve to the unauthenticated
* `/api/public/shares/{code}/files/...` endpoint instead of the
* auth-gated thread artifact route.
*/
publicShareCode?: string;
}
export const ThreadContext = createContext<ThreadContextType | undefined>(
@@ -2,6 +2,8 @@ import type { Message } from "@langchain/langgraph-sdk";
import { FileIcon, Loader2Icon } from "lucide-react";
import {
memo,
useCallback,
useContext,
useMemo,
type AnchorHTMLAttributes,
type ImgHTMLAttributes,
@@ -23,6 +25,7 @@ import {
import { Task, TaskTrigger } from "@/components/ai-elements/task";
import { Badge } from "@/components/ui/badge";
import { resolveArtifactURL } from "@/core/artifacts/utils";
import { getBackendBaseURL } from "@/core/config";
import { useI18n } from "@/core/i18n/hooks";
import {
extractContentFromMessage,
@@ -37,9 +40,33 @@ import { cn } from "@/lib/utils";
import { CopyButton } from "../copy-button";
import { ThreadContext } from "./context";
import { MarkdownContent } from "./markdown-content";
import { MessageTokenUsage } from "./message-token-usage";
/**
* Resolve an in-conversation file path to a fetchable URL.
*
* On a public read-only share (`ThreadContext.publicShareCode` set) artifact
* URLs point at the unauthenticated `/api/public/shares/{code}/files/...`
* endpoint; otherwise they use the normal auth-gated thread artifact route.
*/
function useArtifactResolver(threadId: string): (path: string) => string {
const publicShareCode = useContext(ThreadContext)?.publicShareCode;
return useCallback(
(path: string) => {
if (publicShareCode) {
const normalized = path.startsWith("/") ? path : `/${path}`;
return `${getBackendBaseURL()}/api/public/shares/${encodeURIComponent(
publicShareCode,
)}/files${normalized}`;
}
return resolveArtifactURL(path, threadId);
},
[publicShareCode, threadId],
);
}
export function MessageListItem({
className,
message,
@@ -101,6 +128,7 @@ function MessageImage({
threadId: string;
maxWidth?: string;
}) {
const resolveArtifact = useArtifactResolver(threadId);
if (!src) return null;
const imgClassName = cn("overflow-hidden rounded-lg", `max-w-[${maxWidth}]`);
@@ -109,7 +137,7 @@ function MessageImage({
return <img className={imgClassName} src={src} alt={alt} {...props} />;
}
const url = src.startsWith("/mnt/") ? resolveArtifactURL(src, threadId) : src;
const url = src.startsWith("/mnt/") ? resolveArtifact(src) : src;
return (
<a href={url} target="_blank" rel="noopener noreferrer">
@@ -133,6 +161,7 @@ function MessageContent_({
}) {
const rehypePlugins = useRehypeSplitWordsIntoSpans(isLoading);
const isHuman = message.type === "human";
const resolveArtifact = useArtifactResolver(threadId);
const components = useMemo(
() => ({
img: (props: ImgHTMLAttributes<HTMLImageElement>) => (
@@ -140,7 +169,7 @@ function MessageContent_({
),
a: ({ href, ...props }: AnchorHTMLAttributes<HTMLAnchorElement>) => {
if (href?.startsWith("/mnt/")) {
const url = resolveArtifactURL(href, threadId);
const url = resolveArtifact(href);
return (
<a
{...props}
@@ -153,7 +182,7 @@ function MessageContent_({
return <a {...props} href={href} />;
},
}),
[threadId],
[threadId, resolveArtifact],
);
const rawContent = extractContentFromMessage(message);
@@ -346,6 +375,7 @@ function RichFileCard({
threadId: string;
}) {
const { t } = useI18n();
const resolveArtifact = useArtifactResolver(threadId);
const isUploading = file.status === "uploading";
const isImage = isImageFile(file.filename);
@@ -378,7 +408,7 @@ function RichFileCard({
if (!file.path) return null;
const fileUrl = resolveArtifactURL(file.path, threadId);
const fileUrl = resolveArtifact(file.path);
if (isImage) {
return (
@@ -42,6 +42,7 @@ import {
SidebarMenuButton,
SidebarMenuItem,
} from "@/components/ui/sidebar";
import { ShareConversationDialog } from "@/components/workspace/thread-share-dialogs";
import { getAPIClient } from "@/core/api";
import { useI18n } from "@/core/i18n/hooks";
import {
@@ -76,6 +77,9 @@ export function RecentChatList() {
const [renameThreadId, setRenameThreadId] = useState<string | null>(null);
const [renameValue, setRenameValue] = useState("");
// Share dialog state — the conversation whose share code is being shown.
const [shareThreadId, setShareThreadId] = useState<string | null>(null);
const handleDelete = useCallback(
(threadId: string) => {
deleteThread({ threadId });
@@ -115,18 +119,10 @@ export function RecentChatList() {
}
}, [renameThread, renameThreadId, renameValue]);
const handleShare = useCallback(
async (thread: AgentThread) => {
const shareUrl = `${window.location.origin}${pathOfThread(thread)}`;
try {
await navigator.clipboard.writeText(shareUrl);
toast.success(t.clipboard.linkCopied);
} catch {
toast.error(t.clipboard.failedToCopyToClipboard);
}
},
[t],
);
const handleShare = useCallback((thread: AgentThread) => {
// Opens the share-code dialog; the recipient imports a copy via that code.
setShareThreadId(thread.thread_id);
}, []);
const handleExport = useCallback(
async (thread: AgentThread, format: "markdown" | "json") => {
@@ -173,89 +169,89 @@ export function RecentChatList() {
<div className="flex w-full flex-col gap-1">
{visibleThreads.map((thread) => {
const isActive = pathOfThread(thread) === pathname;
return (
<SidebarMenuItem
key={thread.thread_id}
className="group/side-menu-item"
>
<SidebarMenuButton isActive={isActive} asChild>
<Link
className="text-muted-foreground block w-full whitespace-nowrap group-hover/side-menu-item:overflow-hidden"
href={pathOfThread(thread)}
>
{titleOfThread(thread)}
</Link>
</SidebarMenuButton>
{env.NEXT_PUBLIC_STATIC_WEBSITE_ONLY !== "true" && (
<DropdownMenu>
<DropdownMenuTrigger asChild>
<SidebarMenuAction
showOnHover
className="bg-background/50 hover:bg-background"
>
<MoreHorizontal />
<span className="sr-only">{t.common.more}</span>
</SidebarMenuAction>
</DropdownMenuTrigger>
<DropdownMenuContent
className="w-48 rounded-lg"
side={"right"}
align={"start"}
>
<DropdownMenuItem
onSelect={() =>
handleRenameClick(
thread.thread_id,
titleOfThread(thread),
)
}
>
<Pencil className="text-muted-foreground" />
<span>{t.common.rename}</span>
</DropdownMenuItem>
<DropdownMenuItem
onSelect={() => handleShare(thread)}
>
<Share2 className="text-muted-foreground" />
<span>{t.common.share}</span>
</DropdownMenuItem>
<DropdownMenuSub>
<DropdownMenuSubTrigger>
<Download className="text-muted-foreground" />
<span>{t.common.export}</span>
</DropdownMenuSubTrigger>
<DropdownMenuSubContent>
<DropdownMenuItem
onSelect={() =>
handleExport(thread, "markdown")
}
>
<FileText className="text-muted-foreground" />
<span>{t.common.exportAsMarkdown}</span>
</DropdownMenuItem>
<DropdownMenuItem
onSelect={() =>
handleExport(thread, "json")
}
>
<FileJson className="text-muted-foreground" />
<span>{t.common.exportAsJSON}</span>
</DropdownMenuItem>
</DropdownMenuSubContent>
</DropdownMenuSub>
<DropdownMenuSeparator />
<DropdownMenuItem
onSelect={() => handleDelete(thread.thread_id)}
>
<Trash2 className="text-muted-foreground" />
<span>{t.common.delete}</span>
</DropdownMenuItem>
</DropdownMenuContent>
</DropdownMenu>
)}
</SidebarMenuItem>
);
})}
return (
<SidebarMenuItem
key={thread.thread_id}
className="group/side-menu-item"
>
<SidebarMenuButton isActive={isActive} asChild>
<Link
className="text-muted-foreground block w-full whitespace-nowrap group-hover/side-menu-item:overflow-hidden"
href={pathOfThread(thread)}
>
{titleOfThread(thread)}
</Link>
</SidebarMenuButton>
{env.NEXT_PUBLIC_STATIC_WEBSITE_ONLY !== "true" && (
<DropdownMenu>
<DropdownMenuTrigger asChild>
<SidebarMenuAction
showOnHover
className="bg-background/50 hover:bg-background"
>
<MoreHorizontal />
<span className="sr-only">{t.common.more}</span>
</SidebarMenuAction>
</DropdownMenuTrigger>
<DropdownMenuContent
className="w-48 rounded-lg"
side={"right"}
align={"start"}
>
<DropdownMenuItem
onSelect={() =>
handleRenameClick(
thread.thread_id,
titleOfThread(thread),
)
}
>
<Pencil className="text-muted-foreground" />
<span>{t.common.rename}</span>
</DropdownMenuItem>
<DropdownMenuItem
onSelect={() => handleShare(thread)}
>
<Share2 className="text-muted-foreground" />
<span>{t.common.share}</span>
</DropdownMenuItem>
<DropdownMenuSub>
<DropdownMenuSubTrigger>
<Download className="text-muted-foreground" />
<span>{t.common.export}</span>
</DropdownMenuSubTrigger>
<DropdownMenuSubContent>
<DropdownMenuItem
onSelect={() =>
handleExport(thread, "markdown")
}
>
<FileText className="text-muted-foreground" />
<span>{t.common.exportAsMarkdown}</span>
</DropdownMenuItem>
<DropdownMenuItem
onSelect={() =>
handleExport(thread, "json")
}
>
<FileJson className="text-muted-foreground" />
<span>{t.common.exportAsJSON}</span>
</DropdownMenuItem>
</DropdownMenuSubContent>
</DropdownMenuSub>
<DropdownMenuSeparator />
<DropdownMenuItem
onSelect={() => handleDelete(thread.thread_id)}
>
<Trash2 className="text-muted-foreground" />
<span>{t.common.delete}</span>
</DropdownMenuItem>
</DropdownMenuContent>
</DropdownMenu>
)}
</SidebarMenuItem>
);
})}
</div>
</SidebarMenu>
</SidebarGroupContent>
@@ -291,6 +287,15 @@ export function RecentChatList() {
</DialogFooter>
</DialogContent>
</Dialog>
{/* Share Dialog */}
<ShareConversationDialog
threadId={shareThreadId}
open={shareThreadId !== null}
onOpenChange={(o) => {
if (!o) setShareThreadId(null);
}}
/>
</>
);
}
@@ -0,0 +1,220 @@
"use client";
import {
Select,
SelectContent,
SelectItem,
SelectTrigger,
SelectValue,
} from "@/components/ui/select";
// ---- 时间配置 ----
export type ScheduleFreq = "hourly" | "daily" | "weekly" | "monthly";
export interface ScheduleConfig {
freq: ScheduleFreq;
minute: number;
hour: number;
weekday: number;
date: number;
}
export const DEFAULT_SCHEDULE_CONFIG: ScheduleConfig = {
freq: "daily",
minute: 0,
hour: 9,
weekday: 1,
date: 1,
};
const WEEKDAYS = ["周日", "周一", "周二", "周三", "周四", "周五", "周六"];
const HOURS = Array.from({ length: 24 }, (_, i) => i);
const MINUTES = [0, 15, 30, 45];
const DATES = Array.from({ length: 31 }, (_, i) => i + 1);
export function buildCron(cfg: ScheduleConfig): string {
// Keep the minute slot when the user picked one for daily/weekly/monthly too,
// so "每天 9:30" doesn't silently truncate to 9:00 on save → reload round trips.
const m = Number.isFinite(cfg.minute) ? cfg.minute : 0;
switch (cfg.freq) {
case "hourly":
return `${m} * * * *`;
case "daily":
return `${m} ${cfg.hour} * * *`;
case "weekly":
return `${m} ${cfg.hour} * * ${cfg.weekday}`;
case "monthly":
return `${m} ${cfg.hour} ${cfg.date} * *`;
}
}
// Parse a 5-field cron expression into the form picker state.
//
// Bug we keep getting bitten by: the edit dialog used to call
// `parseCron(task.schedule_text)` and `schedule_text` may be the raw user
// input (free-text like "每天 9 点") rather than a real cron string. The
// fallback `DEFAULT_SCHEDULE_CONFIG` then made the picker silently
// reinterpret the task. Callers should prefer `task.cron_expr` (the canonical
// schedule produced by `_schedule_to_cron`); this function stays the single
// source of truth for cron → picker, and also handles the looser step-syntax
// variants (e.g. "0 */3 * * *") we sometimes generate.
export function parseCron(cron: string | null | undefined): ScheduleConfig | null {
if (!cron) return null;
const parts = cron.trim().split(/\s+/);
if (parts.length !== 5) return null;
const [min, hr, date, , wd] = parts;
const minute = parseInt(min ?? "", 10);
const hour = parseInt(hr ?? "", 10);
// Hourly: ``M * * * *`` — match before the daily branch so a minute-only
// cron isn't misread as daily.
if (hr === "*" && date === "*" && wd === "*") {
if (!isNaN(minute)) {
return { ...DEFAULT_SCHEDULE_CONFIG, freq: "hourly", minute };
}
}
// ``0 */N * * *`` (every N hours): map to hourly with that minute; the
// picker doesn't model "every N hours" yet, so this is the safest fallback
// that at least doesn't silently round to 9:00.
if ((hr ?? "").startsWith("*/") && date === "*" && wd === "*") {
if (!isNaN(minute)) {
return { ...DEFAULT_SCHEDULE_CONFIG, freq: "hourly", minute };
}
}
if (date === "*" && wd === "*") {
if (!isNaN(hour)) {
return {
...DEFAULT_SCHEDULE_CONFIG,
freq: "daily",
hour,
minute: isNaN(minute) ? 0 : minute,
};
}
}
if (date === "*" && wd !== "*") {
const weekday = parseInt(wd ?? "", 10);
if (!isNaN(hour) && !isNaN(weekday)) {
return {
...DEFAULT_SCHEDULE_CONFIG,
freq: "weekly",
hour,
weekday,
minute: isNaN(minute) ? 0 : minute,
};
}
}
if (wd === "*" && date !== "*") {
const d = parseInt(date ?? "", 10);
if (!isNaN(hour) && !isNaN(d)) {
return {
...DEFAULT_SCHEDULE_CONFIG,
freq: "monthly",
hour,
date: d,
minute: isNaN(minute) ? 0 : minute,
};
}
}
return null;
}
export function ScheduleBuilder({
value,
onChange,
}: {
value: ScheduleConfig;
onChange: (cfg: ScheduleConfig) => void;
}) {
const set = (patch: Partial<ScheduleConfig>) => onChange({ ...value, ...patch });
return (
<div className="flex flex-wrap gap-2">
<Select value={value.freq} onValueChange={(v) => set({ freq: v as ScheduleFreq })}>
<SelectTrigger className="w-28">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="hourly">每小时</SelectItem>
<SelectItem value="daily">每天</SelectItem>
<SelectItem value="weekly">每周</SelectItem>
<SelectItem value="monthly">每月</SelectItem>
</SelectContent>
</Select>
{value.freq === "hourly" && (
<Select value={String(value.minute)} onValueChange={(v) => set({ minute: parseInt(v) })}>
<SelectTrigger className="w-24">
<SelectValue />
</SelectTrigger>
<SelectContent>
{MINUTES.map((m) => (
<SelectItem key={m} value={String(m)}>{m} 分</SelectItem>
))}
</SelectContent>
</Select>
)}
{value.freq === "weekly" && (
<Select value={String(value.weekday)} onValueChange={(v) => set({ weekday: parseInt(v) })}>
<SelectTrigger className="w-24">
<SelectValue />
</SelectTrigger>
<SelectContent>
{WEEKDAYS.map((d, i) => (
<SelectItem key={i} value={String(i)}>{d}</SelectItem>
))}
</SelectContent>
</Select>
)}
{value.freq === "monthly" && (
<Select value={String(value.date)} onValueChange={(v) => set({ date: parseInt(v) })}>
<SelectTrigger className="w-24">
<SelectValue />
</SelectTrigger>
<SelectContent>
{DATES.map((d) => (
<SelectItem key={d} value={String(d)}>第 {d} 日</SelectItem>
))}
</SelectContent>
</Select>
)}
{value.freq !== "hourly" && (
<Select value={String(value.hour)} onValueChange={(v) => set({ hour: parseInt(v) })}>
<SelectTrigger className="w-24">
<SelectValue />
</SelectTrigger>
<SelectContent>
{HOURS.map((h) => (
<SelectItem key={h} value={String(h)}>{h} 时</SelectItem>
))}
</SelectContent>
</Select>
)}
</div>
);
}
export function statusLabel(status?: string | null) {
switch (status) {
case "succeeded":
return "成功";
case "failed":
return "失败";
case "running":
return "执行中";
case "pending":
return "等待中";
default:
return status || "未知";
}
}
export function statusVariant(
status?: string | null,
): "default" | "secondary" | "destructive" | "outline" {
if (status === "succeeded") return "default";
if (status === "failed") return "destructive";
if (status === "running") return "secondary";
return "outline";
}
@@ -1,10 +1,24 @@
"use client";
import { Loader2, PencilIcon } from "lucide-react";
import {
ChevronDownIcon,
FileCode2Icon,
FileTextIcon,
GlobeIcon,
Loader2,
PencilIcon,
SparklesIcon,
} from "lucide-react";
import { useEffect, useMemo, useState } from "react";
import { toast } from "sonner";
import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import {
Collapsible,
CollapsibleContent,
CollapsibleTrigger,
} from "@/components/ui/collapsible";
import {
Dialog,
DialogContent,
@@ -23,15 +37,18 @@ import {
} from "@/core/skills/hooks";
import type { Skill } from "@/core/skills/type";
import { reasoningPlugins } from "@/core/streamdown";
import { cn } from "@/lib/utils";
/**
* Whether the current user may edit a skill's detail text.
* Whether the current user may edit a skill's detail text and view its raw
* SKILL.md source.
*
* Admins can edit any skill; the publisher (owner of a custom skill) can edit
* Admins qualify for any skill; the publisher (owner of a custom skill) for
* their own. Built-in skills have no human owner, so only admins qualify.
* Mirrors the backend authz in `PUT /api/skills/{name}/detail`.
* Mirrors the backend authz in `PUT /api/skills/{name}/detail` and
* `GET /api/skills/{name}/content`.
*/
function canEditSkillDetail(skill: Skill | null | undefined): boolean {
function canManageSkill(skill: Skill | null | undefined): boolean {
if (!skill) return false;
const auth = getStoredAuth();
if (!auth) return false;
@@ -45,8 +62,9 @@ function canEditSkillDetail(skill: Skill | null | undefined): boolean {
/**
* Reusable skill detail dialog. Given a skill name it fetches the skill
* metadata (including the admin/owner-authored `detail`) and the raw SKILL.md
* source, and lets authorized users edit the detail inline.
* metadata (including the admin/owner-authored `detail`) and lets authorized
* users edit the detail inline. The raw SKILL.md source is shown only to the
* publisher or an admin — everyone else sees just the description and detail.
*/
export function SkillDetailDialog({
skillName,
@@ -62,17 +80,22 @@ export function SkillDetailDialog({
// Only fetch while the dialog is open to avoid background requests.
const activeName = open ? skillName : null;
const { data: skill, isLoading: isSkillLoading } = useSkill(activeName);
const {
data: contentData,
isLoading: isContentLoading,
error: contentError,
} = useSkillContent(activeName);
const setDetail = useSetSkillDetail();
const [editing, setEditing] = useState(false);
const [draft, setDraft] = useState("");
const [showSource, setShowSource] = useState(false);
const canManage = useMemo(() => canManageSkill(skill), [skill]);
// The SKILL.md source endpoint is publisher/admin-only — only request it
// when the viewer is actually allowed to see it.
const {
data: contentData,
isLoading: isContentLoading,
error: contentError,
} = useSkillContent(open && canManage ? skillName : null);
// Reset transient UI whenever the dialog closes or switches to another skill.
useEffect(() => {
if (!open) {
@@ -81,8 +104,6 @@ export function SkillDetailDialog({
}
}, [open, skillName]);
const canEdit = useMemo(() => canEditSkillDetail(skill), [skill]);
function startEditing() {
setDraft(skill?.detail ?? "");
setEditing(true);
@@ -101,22 +122,54 @@ export function SkillDetailDialog({
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="flex max-h-[80vh] flex-col gap-3 sm:max-w-2xl">
<DialogHeader>
<DialogTitle>{skill?.name ?? skillName ?? ""}</DialogTitle>
<DialogDescription>{skill?.description}</DialogDescription>
<DialogContent className="flex max-h-[85vh] flex-col gap-0 overflow-hidden p-0 sm:max-w-2xl">
{/* Header */}
<DialogHeader className="space-y-0 border-b px-6 py-4 text-left">
<div className="flex items-start gap-3">
<div className="bg-primary/10 text-primary flex size-9 shrink-0 items-center justify-center rounded-lg">
<SparklesIcon className="size-5" />
</div>
<div className="min-w-0 flex-1">
<div className="flex flex-wrap items-center gap-2">
<DialogTitle className="truncate text-base">
{skill?.name ?? skillName ?? ""}
</DialogTitle>
{skill?.builtin && (
<Badge variant="secondary" className="text-xs">
{tx.badgeBuiltin}
</Badge>
)}
{skill?.published && (
<Badge
variant="outline"
className="border-sky-500/30 text-sky-700 dark:text-sky-300 text-xs"
>
<GlobeIcon className="size-3" />
{tx.badgePublished}
</Badge>
)}
</div>
<DialogDescription className="mt-1 text-sm leading-relaxed">
{skill?.description || "—"}
</DialogDescription>
</div>
</div>
</DialogHeader>
<div className="min-h-0 flex-1 space-y-4 overflow-auto pr-1">
{/* Body */}
<div className="min-h-0 flex-1 space-y-5 overflow-auto px-6 py-5">
{/* Detail — human-authored long-form introduction */}
<section className="space-y-2">
<div className="flex items-center justify-between">
<h3 className="text-sm font-medium">{tx.detailSectionTitle}</h3>
{canEdit && !editing && (
<h3 className="text-foreground/90 flex items-center gap-1.5 text-sm font-semibold">
<FileTextIcon className="text-muted-foreground size-4" />
{tx.detailSectionTitle}
</h3>
{canManage && !editing && (
<Button
size="sm"
variant="ghost"
className="h-7"
className="text-muted-foreground hover:text-foreground h-7 px-2"
onClick={startEditing}
>
<PencilIcon className="size-3.5" />
@@ -124,15 +177,17 @@ export function SkillDetailDialog({
</Button>
)}
</div>
{editing ? (
<div className="space-y-2">
<Textarea
value={draft}
onChange={(e) => setDraft(e.currentTarget.value)}
placeholder={tx.detailPlaceholder}
className="max-h-[320px] min-h-40 resize-y overflow-y-auto text-sm"
className="max-h-[320px] min-h-44 resize-y overflow-y-auto text-sm leading-relaxed"
autoFocus
/>
<div className="flex justify-end gap-2">
<div className="flex items-center justify-end gap-2">
<Button
size="sm"
variant="outline"
@@ -154,12 +209,12 @@ export function SkillDetailDialog({
</div>
</div>
) : isSkillLoading ? (
<div className="text-muted-foreground flex items-center gap-2 text-sm">
<div className="text-muted-foreground flex items-center gap-2 rounded-lg border border-dashed px-4 py-8 text-sm">
<Loader2 className="size-4 animate-spin" />
{t.common.loading}
</div>
) : skill?.detail ? (
<div className="bg-muted/40 rounded-md border p-3 text-sm">
<div className="bg-muted/30 rounded-lg border p-4 text-sm leading-relaxed">
<MarkdownContent
content={skill.detail}
isLoading={false}
@@ -168,44 +223,55 @@ export function SkillDetailDialog({
/>
</div>
) : (
<p className="text-muted-foreground text-sm">
{tx.detailEmpty}
{!canEdit ? ` ${tx.detailAdminOnlyHint}` : ""}
</p>
)}
</section>
{/* SKILL.md source — collapsed by default */}
<section className="space-y-2">
<button
type="button"
className="text-muted-foreground hover:text-foreground text-sm font-medium"
onClick={() => setShowSource((v) => !v)}
>
{showSource ? "▾ " : "▸ "}
{tx.contentSectionTitle}
</button>
{showSource && (
<div className="bg-muted/40 overflow-auto rounded-md border">
{isContentLoading ? (
<div className="text-muted-foreground flex items-center justify-center gap-2 py-10 text-sm">
<Loader2 className="size-4 animate-spin" />
{t.common.loading}
</div>
) : contentError ? (
<div className="text-destructive p-4 text-sm">
{contentError instanceof Error
? contentError.message
: String(contentError)}
</div>
) : (
<pre className="text-foreground/90 max-h-[320px] overflow-auto whitespace-pre-wrap wrap-break-word p-4 font-mono text-xs leading-relaxed">
{contentData?.content}
</pre>
<div className="text-muted-foreground rounded-lg border border-dashed px-4 py-8 text-center">
<FileTextIcon className="mx-auto mb-2 size-6 opacity-40" />
<p className="text-sm">{tx.detailEmpty}</p>
{!canManage && (
<p className="mt-1 text-xs">{tx.detailAdminOnlyHint}</p>
)}
</div>
)}
</section>
{/* SKILL.md source — publisher / admin only, collapsed by default */}
{canManage && (
<section>
<Collapsible open={showSource} onOpenChange={setShowSource}>
<CollapsibleTrigger className="bg-muted/30 hover:bg-muted/60 flex w-full items-center justify-between rounded-lg border px-4 py-2.5 text-sm font-semibold transition-colors">
<span className="flex items-center gap-1.5">
<FileCode2Icon className="text-muted-foreground size-4" />
{tx.contentSectionTitle}
</span>
<ChevronDownIcon
className={cn(
"text-muted-foreground size-4 transition-transform",
showSource && "rotate-180",
)}
/>
</CollapsibleTrigger>
<CollapsibleContent className="pt-2">
<div className="bg-muted/50 max-h-[340px] overflow-auto rounded-lg border">
{isContentLoading ? (
<div className="text-muted-foreground flex items-center justify-center gap-2 py-12 text-sm">
<Loader2 className="size-4 animate-spin" />
{t.common.loading}
</div>
) : contentError ? (
<div className="text-destructive p-4 text-sm">
{contentError instanceof Error
? contentError.message
: String(contentError)}
</div>
) : (
<pre className="text-foreground/80 p-4 font-mono text-xs leading-relaxed break-words whitespace-pre-wrap">
{contentData?.content}
</pre>
)}
</div>
</CollapsibleContent>
</Collapsible>
</section>
)}
</div>
</DialogContent>
</Dialog>
@@ -0,0 +1,328 @@
"use client";
import {
CheckIcon,
CopyIcon,
DownloadIcon,
ExternalLinkIcon,
Loader2,
Trash2Icon,
} from "lucide-react";
import { useEffect, useState } from "react";
import { toast } from "sonner";
import { Button } from "@/components/ui/button";
import {
Dialog,
DialogContent,
DialogDescription,
DialogHeader,
DialogTitle,
} from "@/components/ui/dialog";
import { Input } from "@/components/ui/input";
import { useI18n } from "@/core/i18n/hooks";
import type {
SharePreview,
ThreadShare,
ThreadShareMode,
} from "@/core/thread-shares/api";
import { buildPublicShareUrl, extractShareCode } from "@/core/thread-shares/api";
import {
useCreateShare,
useImportShare,
usePreviewShare,
useRevokeShare,
} from "@/core/thread-shares/hooks";
import { cn } from "@/lib/utils";
function errorMessage(err: unknown): string {
return err instanceof Error ? err.message : String(err);
}
/**
* Dialog that produces a share link for a single conversation. Two modes:
*
* - `import` — mints a short code; a recipient enters it in the import dialog
* to copy the conversation into their own account.
* - `view` — mints a public, read-only link that anyone can open without
* logging in (a static page; viewers cannot send new messages).
*/
export function ShareConversationDialog({
threadId,
open,
onOpenChange,
}: {
threadId: string | null;
open: boolean;
onOpenChange: (open: boolean) => void;
}) {
const { t } = useI18n();
const tx = t.threadShare;
const createShare = useCreateShare();
const revokeShare = useRevokeShare();
const [mode, setMode] = useState<ThreadShareMode>("view");
const [share, setShare] = useState<ThreadShare | null>(null);
const [copied, setCopied] = useState(false);
// Reset to the default mode (read-only link) every time the dialog opens.
useEffect(() => {
if (open) setMode("view");
}, [open, threadId]);
// Mint / reuse the link whenever the dialog opens or the mode changes.
useEffect(() => {
if (!open || !threadId) return;
setShare(null);
setCopied(false);
createShare
.mutateAsync({ threadId, mode })
.then(setShare)
.catch((err) => toast.error(errorMessage(err)));
// createShare is stable enough; re-running only on open/threadId/mode is intended.
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [open, threadId, mode]);
const isView = mode === "view";
const publicUrl =
isView && share ? buildPublicShareUrl(share.share_code) : "";
async function handleCopy() {
if (!share) return;
const text = isView ? publicUrl : share.share_code;
try {
await navigator.clipboard.writeText(text);
setCopied(true);
toast.success(isView ? "链接已复制" : tx.codeCopied);
setTimeout(() => setCopied(false), 1500);
} catch {
toast.error(t.clipboard.failedToCopyToClipboard);
}
}
async function handleRevoke() {
if (!share || !window.confirm(tx.revokeConfirm)) return;
try {
await revokeShare.mutateAsync(share.share_code);
toast.success(tx.revokeSuccess);
onOpenChange(false);
} catch (err) {
toast.error(errorMessage(err));
}
}
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="sm:max-w-md">
<DialogHeader>
<DialogTitle>{tx.shareTitle}</DialogTitle>
<DialogDescription>
{isView
? "生成一个只读链接,任何人无需登录都能查看这个对话,但不能继续提问。"
: tx.shareDescription}
</DialogDescription>
</DialogHeader>
{/* Mode selector */}
<div className="bg-muted flex gap-1 rounded-lg p-1">
{(
[
{ value: "view" as const, label: "只读链接" },
{ value: "import" as const, label: "导入副本" },
]
).map((option) => (
<button
key={option.value}
type="button"
onClick={() => setMode(option.value)}
className={cn(
"flex-1 rounded-md px-3 py-1.5 text-sm transition-colors",
mode === option.value
? "bg-background text-foreground shadow-sm"
: "text-muted-foreground hover:text-foreground",
)}
>
{option.label}
</button>
))}
</div>
{!share ? (
<div className="text-muted-foreground flex items-center gap-2 py-8 text-sm">
<Loader2 className="size-4 animate-spin" />
{tx.generating}
</div>
) : (
<div className="space-y-3">
<div>
<p className="text-muted-foreground mb-1 text-xs">
{isView ? "只读链接" : tx.shareCodeLabel}
</p>
<div className="flex items-center gap-2">
{isView ? (
<Input
readOnly
value={publicUrl}
onFocus={(e) => e.currentTarget.select()}
className="flex-1 font-mono text-xs"
/>
) : (
<code className="bg-muted flex-1 rounded-md px-3 py-2 text-center font-mono text-lg tracking-[0.3em]">
{share.share_code}
</code>
)}
<Button variant="outline" onClick={() => void handleCopy()}>
{copied ? (
<CheckIcon className="size-4" />
) : (
<CopyIcon className="size-4" />
)}
{isView ? "复制链接" : tx.copyCode}
</Button>
</div>
</div>
{isView ? (
<div className="space-y-3">
<p className="text-muted-foreground text-xs">
拿到链接的人都能查看这个对话的全部内容(只读),无法继续发送问题。撤销后链接立即失效。
</p>
<Button
variant="outline"
size="sm"
onClick={() => window.open(publicUrl, "_blank", "noopener")}
>
<ExternalLinkIcon className="size-3.5" />
在新标签页打开
</Button>
</div>
) : (
<p className="text-muted-foreground text-xs">{tx.shareHint}</p>
)}
<Button
variant="ghost"
size="sm"
className="text-destructive hover:text-destructive"
disabled={revokeShare.isPending}
onClick={() => void handleRevoke()}
>
<Trash2Icon className="size-3.5" />
{tx.revoke}
</Button>
</div>
)}
</DialogContent>
</Dialog>
);
}
/** Dialog where a user pastes a share code to import someone's conversation. */
export function ImportConversationsDialog({
open,
onOpenChange,
}: {
open: boolean;
onOpenChange: (open: boolean) => void;
}) {
const { t } = useI18n();
const tx = t.threadShare;
const [input, setInput] = useState("");
const [preview, setPreview] = useState<SharePreview | null>(null);
const previewShare = usePreviewShare();
const importShare = useImportShare();
// Reset state whenever the dialog closes.
useEffect(() => {
if (!open) {
setInput("");
setPreview(null);
}
}, [open]);
async function handlePreview() {
const code = extractShareCode(input);
if (!code) {
toast.error(tx.emptyCode);
return;
}
try {
setPreview(await previewShare.mutateAsync(code));
} catch (err) {
setPreview(null);
toast.error(errorMessage(err));
}
}
async function handleImport() {
const code = preview?.share_code ?? extractShareCode(input);
if (!code) {
toast.error(tx.emptyCode);
return;
}
try {
const result = await importShare.mutateAsync(code);
if (result.imported > 0) {
toast.success(tx.importSuccess);
} else {
toast.info(tx.alreadyImported);
}
onOpenChange(false);
} catch (err) {
toast.error(errorMessage(err));
}
}
return (
<Dialog open={open} onOpenChange={onOpenChange}>
<DialogContent className="sm:max-w-md">
<DialogHeader>
<DialogTitle>{tx.importTitle}</DialogTitle>
<DialogDescription>{tx.importDescription}</DialogDescription>
</DialogHeader>
<div className="flex gap-2">
<Input
value={input}
onChange={(e) => {
setInput(e.target.value);
setPreview(null);
}}
placeholder={tx.codeInputPlaceholder}
className="font-mono"
/>
<Button
variant="outline"
onClick={() => void handlePreview()}
disabled={previewShare.isPending || !input.trim()}
>
{previewShare.isPending && (
<Loader2 className="size-4 animate-spin" />
)}
{tx.preview}
</Button>
</div>
{preview && (
<div className="space-y-2 rounded-lg border p-3 text-sm">
<p className="font-medium">
{tx.previewConversation.replace("{title}", preview.title)}
</p>
<p className="text-muted-foreground text-xs">
{tx.previewFrom.replace("{user}", preview.shared_by)}
</p>
<Button
className="w-full"
onClick={() => void handleImport()}
disabled={importShare.isPending}
>
{importShare.isPending ? (
<Loader2 className="size-4 animate-spin" />
) : (
<DownloadIcon className="size-4" />
)}
{importShare.isPending ? tx.importing : tx.doImport}
</Button>
</div>
)}
</DialogContent>
</Dialog>
);
}
@@ -13,7 +13,7 @@ import {
} from "@/components/ui/sidebar";
import { getStoredAuth } from "@/core/auth";
import { useI18n } from "@/core/i18n/hooks";
import { BotIcon, ListChecks, MessageSquarePlus, MessagesSquare, SparklesIcon, UsersIcon } from "lucide-react";
import { BotIcon, CalendarClock, ListChecks, MessageSquarePlus, MessagesSquare, SparklesIcon, UsersIcon } from "lucide-react";
export function WorkspaceNavChatList() {
const { t } = useI18n();
@@ -77,7 +77,7 @@ export function WorkspaceNavChatList() {
{isAdmin && (
<SidebarMenuItem>
<SidebarMenuButton
isActive={pathname.startsWith("/page/workspace/admin")}
isActive={pathname === "/page/workspace/admin/users"}
asChild
>
<Link className="text-sidebar-foreground" href="/page/workspace/admin/users">
@@ -87,6 +87,22 @@ export function WorkspaceNavChatList() {
</SidebarMenuButton>
</SidebarMenuItem>
)}
{isAdmin && (
<SidebarMenuItem>
<SidebarMenuButton
isActive={pathname === "/page/workspace/admin/scheduled-tasks"}
asChild
>
<Link
className="text-sidebar-foreground"
href="/page/workspace/admin/scheduled-tasks"
>
<CalendarClock />
<span>定时任务管理</span>
</Link>
</SidebarMenuButton>
</SidebarMenuItem>
)}
</SidebarMenu>
</SidebarGroup>
);
@@ -250,6 +250,35 @@ export const enUS: Translations = {
searchChats: "Search chats",
},
// Conversation sharing
threadShare: {
importChats: "Import chat",
shareTitle: "Share conversation",
shareDescription:
'Share this conversation with others. They enter the code under "Import chat" to copy it into their own account and keep chatting.',
generating: "Generating code...",
shareCodeLabel: "Share code",
copyCode: "Copy",
codeCopied: "Share code copied",
shareHint:
"Anyone who has the code can copy this conversation — share it carefully.",
revoke: "Revoke share",
revokeConfirm: "Revoking makes the code unusable immediately. Continue?",
revokeSuccess: "Share revoked",
importTitle: "Import a conversation",
importDescription:
"Enter a share code to copy the shared conversation into your account.",
codeInputPlaceholder: "Paste a share code",
preview: "Preview",
previewConversation: "Conversation: {title}",
previewFrom: "From user {user}",
doImport: "Import",
importing: "Importing...",
importSuccess: "Conversation imported",
alreadyImported: "You have already imported this conversation",
emptyCode: "Enter a share code first",
},
// Page titles (document title)
pages: {
appName: "CM Assistant",
@@ -194,6 +194,32 @@ export interface Translations {
searchChats: string;
};
// Conversation sharing
threadShare: {
importChats: string;
shareTitle: string;
shareDescription: string;
generating: string;
shareCodeLabel: string;
copyCode: string;
codeCopied: string;
shareHint: string;
revoke: string;
revokeConfirm: string;
revokeSuccess: string;
importTitle: string;
importDescription: string;
codeInputPlaceholder: string;
preview: string;
previewConversation: string;
previewFrom: string;
doImport: string;
importing: string;
importSuccess: string;
alreadyImported: string;
emptyCode: string;
};
// Page titles (document title)
pages: {
appName: string;
@@ -238,6 +238,33 @@ export const zhCN: Translations = {
searchChats: "搜索对话",
},
// Conversation sharing
threadShare: {
importChats: "导入对话",
shareTitle: "分享对话",
shareDescription:
"把这个对话分享给他人。对方在「导入对话」里填入分享码,即可把它复制一份到自己账号并继续聊天。",
generating: "生成分享码中...",
shareCodeLabel: "分享码",
copyCode: "复制",
codeCopied: "分享码已复制",
shareHint: "知道分享码的人都能复制这个对话,请谨慎分享。",
revoke: "撤销分享",
revokeConfirm: "撤销后该分享码立即失效,确定撤销?",
revokeSuccess: "已撤销分享",
importTitle: "导入对话",
importDescription: "填入对方给你的分享码,把 TA 分享的那个对话复制到你的账号。",
codeInputPlaceholder: "粘贴分享码",
preview: "预览",
previewConversation: "对话:{title}",
previewFrom: "来自用户 {user}",
doImport: "导入",
importing: "导入中...",
importSuccess: "对话已导入",
alreadyImported: "你已经导入过这个对话了",
emptyCode: "请先填入分享码",
},
// Page titles (document title)
pages: {
appName: "CM 助手",
@@ -296,3 +296,61 @@ export async function unsubscribeScheduledTask(taskId: string) {
}),
);
}
// --- Admin (cross-user management) ---
/** List every scheduled task across all users. Admin-only endpoint. */
export async function listAllScheduledTasks(search?: string) {
const qs = new URLSearchParams();
if (search) qs.set("search", search);
const query = qs.toString();
return parseResponse<ScheduledTask[]>(
await apiFetch(`${baseUrl()}/admin/all${query ? `?${query}` : ""}`),
);
}
/** List runs for any task regardless of owner. Admin-only endpoint. */
export async function adminListScheduledTaskRuns(taskId: string) {
return parseResponse<ScheduledTaskRun[]>(
await apiFetch(`${baseUrl()}/admin/${encodeURIComponent(taskId)}/runs`),
);
}
/** Update any task regardless of owner. Admin-only endpoint. */
export async function adminUpdateScheduledTask(
taskId: string,
data: Partial<{
name: string;
prompt: string;
schedule: string;
timezone: string;
enabled: boolean;
execution_agent_name: string | null;
}>,
) {
return parseResponse<ScheduledTask>(
await apiFetch(`${baseUrl()}/admin/${encodeURIComponent(taskId)}`, {
method: "PATCH",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(data),
}),
);
}
/** Delete any task regardless of owner. Admin-only endpoint. */
export async function adminDeleteScheduledTask(taskId: string) {
return parseResponse<{ success: boolean }>(
await apiFetch(`${baseUrl()}/admin/${encodeURIComponent(taskId)}`, {
method: "DELETE",
}),
);
}
/** Trigger an immediate run of any task regardless of owner. Admin-only. */
export async function adminTriggerScheduledTask(taskId: string) {
return parseResponse<{ success: boolean; task: ScheduledTask }>(
await apiFetch(`${baseUrl()}/admin/${encodeURIComponent(taskId)}/trigger`, {
method: "POST",
}),
);
}
@@ -6,12 +6,17 @@ import {
} from "@tanstack/react-query";
import {
adminDeleteScheduledTask,
adminListScheduledTaskRuns,
adminTriggerScheduledTask,
adminUpdateScheduledTask,
createScheduledTask,
deleteScheduledTask,
deleteScheduledTaskRun,
getSchedulerThread,
getScheduledTaskDeliveryProfile,
getScheduledTaskRun,
listAllScheduledTasks,
listPublishedScheduledTasks,
listScheduledTaskRuns,
listScheduledTasks,
@@ -74,6 +79,58 @@ export function useScheduledTaskDeliveryProfile() {
});
}
/** Admin: every scheduled task across all users. */
export function useAllScheduledTasks(
params: { search?: string; enabled?: boolean } = {},
) {
const search = params.search ?? "";
return useQuery({
queryKey: ["scheduled-tasks", "admin", "list", search],
queryFn: () => listAllScheduledTasks(search),
enabled: params.enabled ?? true,
refetchOnWindowFocus: false,
placeholderData: keepPreviousData,
});
}
/** Admin: runs for any task regardless of owner. */
export function useAdminScheduledTaskRuns(taskId?: string | null) {
return useQuery({
queryKey: ["scheduled-tasks", "admin", taskId, "runs"],
queryFn: () => adminListScheduledTaskRuns(taskId!),
enabled: !!taskId,
refetchOnWindowFocus: false,
});
}
/** Admin mutations: edit / delete / trigger any task. */
export function useAdminScheduledTaskActions() {
const queryClient = useQueryClient();
const invalidate = () =>
queryClient.invalidateQueries({ queryKey: ["scheduled-tasks"] });
return {
update: useMutation({
mutationFn: ({
taskId,
data,
}: {
taskId: string;
data: Parameters<typeof adminUpdateScheduledTask>[1];
}) => adminUpdateScheduledTask(taskId, data),
onSuccess: invalidate,
}),
remove: useMutation({
mutationFn: adminDeleteScheduledTask,
onSuccess: invalidate,
}),
trigger: useMutation({
mutationFn: adminTriggerScheduledTask,
onSuccess: invalidate,
}),
};
}
export function usePublishedScheduledTasks(enabled: boolean = true) {
return useQuery({
queryKey: ["scheduled-tasks", "published"],
@@ -94,6 +94,7 @@ export type ScheduledTaskRun = {
status: "pending" | "running" | "succeeded" | "failed" | string;
error?: string | null;
result?: ScheduledTaskRunResult | null;
retry_count?: number | null;
created_at?: string | null;
updated_at?: string | null;
can_edit?: boolean;
+137
View File
@@ -0,0 +1,137 @@
import type { Message } from "@langchain/langgraph-sdk";
import { apiFetch } from "@/core/api";
import { getBackendBaseURL } from "@/core/config";
/**
* Share modes:
* - `import` — recipient (a logged-in user) copies the conversation into
* their own account and can keep chatting in the copy.
* - `view` — a public, read-only static page anyone can open without
* logging in; viewers cannot send new messages.
*/
export type ThreadShareMode = "import" | "view";
export interface ThreadShare {
share_code: string;
owner_user_id: string;
thread_id: string;
revoked: boolean;
mode: ThreadShareMode;
created_at: string | null;
}
/** Read-only conversation snapshot returned by the public view-share endpoint. */
export interface PublicShareSnapshot {
share_code: string;
title: string;
mode: string;
messages: Message[];
artifacts: string[];
created_at: string | null;
}
export interface SharePreview {
share_code: string;
shared_by: string;
title: string;
}
export interface ImportResult {
imported: number;
skipped: number;
}
const baseUrl = () => `${getBackendBaseURL()}/api/thread-shares`;
async function parseError(response: Response, fallback: string): Promise<Error> {
const data = (await response.json().catch(() => ({}))) as { detail?: unknown };
const detail = data.detail;
const message =
typeof detail === "string" && detail
? detail
: `${fallback} (HTTP ${response.status})`;
return new Error(message);
}
/** Create (or reuse) a share code for one conversation in the given mode. */
export async function createShare(
threadId: string,
mode: ThreadShareMode = "import",
): Promise<ThreadShare> {
const response = await apiFetch(baseUrl(), {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ thread_id: threadId, mode }),
});
if (!response.ok) throw await parseError(response, "创建分享失败");
return response.json() as Promise<ThreadShare>;
}
/** List the current user's own share links. */
export async function listShares(): Promise<ThreadShare[]> {
const response = await apiFetch(baseUrl());
if (!response.ok) throw await parseError(response, "加载分享列表失败");
const json = (await response.json()) as { shares: ThreadShare[] };
return json.shares;
}
/** Revoke a share link so it can no longer be previewed or imported. */
export async function revokeShare(shareCode: string): Promise<void> {
const response = await apiFetch(`${baseUrl()}/${encodeURIComponent(shareCode)}`, {
method: "DELETE",
});
if (!response.ok) throw await parseError(response, "撤销分享失败");
}
/** Preview a share link before importing it. */
export async function previewShare(shareCode: string): Promise<SharePreview> {
const response = await apiFetch(
`${baseUrl()}/${encodeURIComponent(shareCode)}/preview`,
);
if (!response.ok) throw await parseError(response, "无法读取该分享");
return response.json() as Promise<SharePreview>;
}
/** Import every conversation behind a share link into the current account. */
export async function importShare(shareCode: string): Promise<ImportResult> {
const response = await apiFetch(
`${baseUrl()}/${encodeURIComponent(shareCode)}/import`,
{ method: "POST" },
);
if (!response.ok) throw await parseError(response, "导入失败");
return response.json() as Promise<ImportResult>;
}
/**
* Fetch the public read-only snapshot of a `view`-mode share.
*
* Uses a plain `fetch` (no auth header) — the `/api/public/` endpoint is
* intentionally world-readable so anyone with the link can open it.
*/
export async function getPublicShareSnapshot(
shareCode: string,
): Promise<PublicShareSnapshot> {
const response = await fetch(
`${getBackendBaseURL()}/api/public/shares/${encodeURIComponent(shareCode)}`,
);
if (!response.ok) throw await parseError(response, "无法打开该分享");
return response.json() as Promise<PublicShareSnapshot>;
}
/** Build the shareable public URL for a `view`-mode share code. */
export function buildPublicShareUrl(shareCode: string): string {
// The app uses HashRouter, so the public viewer lives at `#/share/<code>`.
return `${window.location.origin}${window.location.pathname}#/share/${shareCode}`;
}
/**
* Extract a share code from arbitrary pasted text — accepts either a raw code
* or a URL/link that contains one. Share codes are 6-12 chars from the
* unambiguous alphabet (no 0/1/O/I/L).
*/
export function extractShareCode(input: string): string {
const trimmed = input.trim();
const matches = trimmed.toUpperCase().match(/[A-HJ-NP-Z2-9]{6,12}/g);
return matches && matches.length > 0 ? matches[matches.length - 1] : trimmed;
}
@@ -0,0 +1,61 @@
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
import {
createShare,
importShare,
listShares,
previewShare,
revokeShare,
type ThreadShareMode,
} from "./api";
export function useShares(enabled = true) {
return useQuery({
queryKey: ["thread-shares"],
queryFn: listShares,
enabled,
});
}
export function useCreateShare() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: ({
threadId,
mode = "import",
}: {
threadId: string;
mode?: ThreadShareMode;
}) => createShare(threadId, mode),
onSuccess: () => {
void queryClient.invalidateQueries({ queryKey: ["thread-shares"] });
},
});
}
export function useRevokeShare() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: (shareCode: string) => revokeShare(shareCode),
onSuccess: () => {
void queryClient.invalidateQueries({ queryKey: ["thread-shares"] });
},
});
}
export function usePreviewShare() {
return useMutation({
mutationFn: (shareCode: string) => previewShare(shareCode),
});
}
export function useImportShare() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: (shareCode: string) => importShare(shareCode),
onSuccess: () => {
// Imported conversations show up in the user's own thread list.
void queryClient.invalidateQueries({ queryKey: ["threads"] });
},
});
}
@@ -0,0 +1,440 @@
"use client";
import { useEffect, useMemo, useState } from "react";
import {
Edit3,
Eye,
Pause,
Play,
RefreshCw,
Search,
Trash2,
} from "lucide-react";
import { useRouteNavigate as useNavigate } from "@/core/route-base";
import { toast } from "sonner";
import { Badge } from "@/components/ui/badge";
import { Button } from "@/components/ui/button";
import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
import {
Dialog,
DialogContent,
DialogFooter,
DialogHeader,
DialogTitle,
} from "@/components/ui/dialog";
import { Input } from "@/components/ui/input";
import { ScrollArea } from "@/components/ui/scroll-area";
import { Textarea } from "@/components/ui/textarea";
import {
WorkspaceBody,
WorkspaceContainer,
WorkspaceHeader,
} from "@/components/workspace/workspace-container";
import {
buildCron,
DEFAULT_SCHEDULE_CONFIG,
parseCron,
ScheduleBuilder,
type ScheduleConfig,
statusLabel,
statusVariant,
} from "@/components/workspace/scheduled-tasks/schedule-builder";
import {
type ScheduledTask,
useAdminScheduledTaskActions,
useAdminScheduledTaskRuns,
useAllScheduledTasks,
} from "@/core/scheduled-tasks";
import { cn } from "@/lib/utils";
function formatDate(value?: string | null) {
return value ? new Date(value).toLocaleString() : "暂无";
}
/** Render a creator-like identifier without the email domain part. */
function displayUsername(value?: string | null): string {
const raw = (value ?? "").trim();
if (!raw) return "";
const at = raw.indexOf("@");
return at === -1 ? raw : raw.slice(0, at);
}
const emptyForm = {
name: "",
timezone: "Asia/Shanghai",
prompt: "",
};
export default function AdminScheduledTasksPage() {
const navigate = useNavigate();
const [searchInput, setSearchInput] = useState("");
const [search, setSearch] = useState("");
const { data: tasks = [], isLoading } = useAllScheduledTasks({ search });
const actions = useAdminScheduledTaskActions();
const [selectedTaskId, setSelectedTaskId] = useState<string | null>(null);
const [editing, setEditing] = useState<ScheduledTask | null>(null);
const [form, setForm] = useState(emptyForm);
const [scheduleConfig, setScheduleConfig] = useState<ScheduleConfig>(
DEFAULT_SCHEDULE_CONFIG,
);
const sortedTasks = useMemo(
() =>
[...tasks].sort((a, b) =>
String(a.next_run_at ?? "").localeCompare(String(b.next_run_at ?? "")),
),
[tasks],
);
const selectedTask =
sortedTasks.find((task) => task.task_id === selectedTaskId) ?? null;
const { data: runs = [], isLoading: runsLoading } =
useAdminScheduledTaskRuns(selectedTaskId);
useEffect(() => {
if (!selectedTaskId && sortedTasks[0]) {
setSelectedTaskId(sortedTasks[0].task_id);
}
if (
selectedTaskId &&
!sortedTasks.some((task) => task.task_id === selectedTaskId)
) {
setSelectedTaskId(sortedTasks[0]?.task_id ?? null);
}
}, [selectedTaskId, sortedTasks]);
const openEdit = (task: ScheduledTask) => {
setEditing(task);
setForm({
name: task.name,
timezone: task.timezone,
prompt: task.prompt,
});
setScheduleConfig(
parseCron(task.cron_expr) ??
parseCron(task.schedule_text) ??
DEFAULT_SCHEDULE_CONFIG,
);
};
const closeDialog = () => {
setEditing(null);
setForm(emptyForm);
setScheduleConfig(DEFAULT_SCHEDULE_CONFIG);
};
const saveTask = async () => {
if (!editing) return;
try {
await actions.update.mutateAsync({
taskId: editing.task_id,
data: {
name: form.name,
schedule: buildCron(scheduleConfig),
timezone: form.timezone,
prompt: form.prompt,
},
});
toast.success("任务已更新");
closeDialog();
} catch (error) {
toast.error(error instanceof Error ? error.message : "保存失败");
}
};
const toggleEnabled = async (task: ScheduledTask) => {
try {
await actions.update.mutateAsync({
taskId: task.task_id,
data: { enabled: !task.enabled },
});
toast.success(task.enabled ? "任务已暂停" : "任务已启用");
} catch (error) {
toast.error(error instanceof Error ? error.message : "操作失败");
}
};
const removeTask = async (task: ScheduledTask) => {
if (!window.confirm(`确认删除任务「${task.name}」?该操作不可撤销。`)) return;
try {
await actions.remove.mutateAsync(task.task_id);
toast.success("任务已删除");
} catch (error) {
toast.error(error instanceof Error ? error.message : "删除失败");
}
};
const triggerTask = async (task: ScheduledTask) => {
try {
await actions.trigger.mutateAsync(task.task_id);
toast.success("已触发执行");
} catch (error) {
toast.error(error instanceof Error ? error.message : "触发失败");
}
};
return (
<WorkspaceContainer>
<WorkspaceHeader />
<WorkspaceBody>
<div className="grid size-full min-h-0 grid-cols-1 gap-4 p-4 xl:grid-cols-[460px_minmax(0,1fr)]">
<Card className="flex min-h-0 flex-col">
<CardHeader className="gap-3">
<div>
<CardTitle>定时任务管理</CardTitle>
<p className="text-muted-foreground mt-1 text-sm">
管理员视图:查看、编辑、删除全部用户的定时任务
</p>
</div>
<form
className="flex gap-2"
onSubmit={(event) => {
event.preventDefault();
setSearch(searchInput.trim());
}}
>
<Input
value={searchInput}
onChange={(event) => setSearchInput(event.target.value)}
placeholder="按任务名称搜索"
/>
<Button type="submit" variant="outline">
<Search className="size-4" />
搜索
</Button>
</form>
</CardHeader>
<CardContent className="min-h-0 flex-1 p-0">
<ScrollArea className="size-full">
<div className="flex flex-col gap-3 p-4 pt-0">
{isLoading && (
<div className="text-muted-foreground text-sm">
正在加载任务...
</div>
)}
{!isLoading && sortedTasks.length === 0 && (
<div className="text-muted-foreground rounded-md border p-4 text-sm">
没有匹配的定时任务。
</div>
)}
{sortedTasks.map((task) => (
<button
key={task.task_id}
className={cn(
"bg-card text-card-foreground rounded-md border p-4 text-left transition-colors hover:bg-accent/60",
selectedTaskId === task.task_id &&
"border-primary bg-accent",
)}
onClick={() => setSelectedTaskId(task.task_id)}
>
<div className="flex items-start justify-between gap-3">
<div className="min-w-0">
<div className="truncate font-medium">
{task.name}
</div>
<div className="text-muted-foreground mt-1 truncate text-xs">
{task.schedule_text} · {task.timezone}
</div>
</div>
<Badge variant={task.enabled ? "default" : "secondary"}>
{task.enabled ? "启用" : "暂停"}
</Badge>
</div>
<div className="text-muted-foreground mt-3 grid gap-1 text-xs">
<div>
创建者:
{displayUsername(task.creator_email) ||
task.creator_user_id ||
"未知"}
</div>
<div>下次执行:{formatDate(task.next_run_at)}</div>
<div>上次执行:{formatDate(task.last_run_at)}</div>
<div>最近状态:{statusLabel(task.last_status)}</div>
</div>
<div className="mt-3 flex flex-wrap gap-2">
<Button
size="icon"
variant="outline"
title="编辑"
onClick={(event) => {
event.stopPropagation();
openEdit(task);
}}
>
<Edit3 className="size-4" />
</Button>
<Button
size="icon"
variant="outline"
title={task.enabled ? "暂停" : "启用"}
onClick={(event) => {
event.stopPropagation();
void toggleEnabled(task);
}}
>
{task.enabled ? (
<Pause className="size-4" />
) : (
<Play className="size-4" />
)}
</Button>
<Button
size="icon"
variant="outline"
title="立即执行"
onClick={(event) => {
event.stopPropagation();
void triggerTask(task);
}}
>
<RefreshCw className="size-4" />
</Button>
<Button
size="icon"
variant="destructive"
title="删除"
onClick={(event) => {
event.stopPropagation();
void removeTask(task);
}}
>
<Trash2 className="size-4" />
</Button>
</div>
</button>
))}
</div>
</ScrollArea>
</CardContent>
</Card>
<Card className="flex min-h-0 flex-col">
<CardHeader className="gap-2">
<CardTitle>执行记录</CardTitle>
<p className="text-muted-foreground text-sm">
{selectedTask ? selectedTask.name : "请选择左侧任务"}
</p>
</CardHeader>
<CardContent className="min-h-0 flex-1 p-0">
<ScrollArea className="size-full">
<div className="flex flex-col gap-3 p-4 pt-0">
{runsLoading && (
<div className="text-muted-foreground text-sm">
正在加载执行记录...
</div>
)}
{!selectedTask && (
<div className="text-muted-foreground rounded-md border p-4 text-sm">
请先选择一个定时任务。
</div>
)}
{!runsLoading && selectedTask && runs.length === 0 && (
<div className="text-muted-foreground rounded-md border p-4 text-sm">
这个任务还没有执行记录。
</div>
)}
{runs.map((run) => (
<div
key={run.id}
className="rounded-md border p-4 transition-colors hover:bg-accent/40"
>
<div className="flex flex-wrap items-start justify-between gap-3">
<div className="min-w-0">
<div className="truncate font-medium">
{run.result?.title || formatDate(run.scheduled_for)}
</div>
<div className="text-muted-foreground mt-1 text-xs">
计划:{formatDate(run.scheduled_for)} · 开始:
{formatDate(run.started_at)}
</div>
</div>
<div className="flex shrink-0 items-center gap-2">
{(run.retry_count ?? 0) > 0 && (
<Badge variant="outline" className="text-xs">
已重试 {run.retry_count}
</Badge>
)}
<Badge variant={statusVariant(run.status)}>
{statusLabel(run.status)}
</Badge>
</div>
</div>
{run.error && (
<div className="text-destructive mt-3 line-clamp-2 text-sm">
{run.error}
</div>
)}
<div className="mt-4 flex flex-wrap gap-2">
<Button
size="sm"
variant="outline"
onClick={() =>
navigate(
`/page/workspace/scheduled/runs/${run.id}`,
)
}
>
<Eye className="size-4" />
查看结果
</Button>
</div>
</div>
))}
</div>
</ScrollArea>
</CardContent>
</Card>
</div>
</WorkspaceBody>
<Dialog open={!!editing} onOpenChange={(open) => !open && closeDialog()}>
<DialogContent className="max-w-2xl">
<DialogHeader>
<DialogTitle>编辑定时任务</DialogTitle>
</DialogHeader>
<div className="flex flex-col gap-3">
<Input
value={form.name}
onChange={(event) =>
setForm((value) => ({ ...value, name: event.target.value }))
}
placeholder="任务名称"
/>
<div className="flex flex-col gap-1">
<p className="text-muted-foreground text-xs">执行频率</p>
<ScheduleBuilder
value={scheduleConfig}
onChange={setScheduleConfig}
/>
</div>
<Input
value={form.timezone}
onChange={(event) =>
setForm((value) => ({
...value,
timezone: event.target.value,
}))
}
placeholder="时区,例如:Asia/Shanghai"
/>
<Textarea
className="min-h-40"
value={form.prompt}
onChange={(event) =>
setForm((value) => ({ ...value, prompt: event.target.value }))
}
placeholder="任务 Prompt"
/>
</div>
<DialogFooter>
<Button variant="outline" onClick={closeDialog}>
取消
</Button>
<Button onClick={saveTask}>保存</Button>
</DialogFooter>
</DialogContent>
</Dialog>
</WorkspaceContainer>
);
}
+19 -1
View File
@@ -1,10 +1,13 @@
"use client";
import { DownloadIcon } from "lucide-react";
import Link from "next/link";
import { useEffect, useMemo, useState } from "react";
import { Button } from "@/components/ui/button";
import { Input } from "@/components/ui/input";
import { ScrollArea } from "@/components/ui/scroll-area";
import { ImportConversationsDialog } from "@/components/workspace/thread-share-dialogs";
import {
WorkspaceBody,
WorkspaceContainer,
@@ -20,6 +23,7 @@ export default function ChatsPage() {
const { t } = useI18n();
const { data: threads } = useThreads();
const [search, setSearch] = useState("");
const [importOpen, setImportOpen] = useState(false);
useEffect(() => {
document.title = `${t.pages.chats} - ${t.pages.appName}`;
@@ -38,7 +42,17 @@ export default function ChatsPage() {
<WorkspaceHeader></WorkspaceHeader>
<WorkspaceBody>
<div className="flex size-full flex-col">
<header className="flex shrink-0 items-center justify-center pt-8">
<header className="flex shrink-0 flex-col items-center gap-3 pt-8">
<div className="flex w-full max-w-(--container-width-md) items-center justify-end">
<Button
variant="outline"
size="sm"
onClick={() => setImportOpen(true)}
>
<DownloadIcon className="size-4" />
{t.threadShare.importChats}
</Button>
</div>
<Input
type="search"
className="h-12 w-full max-w-(--container-width-md) text-xl"
@@ -70,6 +84,10 @@ export default function ChatsPage() {
</main>
</div>
</WorkspaceBody>
<ImportConversationsDialog
open={importOpen}
onOpenChange={setImportOpen}
/>
</WorkspaceContainer>
);
}
+105
View File
@@ -0,0 +1,105 @@
"use client";
import type { BaseStream } from "@langchain/langgraph-sdk/react";
import { useQuery } from "@tanstack/react-query";
import { Loader2 } from "lucide-react";
import { useMemo } from "react";
import { useParams } from "react-router-dom";
import { ArtifactsProvider } from "@/components/workspace/artifacts";
import { MessageList } from "@/components/workspace/messages";
import { ThreadContext } from "@/components/workspace/messages/context";
import { SubtasksProvider } from "@/core/tasks/context";
import { getPublicShareSnapshot } from "@/core/thread-shares/api";
import type { AgentThreadState } from "@/core/threads";
/**
* Public, read-only conversation viewer.
*
* Reachable at `#/share/<code>` with NO authentication — anyone with the link
* can open it. It renders a frozen snapshot of a `view`-mode shared
* conversation; there is no input box, so viewers cannot send new messages.
*/
export default function PublicSharePage() {
const { shareCode } = useParams<{ shareCode: string }>();
const { data, isLoading, isError } = useQuery({
queryKey: ["public-share", shareCode],
queryFn: () => getPublicShareSnapshot(shareCode!),
enabled: !!shareCode,
retry: false,
refetchOnWindowFocus: false,
});
// A minimal static stand-in for the live LangGraph stream. MessageList only
// reads `messages`, `values.artifacts`, `isLoading` and `isThreadLoading`.
const thread = useMemo(() => {
const messages = data?.messages ?? [];
return {
messages,
values: {
title: data?.title ?? "",
messages,
artifacts: data?.artifacts ?? [],
todos: [],
},
isLoading: false,
isThreadLoading: false,
error: undefined,
interrupt: undefined,
} as unknown as BaseStream<AgentThreadState>;
}, [data]);
if (isLoading) {
return (
<div className="bg-background text-muted-foreground flex h-screen items-center justify-center gap-2 text-sm">
<Loader2 className="size-4 animate-spin" />
正在加载分享内容…
</div>
);
}
if (isError || !data) {
return (
<div className="bg-background flex h-screen flex-col items-center justify-center gap-2 px-6 text-center">
<p className="text-lg font-medium">分享不可用</p>
<p className="text-muted-foreground text-sm">
该分享链接不存在、已被撤销,或不是只读分享。
</p>
</div>
);
}
return (
<div className="bg-background flex h-screen flex-col">
<header className="flex shrink-0 flex-col gap-1 border-b px-6 py-3">
<div className="flex items-center gap-2">
<h1 className="truncate text-base font-semibold">
{data.title || "未命名对话"}
</h1>
<span className="bg-muted text-muted-foreground shrink-0 rounded-full px-2 py-0.5 text-xs">
只读分享
</span>
</div>
<p className="text-muted-foreground text-xs">
这是一个只读对话快照,你可以查看全部内容,但无法继续提问。
</p>
</header>
<main className="min-h-0 flex-1">
<SubtasksProvider>
<ArtifactsProvider>
<ThreadContext.Provider
value={{ thread, isMock: true, publicShareCode: shareCode }}
>
<MessageList
threadId={shareCode ?? "public-share"}
thread={thread}
paddingBottom={48}
/>
</ThreadContext.Provider>
</ArtifactsProvider>
</SubtasksProvider>
</main>
</div>
);
}
+9 -213
View File
@@ -51,199 +51,17 @@ import {
useScheduledTasks,
usePublishedScheduledTasks,
} from "@/core/scheduled-tasks";
import {
buildCron,
DEFAULT_SCHEDULE_CONFIG,
parseCron,
ScheduleBuilder,
type ScheduleConfig,
statusLabel,
statusVariant,
} from "@/components/workspace/scheduled-tasks/schedule-builder";
import { cn } from "@/lib/utils";
// ---- 时间配置 ----
type ScheduleFreq = "hourly" | "daily" | "weekly" | "monthly";
interface ScheduleConfig {
freq: ScheduleFreq;
minute: number;
hour: number;
weekday: number;
date: number;
}
const DEFAULT_SCHEDULE_CONFIG: ScheduleConfig = {
freq: "daily",
minute: 0,
hour: 9,
weekday: 1,
date: 1,
};
const WEEKDAYS = ["周日", "周一", "周二", "周三", "周四", "周五", "周六"];
const HOURS = Array.from({ length: 24 }, (_, i) => i);
const MINUTES = [0, 15, 30, 45];
const DATES = Array.from({ length: 31 }, (_, i) => i + 1);
function buildCron(cfg: ScheduleConfig): string {
// Keep the minute slot when the user picked one for daily/weekly/monthly too,
// so "每天 9:30" doesn't silently truncate to 9:00 on save → reload round trips.
const m = Number.isFinite(cfg.minute) ? cfg.minute : 0;
switch (cfg.freq) {
case "hourly":
return `${m} * * * *`;
case "daily":
return `${m} ${cfg.hour} * * *`;
case "weekly":
return `${m} ${cfg.hour} * * ${cfg.weekday}`;
case "monthly":
return `${m} ${cfg.hour} ${cfg.date} * *`;
}
}
// Parse a 5-field cron expression into the form picker state.
//
// Bug we keep getting bitten by: the edit dialog used to call
// `parseCron(task.schedule_text)` and `schedule_text` may be the raw user
// input (free-text like "每天 9 点") rather than a real cron string. The
// fallback `DEFAULT_SCHEDULE_CONFIG` then made the picker silently
// reinterpret the task as "every day at 09:00" — or, when natural-language
// resolved to "0 * * * *" on the backend, the picker honestly reflected
// that as "每小时 0 分". Either way the user lost their original intent.
//
// Caller now prefers `task.cron_expr` (the canonical schedule produced by
// `_schedule_to_cron`); this function stays the single source of truth for
// cron → picker, and also handles the looser step-syntax variants
// (e.g. "0 [STAR]/3 * * *") we sometimes generate.
function parseCron(cron: string | null | undefined): ScheduleConfig | null {
if (!cron) return null;
const parts = cron.trim().split(/\s+/);
if (parts.length !== 5) return null;
const [min, hr, date, , wd] = parts;
const minute = parseInt(min ?? "", 10);
const hour = parseInt(hr ?? "", 10);
// Hourly: ``M * * * *`` — match before the daily branch so a minute-only
// cron isn't misread as daily.
if (hr === "*" && date === "*" && wd === "*") {
if (!isNaN(minute)) {
return { ...DEFAULT_SCHEDULE_CONFIG, freq: "hourly", minute };
}
}
// ``0 */N * * *`` (every N hours): map to hourly with that minute; the
// picker doesn't model "every N hours" yet, so this is the safest fallback
// that at least doesn't silently round to 9:00.
if ((hr ?? "").startsWith("*/") && date === "*" && wd === "*") {
if (!isNaN(minute)) {
return { ...DEFAULT_SCHEDULE_CONFIG, freq: "hourly", minute };
}
}
if (date === "*" && wd === "*") {
if (!isNaN(hour)) {
return {
...DEFAULT_SCHEDULE_CONFIG,
freq: "daily",
hour,
minute: isNaN(minute) ? 0 : minute,
};
}
}
if (date === "*" && wd !== "*") {
const weekday = parseInt(wd ?? "", 10);
if (!isNaN(hour) && !isNaN(weekday)) {
return {
...DEFAULT_SCHEDULE_CONFIG,
freq: "weekly",
hour,
weekday,
minute: isNaN(minute) ? 0 : minute,
};
}
}
if (wd === "*" && date !== "*") {
const d = parseInt(date ?? "", 10);
if (!isNaN(hour) && !isNaN(d)) {
return {
...DEFAULT_SCHEDULE_CONFIG,
freq: "monthly",
hour,
date: d,
minute: isNaN(minute) ? 0 : minute,
};
}
}
return null;
}
function ScheduleBuilder({
value,
onChange,
}: {
value: ScheduleConfig;
onChange: (cfg: ScheduleConfig) => void;
}) {
const set = (patch: Partial<ScheduleConfig>) => onChange({ ...value, ...patch });
return (
<div className="flex flex-wrap gap-2">
<Select value={value.freq} onValueChange={(v) => set({ freq: v as ScheduleFreq })}>
<SelectTrigger className="w-28">
<SelectValue />
</SelectTrigger>
<SelectContent>
<SelectItem value="hourly">每小时</SelectItem>
<SelectItem value="daily">每天</SelectItem>
<SelectItem value="weekly">每周</SelectItem>
<SelectItem value="monthly">每月</SelectItem>
</SelectContent>
</Select>
{value.freq === "hourly" && (
<Select value={String(value.minute)} onValueChange={(v) => set({ minute: parseInt(v) })}>
<SelectTrigger className="w-24">
<SelectValue />
</SelectTrigger>
<SelectContent>
{MINUTES.map((m) => (
<SelectItem key={m} value={String(m)}>{m} 分</SelectItem>
))}
</SelectContent>
</Select>
)}
{value.freq === "weekly" && (
<Select value={String(value.weekday)} onValueChange={(v) => set({ weekday: parseInt(v) })}>
<SelectTrigger className="w-24">
<SelectValue />
</SelectTrigger>
<SelectContent>
{WEEKDAYS.map((d, i) => (
<SelectItem key={i} value={String(i)}>{d}</SelectItem>
))}
</SelectContent>
</Select>
)}
{value.freq === "monthly" && (
<Select value={String(value.date)} onValueChange={(v) => set({ date: parseInt(v) })}>
<SelectTrigger className="w-24">
<SelectValue />
</SelectTrigger>
<SelectContent>
{DATES.map((d) => (
<SelectItem key={d} value={String(d)}>第 {d} 日</SelectItem>
))}
</SelectContent>
</Select>
)}
{value.freq !== "hourly" && (
<Select value={String(value.hour)} onValueChange={(v) => set({ hour: parseInt(v) })}>
<SelectTrigger className="w-24">
<SelectValue />
</SelectTrigger>
<SelectContent>
{HOURS.map((h) => (
<SelectItem key={h} value={String(h)}>{h} 时</SelectItem>
))}
</SelectContent>
</Select>
)}
</div>
);
}
function formatDate(value?: string | null) {
return value ? new Date(value).toLocaleString() : "暂无";
}
@@ -256,28 +74,6 @@ function displayUsername(value?: string | null): string {
return at === -1 ? raw : raw.slice(0, at);
}
function statusLabel(status?: string | null) {
switch (status) {
case "succeeded":
return "成功";
case "failed":
return "失败";
case "running":
return "执行中";
case "pending":
return "等待中";
default:
return status || "未知";
}
}
function statusVariant(status?: string | null): "default" | "secondary" | "destructive" | "outline" {
if (status === "succeeded") return "default";
if (status === "failed") return "destructive";
if (status === "running") return "secondary";
return "outline";
}
const emptyForm = {
name: "",
schedule: "",
@@ -18,6 +18,7 @@ import ChatPage from "./ChatPage";
import AgentsPage from "./AgentsPage";
import AgentChatPage from "./AgentChatPage";
import NewAgentPage from "./NewAgentPage";
import AdminScheduledTasksPage from "./AdminScheduledTasksPage";
import ScheduledTaskRunDetailPage from "./ScheduledTaskRunDetailPage";
import ScheduledTasksPage from "./ScheduledTasksPage";
import LlmMetricsPage from "./LlmMetricsPage";
@@ -184,6 +185,18 @@ export default function WorkspaceRoutes({
</WorkspaceLayout>
}
/>
<Route
path="admin/scheduled-tasks"
element={
getStoredAuth()?.system_role === "admin" ? (
<WorkspaceLayout {...layoutProps}>
<AdminScheduledTasksPage />
</WorkspaceLayout>
) : (
<Navigate to="/page/workspace/chats/new" replace />
)
}
/>
<Route
path="admin/tags"
element={
+10 -1
View File
@@ -214,7 +214,7 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
|--------|-----------|
| **Models** (`/api/models`) | `GET /` - list models; `GET /{name}` - model details |
| **MCP** (`/api/mcp`) | `GET /config` - get config; `PUT /config` - update config (saves to extensions_config.json) |
| **Skills** (`/api/skills`) | `GET /` - list skills (optional `search` fuzzy-name + `tag_id` filters; each item carries `tags` + owner-authored `detail`); `GET /{name}` - details; `PUT /{name}` - update enabled; `PUT /{name}/detail` - set the long-form `detail` text (custom skill: owner or admin; built-in/public: admin only — materializes a `skills` row on demand); `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`) |
| **Skills** (`/api/skills`) | `GET /` - list skills (optional `search` fuzzy-name + `tag_id` filters; each item carries `tags` + owner-authored `detail`); `GET /{name}` - details; `PUT /{name}` - update enabled; `PUT /{name}/detail` - set the long-form `detail` text (custom skill: owner or admin; built-in/public: admin only — materializes a `skills` row on demand); `GET /{name}/content` - raw SKILL.md source, restricted to an admin or the publisher (custom-skill owner); other users get 403 and see only description + detail; `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`) |
| **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data |
| **Agents** (`/api/agents`) | `GET /` - list built-in, owned, and published agents (optional `search` + `tag_id` filters; each item carries `tags`); `POST /` - create user-owned agent; `GET/PUT/DELETE /{id}` - read/update/delete by stable agent id; `GET /check` - validate non-empty display name (duplicates allowed) |
| **Tags** (`/api/tags`) | `GET /` - list tags (optional `tag_type` filter; open to all); `POST /` - create tag (admin); `PUT /{id}` / `DELETE /{id}` - update/delete tag (admin); `GET /assignments/{target_type}/{target_id}` - tags on a resource; `PUT /assignments/{target_type}/{target_id}` - replace a resource's tags (admin). Tags are typed `agent`/`skill`/`scheduled_task` and may only tag resources of their own type. |
@@ -225,9 +225,18 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA
| **Thread Runs** (`/api/threads/{id}/runs`) | `POST /` - create background run; `POST /stream` - create + SSE stream; `POST /wait` - create + block; `GET /` - list runs; `GET /{rid}` - run details; `POST /{rid}/cancel` - cancel; `GET /{rid}/join` - join SSE; `GET /{rid}/messages` - paginated messages `{data, has_more}`; `GET /{rid}/events` - full event stream; `GET /../messages` - thread messages with feedback; `GET /../token-usage` - aggregate tokens |
| **Feedback** (`/api/threads/{id}/runs/{rid}/feedback`) | `PUT /` - upsert feedback; `DELETE /` - delete user feedback; `POST /` - create feedback; `GET /` - list feedback; `GET /stats` - aggregate stats; `DELETE /{fid}` - delete specific |
| **Runs** (`/api/runs`) | `POST /stream` - stateless run + SSE; `POST /wait` - stateless run + block; `GET /{rid}/messages` - paginated messages by run_id `{data, has_more}` (cursor: `after_seq`/`before_seq`); `GET /{rid}/feedback` - list feedback by run_id |
| **Thread Shares** (`/api/thread-shares`) | Per-conversation sharing with two modes (`thread_shares.mode`). **`import`**: `POST /` `{thread_id, mode}` - create (or reuse) a revocable share code; `GET /{code}/preview` + `POST /{code}/import` - copy that conversation into the caller's account (fresh `thread_id` via `app/gateway/thread_copy.py` + new `threads_meta` row + `user-data` copy; dedup on `metadata.shared_origin`). **`view`**: a public, read-only static page — see Public Shares below. Both modes: `GET /` - list my codes; `DELETE /{code}` - revoke. |
| **Public Shares** (`/api/public/shares`) | **Unauthenticated** (the `/api/public/` prefix is whitelisted in `auth_middleware._PUBLIC_PATH_PREFIXES`). `GET /{code}` - read-only conversation snapshot (title + serialized messages + artifacts) for a `view`-mode share; messages are read straight from the latest checkpoint. `GET /{code}/files/{path}` - serves an **image** referenced by that conversation (images only — non-image paths 404; SVG forced to download). Missing/revoked/non-`view` codes all return 404. |
| **Scheduled Tasks** (`/api/scheduled-tasks`) | Per-user cron-like agent tasks. `GET/POST /` - list/create owned tasks; `PATCH/DELETE /{id}` - update/delete (owner); `POST /{id}/{pause,resume,trigger}`; `GET /{id}/runs`, `GET/DELETE /runs/{rid}`, run-file preview/edit/download; publish/subscribe under `/{id}/{publish,unpublish,subscribe}` and `GET /published`. **Admin (cross-user):** `GET /admin/all` - every task; `PATCH/DELETE /admin/{id}` - edit/delete any task; `POST /admin/{id}/trigger`; `GET /admin/{id}/runs`. Admin routes require `system_role == "admin"`. |
Proxied through nginx: `/api/langgraph/*` → LangGraph, all other `/api/*` → Gateway.
### Scheduled Task Runtime
`ScheduledTaskService` (`deerflow/runtime/scheduler/service.py`) polls due tasks every `poll_interval_seconds` and executes each in its own throwaway thread, serialized by a shared `_agent_run_lock`.
**Stuck-run recovery**: each execution attempt is bounded by `_RUN_ATTEMPT_TIMEOUT_SECONDS` (30 min) via `asyncio.wait_for`. On timeout the run is retried up to `_MAX_RUN_RETRIES` (3) times, then marked `failed`. A watchdog (`_reconcile_stuck_runs`, run every poll) recovers runs left in `running` with no in-process owner — e.g. orphaned by a crash or restart — re-driving them through the same retry budget so a task can never stay perpetually "executing". The `scheduled_task_runs.retry_count` column tracks consumed retries.
### Custom Agent Persistence
Custom agent metadata lives in the `agents` ORM table (`deerflow.persistence.agents`). `agents.id` is the stable runtime id and the filesystem directory name under `.deer-flow/agents/{id}/`; `agents.name` is display-only, can contain Chinese characters, and is not unique. `user_id IS NULL` marks built-in agents, user-owned agents are visible only to their owner unless `published = true`, and update/delete operations require ownership. The runtime accepts `agent_id` in context/configurable, and non-default `assistant_id` values are treated as agent ids after visibility checks.
@@ -41,6 +41,7 @@ from app.gateway.routers import (
system_settings,
tags,
thread_runs,
thread_shares,
threads,
uploads,
)
@@ -701,6 +702,10 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
# Thread Runs API (LangGraph Platform-compatible runs lifecycle)
app.include_router(thread_runs.router)
app.include_router(thread_shares.router)
# Public, unauthenticated read-only share snapshots (/api/public/shares/*).
app.include_router(thread_shares.public_router)
# Stateless Runs API (stream/wait without a pre-existing thread)
app.include_router(runs.router)
@@ -23,11 +23,16 @@ from app.gateway.internal_auth import INTERNAL_AUTH_HEADER_NAME, get_internal_us
from deerflow.runtime.user_context import reset_current_user, set_current_user
# Paths that never require authentication.
#
# ``/api/public/`` is a namespace reserved for intentionally world-readable,
# read-only endpoints (e.g. public conversation share snapshots). Only mount
# safe read-only routes under it — everything there bypasses auth entirely.
_PUBLIC_PATH_PREFIXES: tuple[str, ...] = (
"/health",
"/docs",
"/redoc",
"/openapi.json",
"/api/public/",
)
# Exact auth paths that are public (login/register/status check).
@@ -122,6 +122,10 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
app.state.tag_store = make_tag_store(sf)
from deerflow.persistence.thread_shares import make_thread_share_store
app.state.thread_share_store = make_thread_share_store(sf)
# Run event store (has its own factory with config-driven backend selection)
run_events_config = getattr(config, "run_events", None)
app.state.run_event_store = make_run_event_store(run_events_config)
@@ -222,6 +226,14 @@ def get_tag_store(request: Request) -> TagStore:
return val
def get_thread_share_store(request: Request):
"""Return the conversation share-link store."""
val = getattr(request.app.state, "thread_share_store", None)
if val is None:
raise HTTPException(status_code=503, detail="Thread share store not available")
return val
def get_run_context(request: Request) -> RunContext:
"""Build a :class:`RunContext` from ``app.state`` singletons.
@@ -14,7 +14,7 @@ from fastapi import APIRouter, HTTPException, Query, Request
from fastapi.responses import FileResponse, Response, StreamingResponse
from pydantic import BaseModel, Field
from app.gateway.deps import get_current_user, get_local_provider, get_tag_store
from app.gateway.deps import get_current_user, get_local_provider, get_optional_user_from_request, get_tag_store
from deerflow.runtime.scheduler import get_scheduled_task_service
router = APIRouter(prefix="/api/scheduled-tasks", tags=["scheduled-tasks"])
@@ -61,6 +61,26 @@ async def _user_id(request: Request) -> str:
return user_id
async def _is_admin(request: Request) -> bool:
"""True when the caller is an admin (or auth is disabled entirely)."""
user = await get_optional_user_from_request(request)
if user is None:
return True
return getattr(user, "system_role", None) == "admin"
async def _load_run(service, request: Request, run_id: str, user_id: str) -> dict[str, Any] | None:
"""Owner/subscriber run lookup, with an admin fallback to any run.
Lets admins open run results for tasks they do not own (the admin
management view links into the shared run-detail page).
"""
run = await _load_run(service, request, run_id, user_id)
if run is None and await _is_admin(request):
run = await service.admin_get_run(run_id)
return run
async def _enrich_with_creator(tasks: list[dict[str, Any]] | dict[str, Any] | None) -> Any:
"""Attach ``creator_email`` / ``creator_user_id`` to one or many task dicts.
@@ -208,7 +228,7 @@ async def list_scheduled_task_runs(task_id: str, request: Request, limit: int =
async def get_scheduled_task_run(run_id: str, request: Request) -> dict:
service = get_scheduled_task_service()
user_id = await _user_id(request)
run = await service.get_run(user_id, run_id)
run = await _load_run(service, request, run_id, user_id)
if run is None:
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
return run
@@ -231,7 +251,7 @@ async def delete_scheduled_task_run(run_id: str, request: Request) -> dict:
async def get_scheduled_task_run_file(run_id: str, file_index: int, request: Request) -> dict:
service = get_scheduled_task_service()
user_id = await _user_id(request)
run = await service.get_run(user_id, run_id)
run = await _load_run(service, request, run_id, user_id)
if run is None:
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
item, path = _run_file_path(run, file_index)
@@ -279,7 +299,7 @@ async def update_scheduled_task_run_file(
async def download_scheduled_task_run(run_id: str, request: Request):
service = get_scheduled_task_service()
user_id = await _user_id(request)
run = await service.get_run(user_id, run_id)
run = await _load_run(service, request, run_id, user_id)
if run is None:
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
@@ -328,7 +348,7 @@ async def download_scheduled_task_run(run_id: str, request: Request):
async def download_scheduled_task_run_file(run_id: str, file_index: int, request: Request):
service = get_scheduled_task_service()
user_id = await _user_id(request)
run = await service.get_run(user_id, run_id)
run = await _load_run(service, request, run_id, user_id)
if run is None:
raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
item, path = _run_file_path(run, file_index)
@@ -497,3 +517,82 @@ async def delete_scheduled_task_subscription(task_id: str, request: Request) ->
if not deleted:
raise HTTPException(status_code=404, detail="Subscription not found")
return {"success": True}
# --- Admin (cross-user management) ---
async def _require_admin(request: Request) -> None:
"""Reject non-admins. Lets the call through when auth is disabled."""
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="定时任务管理仅限管理员")
@router.get("/admin/all")
async def admin_list_scheduled_tasks(
request: Request,
search: str | None = Query(default=None, description="Case-insensitive fuzzy match on task name"),
) -> list[dict]:
"""List every scheduled task across all users. Admin-only."""
await _require_admin(request)
service = get_scheduled_task_service()
tasks = await service.admin_list_tasks()
await _enrich_with_creator(tasks)
tag_store = get_tag_store(request)
assignments = await tag_store.list_assignments("scheduled_task", [str(t.get("task_id")) for t in tasks])
for task in tasks:
task["tags"] = assignments.get(str(task.get("task_id")), [])
if search:
needle = search.strip().lower()
tasks = [t for t in tasks if needle in str(t.get("name", "")).lower()]
return tasks
@router.get("/admin/{task_id}/runs")
async def admin_list_scheduled_task_runs(task_id: str, request: Request, limit: int = 50) -> list[dict]:
"""List runs for any task regardless of owner. Admin-only."""
await _require_admin(request)
service = get_scheduled_task_service()
return await service.admin_list_runs(task_id, limit=max(1, min(limit, 200)))
@router.patch("/admin/{task_id}")
async def admin_update_scheduled_task(task_id: str, body: ScheduledTaskUpdateRequest, request: Request) -> dict:
"""Update any task regardless of owner. Admin-only."""
await _require_admin(request)
service = get_scheduled_task_service()
task = await service.admin_update_task(task_id, body.model_dump(exclude_unset=True))
if task is None:
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
await _enrich_with_creator(task)
return task
@router.delete("/admin/{task_id}")
async def admin_delete_scheduled_task(task_id: str, request: Request) -> dict:
"""Delete any task regardless of owner. Admin-only."""
await _require_admin(request)
service = get_scheduled_task_service()
deleted = await service.admin_delete_task(task_id)
if not deleted:
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
try:
await get_tag_store(request).unassign_all("scheduled_task", task_id)
except Exception: # noqa: BLE001 — tag cleanup must not block deletion
pass
return {"success": True}
@router.post("/admin/{task_id}/trigger")
async def admin_trigger_scheduled_task(task_id: str, request: Request) -> dict:
"""Trigger an immediate run of any task regardless of owner. Admin-only."""
await _require_admin(request)
service = get_scheduled_task_service()
task = await service.admin_trigger_task(task_id)
if task is None:
raise HTTPException(status_code=404, detail=f"Task {task_id} not found")
return {"success": True, "task": task}
@@ -45,6 +45,24 @@ def _is_admin_user(request: Request) -> bool:
return getattr(user, "system_role", None) == "admin"
def _can_view_skill_source(
category: SkillCategory,
owner_user_id: str | None,
user_id: str,
*,
is_admin: bool,
) -> bool:
"""Whether the raw SKILL.md source is visible to a user.
Admins may view any skill's source; otherwise only the publisher (the
owner of a custom skill) qualifies. Built-in / public skills have no human
owner, so for non-admins they are never source-visible.
"""
if is_admin:
return True
return category == SkillCategory.CUSTOM and owner_user_id is not None and owner_user_id == user_id
async def _get_owned_skill_or_raise(store: SkillStore, name: str, user_id: str) -> dict[str, Any]:
owned = await store.get_owned(name, user_id)
if owned is not None:
@@ -870,15 +888,27 @@ class SkillContentResponse(BaseModel):
"/skills/{skill_name}/content",
response_model=SkillContentResponse,
summary="Get Skill Markdown Content",
description="Retrieve the raw SKILL.md content of a specific skill (public or custom).",
description="Retrieve the raw SKILL.md content of a skill. Only an admin or the skill's publisher (owner of a custom skill) may read it; other users see only the description and detail.",
)
async def get_skill_content(skill_name: str, config: AppConfig = Depends(get_config)) -> SkillContentResponse:
async def get_skill_content(
skill_name: str,
request: Request,
config: AppConfig = Depends(get_config),
skill_store: SkillStore = Depends(get_skill_store),
) -> SkillContentResponse:
try:
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
skills = get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False)
skill = next((s for s in skills if s.name == skill_name), None)
if skill is None:
raise HTTPException(status_code=404, detail=f"Skill '{skill_name}' not found")
# The raw SKILL.md is implementation detail — only an admin or the
# publisher (owner of a custom skill) may view it. Everyone else is
# limited to the description + detail surfaced by the list/detail APIs.
record = await skill_store.get_any(skill_name)
owner_user_id = record.get("owner_user_id") if record else None
if not _can_view_skill_source(skill.category, owner_user_id, _current_user_id(request), is_admin=_is_admin_user(request)):
raise HTTPException(status_code=403, detail="Only the skill publisher or an admin can view the SKILL.md source")
content = skill.skill_file.read_text(encoding="utf-8")
return SkillContentResponse(name=skill.name, category=skill.category, content=content)
except HTTPException:
@@ -0,0 +1,340 @@
"""Conversation sharing API.
A user creates a share code for *one* of their conversations; redeeming the
code copies that conversation into the redeemer's account as an independent
copy they fully own and can keep chatting in. See ``app/gateway/thread_copy.py``
for the checkpoint duplication.
"""
from __future__ import annotations
import logging
import mimetypes
import shutil
import uuid
from pathlib import Path
from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field
from app.gateway.deps import get_checkpointer, get_thread_share_store, get_thread_store
from app.gateway.thread_copy import copy_thread_state
from deerflow.config.paths import get_paths
from deerflow.persistence.thread_meta.base import ThreadMetaStore
from deerflow.persistence.thread_shares.base import ThreadShareStore
from deerflow.runtime import serialize_channel_values
from deerflow.runtime.user_context import get_effective_user_id
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/thread-shares", tags=["thread-shares"])
# Public, unauthenticated router. Its prefix is whitelisted in
# ``auth_middleware._PUBLIC_PATH_PREFIXES`` — only mount read-only endpoints
# that are intentionally world-readable here.
public_router = APIRouter(prefix="/api/public/shares", tags=["thread-shares-public"])
_UNTITLED = "未命名对话"
# Valid share modes. "import" copies the conversation into the recipient's
# account; "view" is a public read-only static page anyone can open.
_SHARE_MODES = ("import", "view")
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()
# ---------------------------------------------------------------------------
# Request / response models
# ---------------------------------------------------------------------------
class CreateShareRequest(BaseModel):
thread_id: str = Field(..., description="The conversation to share")
mode: str = Field(
default="import",
description='"import" — recipient copies the conversation; "view" — public read-only page',
)
class ShareResponse(BaseModel):
share_code: str = Field(..., description="Short code that recipients enter to import")
owner_user_id: str = Field(..., description="User who created the share")
thread_id: str = Field(..., description="The shared conversation")
revoked: bool = Field(default=False, description="Whether the code has been revoked")
mode: str = Field(default="import", description='"import" or "view"')
created_at: str | None = Field(default=None, description="Creation timestamp (ISO-8601)")
class PublicShareSnapshotResponse(BaseModel):
"""Read-only conversation snapshot served to anonymous viewers."""
share_code: str
title: str = Field(..., description="Conversation title")
mode: str = Field(default="view")
messages: list[dict] = Field(default_factory=list, description="Serialized conversation messages")
artifacts: list[str] = Field(default_factory=list, description="Artifact virtual paths produced in the conversation")
created_at: str | None = Field(default=None, description="When the share was created (ISO-8601)")
class SharesListResponse(BaseModel):
shares: list[ShareResponse]
class SharePreviewResponse(BaseModel):
share_code: str
shared_by: str = Field(..., description="Owner user id of the shared conversation")
title: str = Field(..., description="Title of the shared conversation")
class ImportResultResponse(BaseModel):
imported: int = Field(..., description="1 when the conversation was copied, else 0")
skipped: int = Field(..., description="1 when it was already imported, else 0")
# ---------------------------------------------------------------------------
# Owner side — create / list / revoke
# ---------------------------------------------------------------------------
@router.post("", response_model=ShareResponse, summary="Create a share code for one conversation")
async def create_share(
body: CreateShareRequest,
request: Request,
share_store: ThreadShareStore = Depends(get_thread_share_store),
thread_store: ThreadMetaStore = Depends(get_thread_store),
) -> ShareResponse:
me = _current_user_id(request)
mode = body.mode if body.mode in _SHARE_MODES else "import"
# Only the conversation's owner may share it.
meta = await thread_store.get(body.thread_id, user_id=me)
if meta is None:
raise HTTPException(status_code=404, detail="Conversation not found")
# Reuse an existing active code (of the same mode) so re-sharing is idempotent.
existing = await share_store.get_active_for_thread(me, body.thread_id, mode=mode)
if existing is not None:
return ShareResponse(**existing)
record = await share_store.create(me, body.thread_id, mode=mode)
return ShareResponse(**record)
@router.get("", response_model=SharesListResponse, summary="List my conversation share codes")
async def list_shares(
request: Request,
share_store: ThreadShareStore = Depends(get_thread_share_store),
) -> SharesListResponse:
records = await share_store.list_for_owner(_current_user_id(request))
return SharesListResponse(shares=[ShareResponse(**r) for r in records])
@router.delete("/{share_code}", summary="Revoke a conversation share code")
async def revoke_share(
share_code: str,
request: Request,
share_store: ThreadShareStore = Depends(get_thread_share_store),
) -> dict[str, bool]:
ok = await share_store.revoke(share_code, _current_user_id(request))
if not ok:
raise HTTPException(status_code=404, detail="Share code not found")
return {"success": True}
# ---------------------------------------------------------------------------
# Recipient side — preview / import
# ---------------------------------------------------------------------------
async def _resolve_active_share(share_code: str, share_store: ThreadShareStore) -> dict:
share = await share_store.get(share_code)
if share is None:
raise HTTPException(status_code=404, detail="Share code not found")
if share.get("revoked"):
raise HTTPException(status_code=410, detail="This share code has been revoked")
return share
@router.get("/{share_code}/preview", response_model=SharePreviewResponse, summary="Preview a shared conversation")
async def preview_share(
share_code: str,
share_store: ThreadShareStore = Depends(get_thread_share_store),
thread_store: ThreadMetaStore = Depends(get_thread_store),
) -> SharePreviewResponse:
share = await _resolve_active_share(share_code, share_store)
owner = share["owner_user_id"]
meta = await thread_store.get(share["thread_id"], user_id=owner)
if meta is None:
raise HTTPException(status_code=404, detail="The shared conversation no longer exists")
return SharePreviewResponse(
share_code=share_code,
shared_by=owner,
title=meta.get("display_name") or _UNTITLED,
)
def _copy_thread_files(origin_thread_id: str, owner_user_id: str, new_thread_id: str, recipient_user_id: str) -> None:
"""Copy a thread's user-data directory (uploads + outputs/artifacts)."""
paths = get_paths()
src_data = paths.thread_dir(origin_thread_id, user_id=owner_user_id) / "user-data"
if not src_data.is_dir():
return
dst_data = paths.thread_dir(new_thread_id, user_id=recipient_user_id) / "user-data"
shutil.copytree(src_data, dst_data, dirs_exist_ok=True)
@router.post("/{share_code}/import", response_model=ImportResultResponse, summary="Import a shared conversation")
async def import_share(
share_code: str,
request: Request,
share_store: ThreadShareStore = Depends(get_thread_share_store),
thread_store: ThreadMetaStore = Depends(get_thread_store),
) -> ImportResultResponse:
share = await _resolve_active_share(share_code, share_store)
if share.get("mode") == "view":
raise HTTPException(status_code=400, detail="This is a view-only share — open its link to read it instead of importing")
owner = share["owner_user_id"]
origin_id = share["thread_id"]
recipient = _current_user_id(request)
if owner == recipient:
raise HTTPException(status_code=400, detail="You cannot import your own share code")
src = await thread_store.get(origin_id, user_id=owner)
if src is None:
raise HTTPException(status_code=404, detail="The shared conversation no longer exists")
# Dedup: skip if this user already imported this conversation.
mine = await thread_store.search(user_id=recipient, limit=2000)
already = {(t.get("metadata") or {}).get("shared_origin") for t in mine}
if origin_id in already:
return ImportResultResponse(imported=0, skipped=1)
new_id = str(uuid.uuid4())
checkpointer = get_checkpointer(request)
await copy_thread_state(checkpointer, origin_id, new_id)
metadata = dict(src.get("metadata") or {})
metadata.update({"shared_from_code": share_code, "shared_origin": origin_id, "shared_by": owner})
await thread_store.create(
new_id,
assistant_id=src.get("assistant_id"),
user_id=recipient,
display_name=src.get("display_name"),
metadata=metadata,
)
try:
_copy_thread_files(origin_id, owner, new_id, recipient)
except Exception as exc: # noqa: BLE001 — file copy failure must not lose the conversation
logger.warning("Failed to copy files for thread %s -> %s: %s", origin_id, new_id, exc)
logger.info("Share %s imported by %s as thread %s", share_code, recipient, new_id)
return ImportResultResponse(imported=1, skipped=0)
# ---------------------------------------------------------------------------
# Public read-only view — no authentication required
# ---------------------------------------------------------------------------
async def _load_thread_snapshot(checkpointer, thread_id: str) -> tuple[list[dict], list[str]]:
"""Return (messages, artifacts) from a thread's latest checkpoint.
Reads straight from the checkpointer — the same source the live chat
streams from — so the static snapshot matches the conversation exactly.
"""
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
checkpoint_tuple = await checkpointer.aget_tuple(config)
if checkpoint_tuple is None:
return [], []
checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
values = serialize_channel_values(checkpoint.get("channel_values", {}) or {})
messages = values.get("messages")
artifacts = values.get("artifacts")
return (
[m for m in messages if isinstance(m, dict)] if isinstance(messages, list) else [],
[a for a in artifacts if isinstance(a, str)] if isinstance(artifacts, list) else [],
)
@public_router.get(
"/{share_code}",
response_model=PublicShareSnapshotResponse,
summary="Public read-only snapshot of a view-shared conversation",
)
async def get_public_share_snapshot(
share_code: str,
share_store: ThreadShareStore = Depends(get_thread_share_store),
thread_store: ThreadMetaStore = Depends(get_thread_store),
checkpointer=Depends(get_checkpointer),
) -> PublicShareSnapshotResponse:
"""Serve a frozen, read-only conversation. Anyone with the link can read it.
Only ``mode="view"`` shares are exposed; missing, revoked, or import-mode
codes all return 404 so the endpoint never reveals which case applies.
"""
share = await share_store.get(share_code)
if share is None or share.get("revoked") or share.get("mode") != "view":
raise HTTPException(status_code=404, detail="Share not found")
meta = await thread_store.get(share["thread_id"], user_id=share["owner_user_id"])
if meta is None:
raise HTTPException(status_code=404, detail="The shared conversation no longer exists")
messages, artifacts = await _load_thread_snapshot(checkpointer, share["thread_id"])
return PublicShareSnapshotResponse(
share_code=share_code,
title=meta.get("display_name") or _UNTITLED,
mode="view",
messages=messages,
artifacts=artifacts,
created_at=share.get("created_at"),
)
@public_router.get(
"/{share_code}/files/{path:path}",
summary="Public image attachment from a view-shared conversation",
)
async def get_public_share_file(
share_code: str,
path: str,
share_store: ThreadShareStore = Depends(get_thread_share_store),
) -> FileResponse:
"""Serve an image referenced by a view-shared conversation.
Restricted to images only — the public share page renders inline images,
but the endpoint must not become a way to read arbitrary thread files.
Non-image paths return 404 just like a missing file.
"""
share = await share_store.get(share_code)
if share is None or share.get("revoked") or share.get("mode") != "view":
raise HTTPException(status_code=404, detail="Share not found")
# Resolve the virtual path inside the *owner's* thread directory.
try:
actual = get_paths().resolve_virtual_path(share["thread_id"], path, user_id=share["owner_user_id"])
except ValueError as exc:
status = 403 if "traversal" in str(exc) else 400
raise HTTPException(status_code=status, detail=str(exc))
actual_path = Path(actual)
if not actual_path.exists() or not actual_path.is_file():
raise HTTPException(status_code=404, detail="File not found")
mime_type, _ = mimetypes.guess_type(actual_path.name)
if not mime_type or not mime_type.startswith("image/"):
# Only images are public; everything else is treated as not found.
raise HTTPException(status_code=404, detail="File not found")
# SVG can carry scripts — serve it as a download so it can never execute
# as a top-level document in the app origin. Raster images render inline.
is_svg = mime_type == "image/svg+xml"
return FileResponse(
actual_path,
media_type=mime_type,
content_disposition_type="attachment" if is_svg else "inline",
headers={"Cache-Control": "public, max-age=300"},
)
@@ -0,0 +1,67 @@
"""Copy a conversation's LangGraph checkpoint state from one thread to another.
Used by the conversation-sharing feature: redeeming a share link duplicates the
sharer's threads into the redeemer's account. Each duplicated thread needs its
full checkpoint history copied under a fresh ``thread_id``.
The copy goes through the checkpointer's public API (``alist`` / ``aput`` /
``aput_writes``) rather than raw SQL, so it is independent of the checkpoint
table schema and works for every backend (SQLite, Postgres, in-memory).
"""
from __future__ import annotations
import logging
logger = logging.getLogger(__name__)
async def copy_thread_state(checkpointer, src_thread_id: str, dst_thread_id: str) -> bool:
"""Duplicate every checkpoint of ``src_thread_id`` under ``dst_thread_id``.
Returns True when at least one checkpoint was copied, False when the source
thread has no checkpoints (e.g. a thread that was never run).
The destination keeps the source's checkpoint ids — they only need to be
unique within a ``thread_id``, and reusing them preserves the parent chain.
"""
src_config = {"configurable": {"thread_id": src_thread_id}}
# alist yields newest-first; reverse so parents are written before children.
tuples = [t async for t in checkpointer.alist(src_config)]
if not tuples:
return False
copied = 0
for tup in reversed(tuples):
configurable = (tup.config or {}).get("configurable", {})
checkpoint_ns = configurable.get("checkpoint_ns", "") or ""
parent_id = ((tup.parent_config or {}).get("configurable", {}) or {}).get("checkpoint_id")
put_config: dict = {"configurable": {"thread_id": dst_thread_id, "checkpoint_ns": checkpoint_ns}}
if parent_id:
# aput reads the parent checkpoint id from the incoming config.
put_config["configurable"]["checkpoint_id"] = parent_id
new_versions = tup.checkpoint.get("channel_versions", {}) or {}
await checkpointer.aput(put_config, tup.checkpoint, tup.metadata or {}, new_versions)
# pending_writes is a list of (task_id, channel, value); aput_writes
# wants them grouped per task as (channel, value) pairs.
if tup.pending_writes:
by_task: dict[str, list] = {}
for task_id, channel, value in tup.pending_writes:
by_task.setdefault(task_id, []).append((channel, value))
checkpoint_id = tup.checkpoint.get("id")
for task_id, writes in by_task.items():
writes_config = {
"configurable": {
"thread_id": dst_thread_id,
"checkpoint_ns": checkpoint_ns,
"checkpoint_id": checkpoint_id,
}
}
await checkpointer.aput_writes(writes_config, writes, task_id)
copied += 1
logger.info("Copied %d checkpoint(s) from thread %s to %s", copied, src_thread_id, dst_thread_id)
return copied > 0
@@ -0,0 +1,53 @@
"""Create thread_shares table (per-conversation share links).
Revision ID: 20260518_02
Revises: 20260518_01
Create Date: 2026-05-18
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260518_02"
down_revision = "20260518_01"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
inspector = inspect(op.get_bind())
return table_name in inspector.get_table_names()
def _has_column(table_name: str, column_name: str) -> bool:
if not _has_table(table_name):
return False
inspector = inspect(op.get_bind())
return any(col["name"] == column_name for col in inspector.get_columns(table_name))
def upgrade() -> None:
if not _has_table("thread_shares"):
op.create_table(
"thread_shares",
sa.Column("share_code", sa.String(32), primary_key=True),
sa.Column("owner_user_id", sa.String(64), nullable=False),
sa.Column("thread_id", sa.String(64), nullable=False),
sa.Column("revoked", sa.Boolean(), nullable=False, server_default=sa.false()),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
)
op.create_index("ix_thread_shares_owner_user_id", "thread_shares", ["owner_user_id"])
op.create_index("ix_thread_shares_thread_id", "thread_shares", ["thread_id"])
elif not _has_column("thread_shares", "thread_id"):
# An earlier all-conversations variant of this table — add the column.
op.add_column("thread_shares", sa.Column("thread_id", sa.String(64), nullable=False, server_default=""))
op.create_index("ix_thread_shares_thread_id", "thread_shares", ["thread_id"])
def downgrade() -> None:
if _has_table("thread_shares"):
op.drop_table("thread_shares")
@@ -0,0 +1,57 @@
"""Ensure thread_shares.thread_id exists (per-conversation share links).
Standalone migration so databases that already applied an earlier
all-conversations variant of ``20260518_02`` (which had no ``thread_id``
column) get the column added. Idempotent — a no-op when the column is
already present.
Revision ID: 20260518_03
Revises: 20260518_02
Create Date: 2026-05-18
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260518_03"
down_revision = "20260518_02"
branch_labels = None
depends_on = None
def _has_table(table_name: str) -> bool:
inspector = inspect(op.get_bind())
return table_name in inspector.get_table_names()
def _has_column(table_name: str, column_name: str) -> bool:
if not _has_table(table_name):
return False
inspector = inspect(op.get_bind())
return any(col["name"] == column_name for col in inspector.get_columns(table_name))
def _has_index(table_name: str, index_name: str) -> bool:
if not _has_table(table_name):
return False
inspector = inspect(op.get_bind())
return any(ix["name"] == index_name for ix in inspector.get_indexes(table_name))
def upgrade() -> None:
if not _has_table("thread_shares"):
# Fresh database — 20260518_02 already created the table with thread_id.
return
if not _has_column("thread_shares", "thread_id"):
op.add_column("thread_shares", sa.Column("thread_id", sa.String(64), nullable=False, server_default=""))
if not _has_index("thread_shares", "ix_thread_shares_thread_id"):
op.create_index("ix_thread_shares_thread_id", "thread_shares", ["thread_id"])
def downgrade() -> None:
# Leave the column in place — dropping it would break the share feature.
pass
@@ -0,0 +1,39 @@
"""Add retry_count to scheduled_task_runs.
Tracks how many recovery attempts a run has consumed so the scheduler can
cap retries on stuck/timed-out runs.
Revision ID: 20260518_04
Revises: 20260518_03
Create Date: 2026-05-18
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260518_04"
down_revision = "20260518_03"
branch_labels = None
depends_on = None
def _has_column(table_name: str, column_name: str) -> bool:
inspector = inspect(op.get_bind())
return any(column["name"] == column_name for column in inspector.get_columns(table_name))
def upgrade() -> None:
if not _has_column("scheduled_task_runs", "retry_count"):
op.add_column(
"scheduled_task_runs",
sa.Column("retry_count", sa.Integer(), nullable=False, server_default="0"),
)
def downgrade() -> None:
if _has_column("scheduled_task_runs", "retry_count"):
op.drop_column("scheduled_task_runs", "retry_count")
@@ -0,0 +1,44 @@
"""Add mode to thread_shares.
Distinguishes "import" shares (recipient copies the conversation into their
account) from "view" shares (a public, read-only static page).
Revision ID: 20260518_05
Revises: 20260518_04
Create Date: 2026-05-18
"""
from __future__ import annotations
import sqlalchemy as sa
from alembic import op
from sqlalchemy import inspect
revision = "20260518_05"
down_revision = "20260518_04"
branch_labels = None
depends_on = None
def _has_column(table_name: str, column_name: str) -> bool:
inspector = inspect(op.get_bind())
if table_name not in inspector.get_table_names():
return True # table missing entirely — skip (created with the column)
return any(column["name"] == column_name for column in inspector.get_columns(table_name))
def upgrade() -> None:
if not _has_column("thread_shares", "mode"):
op.add_column(
"thread_shares",
sa.Column("mode", sa.String(16), nullable=False, server_default="import"),
)
def downgrade() -> None:
inspector = inspect(op.get_bind())
if "thread_shares" in inspector.get_table_names() and any(
column["name"] == "mode" for column in inspector.get_columns("thread_shares")
):
op.drop_column("thread_shares", "mode")
@@ -32,6 +32,7 @@ from deerflow.persistence.scheduled_tasks.model import (
from deerflow.persistence.skills.model import SkillRow
from deerflow.persistence.tags.model import TagAssignmentRow, TagRow
from deerflow.persistence.thread_meta.model import ThreadMetaRow
from deerflow.persistence.thread_shares.model import ThreadShareRow
from deerflow.persistence.user.model import UserRow
__all__ = [
@@ -51,5 +52,6 @@ __all__ = [
"TagAssignmentRow",
"TagRow",
"ThreadMetaRow",
"ThreadShareRow",
"UserRow",
]
@@ -36,6 +36,35 @@ class ScheduledTaskStore(ABC):
async def due_tasks(self, now: datetime, *, limit: int = 20) -> list[dict[str, Any]]:
raise NotImplementedError
# --- Admin (cross-user) ---
@abstractmethod
async def list_all_tasks(self) -> list[dict[str, Any]]:
"""Return every task across all users. Admin-only callers."""
raise NotImplementedError
@abstractmethod
async def update_task_any(self, task_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
"""Update a task by id without restricting by owner. Admin-only callers."""
raise NotImplementedError
@abstractmethod
async def delete_task_any(self, task_id: str) -> bool:
"""Delete a task by id without restricting by owner. Admin-only callers."""
raise NotImplementedError
@abstractmethod
async def list_runs_any(self, task_id: str, *, limit: int = 50) -> list[dict[str, Any]]:
"""Return runs for a task without restricting by owner. Admin-only callers."""
raise NotImplementedError
@abstractmethod
async def list_stuck_runs(self, cutoff: datetime, *, limit: int = 50) -> list[dict[str, Any]]:
"""Return runs still in a non-terminal state whose ``started_at`` is older
than ``cutoff``. Used by the scheduler watchdog to recover runs that hung
or were orphaned by a service restart."""
raise NotImplementedError
@abstractmethod
async def create_run_once(self, data: dict[str, Any]) -> dict[str, Any] | None:
raise NotImplementedError
@@ -59,6 +59,37 @@ class MemoryScheduledTaskStore(ScheduledTaskStore):
due.sort(key=lambda task: task["next_run_at"])
return due[:limit]
# --- Admin (cross-user) ---
async def list_all_tasks(self) -> list[dict[str, Any]]:
return [dict(task) for task in self._tasks.values()]
async def update_task_any(self, task_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
task = self._tasks.get(task_id)
if task is None:
return None
task.update(data)
return dict(task)
async def delete_task_any(self, task_id: str) -> bool:
return self._tasks.pop(task_id, None) is not None
async def list_runs_any(self, task_id: str, *, limit: int = 50) -> list[dict[str, Any]]:
runs = [dict(run) for run in self._runs.values() if run["task_id"] == task_id]
runs.sort(key=lambda run: run.get("scheduled_for"), reverse=True)
return runs[:limit]
async def list_stuck_runs(self, cutoff: datetime, *, limit: int = 50) -> list[dict[str, Any]]:
stuck = [
dict(run)
for run in self._runs.values()
if run.get("status") in ("running", "pending")
and run.get("started_at") is not None
and run["started_at"] <= cutoff
]
stuck.sort(key=lambda run: run["started_at"])
return stuck[:limit]
async def create_run_once(self, data: dict[str, Any]) -> dict[str, Any] | None:
key = (data["task_id"], data["scheduled_for"].isoformat())
if key in self._runs_by_key:
@@ -4,7 +4,7 @@ from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import JSON, Boolean, DateTime, String, Text, UniqueConstraint
from sqlalchemy import JSON, Boolean, DateTime, Integer, String, Text, UniqueConstraint
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
@@ -72,6 +72,10 @@ class ScheduledTaskRunRow(Base):
status: Mapped[str] = mapped_column(String(32), default="pending")
error: Mapped[str | None] = mapped_column(Text)
result_json: Mapped[dict | None] = mapped_column(JSON, default=dict)
# Number of recovery attempts made after the first try. A run is retried
# when it overruns the per-attempt timeout or is found orphaned in the
# "running" state (e.g. after a service restart). See ScheduledTaskService.
retry_count: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC))
updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC))
@@ -124,6 +124,60 @@ class ScheduledTaskRepository(ScheduledTaskStore):
)
return [_to_dict(row) for row in result.scalars()]
# --- Admin (cross-user) ---
async def list_all_tasks(self) -> list[dict[str, Any]]:
async with self._sf() as session:
result = await session.execute(select(ScheduledTaskRow).order_by(ScheduledTaskRow.created_at.desc()))
return [_to_dict(row) for row in result.scalars()]
async def update_task_any(self, task_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(ScheduledTaskRow, task_id)
if row is None:
return None
for key, value in data.items():
attr = "execution_context_json" if key == "execution_context" else key
if hasattr(row, attr):
setattr(row, attr, value)
row.updated_at = datetime.now(UTC)
await session.commit()
await session.refresh(row)
return _to_dict(row)
async def delete_task_any(self, task_id: str) -> bool:
async with self._sf() as session:
row = await session.get(ScheduledTaskRow, task_id)
if row is None:
return False
await session.delete(row)
await session.commit()
return True
async def list_runs_any(self, task_id: str, *, limit: int = 50) -> list[dict[str, Any]]:
async with self._sf() as session:
result = await session.execute(
select(ScheduledTaskRunRow)
.where(ScheduledTaskRunRow.task_id == task_id)
.order_by(ScheduledTaskRunRow.scheduled_for.desc())
.limit(limit)
)
return [_to_dict(row) for row in result.scalars()]
async def list_stuck_runs(self, cutoff: datetime, *, limit: int = 50) -> list[dict[str, Any]]:
async with self._sf() as session:
result = await session.execute(
select(ScheduledTaskRunRow)
.where(
ScheduledTaskRunRow.status.in_(("running", "pending")),
ScheduledTaskRunRow.started_at.is_not(None),
ScheduledTaskRunRow.started_at <= cutoff,
)
.order_by(ScheduledTaskRunRow.started_at.asc())
.limit(limit)
)
return [_to_dict(row) for row in result.scalars()]
async def create_run_once(self, data: dict[str, Any]) -> dict[str, Any] | None:
row = ScheduledTaskRunRow(
id=data["id"],
@@ -0,0 +1,19 @@
"""Persistence layer for conversation-history share links."""
from deerflow.persistence.thread_shares.base import ThreadShareStore, generate_share_code
from deerflow.persistence.thread_shares.memory import MemoryThreadShareStore
from deerflow.persistence.thread_shares.sql import ThreadShareRepository
__all__ = [
"ThreadShareStore",
"MemoryThreadShareStore",
"ThreadShareRepository",
"generate_share_code",
"make_thread_share_store",
]
def make_thread_share_store(session_factory):
if session_factory is None:
return MemoryThreadShareStore()
return ThreadShareRepository(session_factory)
@@ -0,0 +1,50 @@
"""Abstract persistence API for per-conversation share links."""
from __future__ import annotations
import secrets
from abc import ABC, abstractmethod
from typing import Any
# Unambiguous alphabet — no 0/O/1/I/L so codes are easy to read and type.
_CODE_ALPHABET = "ABCDEFGHJKMNPQRSTUVWXYZ23456789"
def generate_share_code(length: int = 8) -> str:
"""Return a random, unguessable, human-readable share code."""
return "".join(secrets.choice(_CODE_ALPHABET) for _ in range(length))
class ThreadShareStore(ABC):
@abstractmethod
async def create(self, owner_user_id: str, thread_id: str, *, mode: str = "import") -> dict[str, Any]:
"""Create a new share link for one conversation and return its record.
``mode`` is ``"import"`` (recipient copies the conversation) or
``"view"`` (a public read-only static page).
"""
raise NotImplementedError
@abstractmethod
async def get(self, share_code: str) -> dict[str, Any] | None:
"""Look up a share by code regardless of owner (None if missing)."""
raise NotImplementedError
@abstractmethod
async def get_active_for_thread(self, owner_user_id: str, thread_id: str, *, mode: str = "import") -> dict[str, Any] | None:
"""Return the owner's non-revoked share of ``mode`` for ``thread_id``, if any.
Used so re-sharing the same conversation in the same mode reuses its
existing code instead of minting a new one each time.
"""
raise NotImplementedError
@abstractmethod
async def list_for_owner(self, owner_user_id: str) -> list[dict[str, Any]]:
"""List all share links created by ``owner_user_id`` (newest first)."""
raise NotImplementedError
@abstractmethod
async def revoke(self, share_code: str, owner_user_id: str) -> bool:
"""Revoke a share link. Returns False if missing or not owned."""
raise NotImplementedError
@@ -0,0 +1,59 @@
"""In-memory ``ThreadShareStore`` implementation for tests / memory backend."""
from __future__ import annotations
import copy
from datetime import UTC, datetime
from typing import Any
from deerflow.persistence.thread_shares.base import ThreadShareStore, generate_share_code
class MemoryThreadShareStore(ThreadShareStore):
def __init__(self) -> None:
self._shares: dict[str, dict[str, Any]] = {}
async def create(self, owner_user_id: str, thread_id: str, *, mode: str = "import") -> dict[str, Any]:
code = generate_share_code()
while code in self._shares:
code = generate_share_code()
record = {
"share_code": code,
"owner_user_id": owner_user_id,
"thread_id": thread_id,
"revoked": False,
"mode": mode,
"created_at": datetime.now(UTC).isoformat(),
}
self._shares[code] = record
return copy.deepcopy(record)
async def get(self, share_code: str) -> dict[str, Any] | None:
record = self._shares.get(share_code)
return copy.deepcopy(record) if record is not None else None
async def get_active_for_thread(self, owner_user_id: str, thread_id: str, *, mode: str = "import") -> dict[str, Any] | None:
matches = [
r
for r in self._shares.values()
if r["owner_user_id"] == owner_user_id
and r["thread_id"] == thread_id
and r.get("mode", "import") == mode
and not r["revoked"]
]
if not matches:
return None
matches.sort(key=lambda r: r["created_at"], reverse=True)
return copy.deepcopy(matches[0])
async def list_for_owner(self, owner_user_id: str) -> list[dict[str, Any]]:
rows = [r for r in self._shares.values() if r["owner_user_id"] == owner_user_id]
rows.sort(key=lambda r: r["created_at"], reverse=True)
return [copy.deepcopy(r) for r in rows]
async def revoke(self, share_code: str, owner_user_id: str) -> bool:
record = self._shares.get(share_code)
if record is None or record["owner_user_id"] != owner_user_id:
return False
record["revoked"] = True
return True
@@ -0,0 +1,36 @@
"""ORM model for conversation share links.
A share link is a short, revocable code created by a user for *one* of their
conversations. Redeeming it copies that single conversation into the redeemer's
account. The table only stores the link itself — the conversation is copied
lazily at import time.
"""
from __future__ import annotations
from datetime import UTC, datetime
from sqlalchemy import Boolean, DateTime, String
from sqlalchemy.orm import Mapped, mapped_column
from deerflow.persistence.base import Base
class ThreadShareRow(Base):
__tablename__ = "thread_shares"
# Short, URL-safe, unguessable code embedded in the share link.
share_code: Mapped[str] = mapped_column(String(32), primary_key=True)
# The user who created the share.
owner_user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
# The single conversation (thread) this link shares.
thread_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
# Soft-disable: a revoked link can no longer be previewed or imported.
revoked: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
# Share mode:
# "import" — recipient (a logged-in user) copies the conversation into
# their own account and can keep chatting in the copy.
# "view" — a public, read-only static page anyone can open without
# logging in; viewers cannot send new messages.
mode: Mapped[str] = mapped_column(String(16), nullable=False, default="import", server_default="import")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC))
@@ -0,0 +1,80 @@
"""SQLAlchemy-backed per-conversation share-link repository."""
from __future__ import annotations
from datetime import datetime
from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from deerflow.persistence.thread_shares.base import ThreadShareStore, generate_share_code
from deerflow.persistence.thread_shares.model import ThreadShareRow
def _row_to_dict(row: ThreadShareRow) -> dict[str, Any]:
data = row.to_dict()
created = data.get("created_at")
if isinstance(created, datetime):
data["created_at"] = created.isoformat()
return data
class ThreadShareRepository(ThreadShareStore):
def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
self._sf = session_factory
async def create(self, owner_user_id: str, thread_id: str, *, mode: str = "import") -> dict[str, Any]:
# Retry on the (astronomically unlikely) code collision.
async with self._sf() as session:
for _ in range(8):
code = generate_share_code()
if await session.get(ThreadShareRow, code) is not None:
continue
row = ThreadShareRow(
share_code=code,
owner_user_id=owner_user_id,
thread_id=thread_id,
revoked=False,
mode=mode,
)
session.add(row)
await session.commit()
await session.refresh(row)
return _row_to_dict(row)
raise RuntimeError("Failed to allocate a unique share code")
async def get(self, share_code: str) -> dict[str, Any] | None:
async with self._sf() as session:
row = await session.get(ThreadShareRow, share_code)
return _row_to_dict(row) if row is not None else None
async def get_active_for_thread(self, owner_user_id: str, thread_id: str, *, mode: str = "import") -> dict[str, Any] | None:
stmt = (
select(ThreadShareRow)
.where(
ThreadShareRow.owner_user_id == owner_user_id,
ThreadShareRow.thread_id == thread_id,
ThreadShareRow.mode == mode,
ThreadShareRow.revoked.is_(False),
)
.order_by(ThreadShareRow.created_at.desc())
)
async with self._sf() as session:
row = (await session.execute(stmt)).scalars().first()
return _row_to_dict(row) if row is not None else None
async def list_for_owner(self, owner_user_id: str) -> list[dict[str, Any]]:
stmt = select(ThreadShareRow).where(ThreadShareRow.owner_user_id == owner_user_id).order_by(ThreadShareRow.created_at.desc())
async with self._sf() as session:
result = await session.execute(stmt)
return [_row_to_dict(row) for row in result.scalars()]
async def revoke(self, share_code: str, owner_user_id: str) -> bool:
async with self._sf() as session:
row = await session.get(ThreadShareRow, share_code)
if row is None or row.owner_user_id != owner_user_id:
return False
row.revoked = True
await session.commit()
return True
@@ -9,7 +9,7 @@ import re
import shutil
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from datetime import UTC, datetime, timedelta
from pathlib import Path
from types import SimpleNamespace
from typing import Any
@@ -27,6 +27,14 @@ from deerflow.utils.time import now_iso
logger = logging.getLogger(__name__)
# A single execution attempt may not exceed this wall-clock budget. When it
# does, the run is treated as stuck (the agent loop hung, a tool blocked, etc.)
# and retried. The watchdog uses the same window to recover runs orphaned by a
# crash or restart — those stay "running" in the DB with no in-process owner.
_RUN_ATTEMPT_TIMEOUT_SECONDS = 30 * 60 # 30 minutes
# Retries allowed after the first attempt before the run is marked "failed".
_MAX_RUN_RETRIES = 3
@dataclass
class _RuntimeUser:
@@ -398,7 +406,13 @@ class ScheduledTaskService:
)
return task
async def update_task(self, user_id: str, task_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
@staticmethod
def _build_task_update(data: dict[str, Any], existing: dict[str, Any] | None) -> dict[str, Any]:
"""Translate an API update payload into a storage update dict.
Shared by owner-scoped :meth:`update_task` and admin :meth:`admin_update_task`
so both apply identical schedule/timezone normalization.
"""
update_data: dict[str, Any] = {}
for key in ("name", "prompt", "enabled", "execution_agent_name"):
if key in data and data[key] is not None:
@@ -409,20 +423,74 @@ class ScheduledTaskService:
if data.get("schedule") is not None:
schedule = str(data["schedule"])
cron_expr = _schedule_to_cron(schedule)
tz = update_data.get("timezone")
if tz is None:
existing = await self.store.get_task(task_id, user_id)
tz = (existing or {}).get("timezone", "UTC")
tz = update_data.get("timezone") or (existing or {}).get("timezone") or "UTC"
update_data["schedule_text"] = schedule
update_data["cron_expr"] = cron_expr
update_data["next_run_at"] = _next_run_at(cron_expr, tz)
if data.get("execution_context") is not None:
update_data["execution_context"] = data["execution_context"]
# Re-enabling a task must also schedule it forward from now, otherwise a
# paused task with a stale/empty next_run_at would never fire again.
if update_data.get("enabled") is True and "next_run_at" not in update_data:
cron = (existing or {}).get("cron_expr")
if cron:
tz = update_data.get("timezone") or (existing or {}).get("timezone") or "UTC"
update_data["next_run_at"] = _next_run_at(cron, tz)
return update_data
async def update_task(self, user_id: str, task_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
existing = await self.store.get_task(task_id, user_id)
if existing is None:
return None
update_data = self._build_task_update(data, existing)
return await self.store.update_task(task_id, user_id, update_data)
async def delete_task(self, user_id: str, task_id: str) -> bool:
return await self.store.delete_task(task_id, user_id)
# --- Admin (cross-user management) ---
async def admin_list_tasks(self) -> list[dict[str, Any]]:
"""List every scheduled task across all users. Admin-only callers."""
tasks = await self.store.list_all_tasks()
for task in tasks:
task["role"] = "owner"
return tasks
async def admin_update_task(self, task_id: str, data: dict[str, Any]) -> dict[str, Any] | None:
"""Update any task regardless of owner. Admin-only callers."""
existing = await self.store.get_task_any(task_id)
if existing is None:
return None
update_data = self._build_task_update(data, existing)
return await self.store.update_task_any(task_id, update_data)
async def admin_delete_task(self, task_id: str) -> bool:
"""Delete any task regardless of owner. Admin-only callers."""
return await self.store.delete_task_any(task_id)
async def admin_list_runs(self, task_id: str, *, limit: int = 50) -> list[dict[str, Any]]:
"""List runs for any task regardless of owner. Admin-only callers."""
runs = await self.store.list_runs_any(task_id, limit=limit)
for run in runs:
run["can_edit"] = True
return runs
async def admin_get_run(self, run_id: str) -> dict[str, Any] | None:
"""Fetch a single run regardless of owner. Admin-only callers."""
run = await self.store.get_run_any(run_id)
if run is not None:
run["can_edit"] = True
return run
async def admin_trigger_task(self, task_id: str) -> dict[str, Any] | None:
"""Trigger an immediate run of any task regardless of owner."""
task = await self.store.get_task_any(task_id)
if task is None:
return None
asyncio.create_task(self._execute_tracked(task, scheduled_for=datetime.now(UTC), manual=True))
return task
async def pause_task(self, user_id: str, task_id: str) -> dict[str, Any] | None:
return await self.store.update_task(task_id, user_id, {"enabled": False})
@@ -437,11 +505,19 @@ class ScheduledTaskService:
task = await self.store.get_task(task_id, user_id)
if task is None:
return None
asyncio.create_task(self._execute_task(task, scheduled_for=datetime.now(UTC), manual=True))
asyncio.create_task(self._execute_tracked(task, scheduled_for=datetime.now(UTC), manual=True))
return task
async def _run_loop(self) -> None:
while not self._stopping.is_set():
# Recover hung / orphaned runs first so a single stuck task cannot
# silently wedge the scheduler indefinitely.
try:
await self._reconcile_stuck_runs()
except asyncio.CancelledError:
raise
except Exception:
logger.exception("Scheduled task watchdog scan failed")
try:
due = await self.store.due_tasks(datetime.now(UTC), limit=20)
for task in due:
@@ -466,64 +542,195 @@ class ScheduledTaskService:
finally:
self._running_task_ids.discard(task["task_id"])
async def _execute_task(self, task: dict[str, Any], *, scheduled_for: datetime, manual: bool) -> None:
async def _execute_tracked(self, task: dict[str, Any], *, scheduled_for: datetime, manual: bool) -> None:
"""Run a task while marking it busy so the watchdog won't double-drive it."""
self._running_task_ids.add(task["task_id"])
try:
await self._execute_task(task, scheduled_for=scheduled_for, manual=manual)
finally:
self._running_task_ids.discard(task["task_id"])
async def _reconcile_stuck_runs(self) -> None:
"""Recover runs left in the ``running`` state with no in-process owner.
A run becomes stuck when the agent loop hangs past the per-attempt
timeout, or when the process crashes / restarts mid-run (the original
asyncio task is gone but the DB row still says ``running``). For each
such run we retry up to ``_MAX_RUN_RETRIES`` times, then mark it
``failed`` so it no longer shows as perpetually executing.
"""
cutoff = datetime.now(UTC) - timedelta(seconds=_RUN_ATTEMPT_TIMEOUT_SECONDS)
stuck = await self.store.list_stuck_runs(cutoff, limit=20)
for run in stuck:
task_id = run["task_id"]
# A run still owned by a live in-process execution is handled by
# that execution's own timeout/retry loop — don't touch it here.
if task_id in self._running_task_ids:
continue
self._running_task_ids.add(task_id)
asyncio.create_task(self._recover_and_untrack(run))
async def _recover_and_untrack(self, run: dict[str, Any]) -> None:
try:
await self._recover_stuck_run(run)
except Exception:
logger.exception("Failed to recover stuck scheduled task run %s", run.get("id"))
finally:
self._running_task_ids.discard(run["task_id"])
async def _recover_stuck_run(self, run: dict[str, Any]) -> None:
run_id = run["id"]
retry_count = int(run.get("retry_count") or 0)
task = await self.store.get_task_any(run["task_id"])
scheduled_for = _coerce_datetime(run.get("scheduled_for")) or datetime.now(UTC)
if task is None:
# Parent task was deleted — just close out the orphaned run.
await self.store.update_run(
run_id,
{"status": "failed", "finished_at": datetime.now(UTC), "error": "执行超时:所属任务已被删除"},
)
return
if retry_count >= _MAX_RUN_RETRIES:
await self._fail_run(
task,
run_id,
scheduled_for,
error=f"执行超时:超过 {_RUN_ATTEMPT_TIMEOUT_SECONDS // 60} 分钟未完成,重试 {_MAX_RUN_RETRIES} 次后仍失败",
retry_count=retry_count,
)
return
logger.warning("Recovering stuck scheduled task run %s (retry budget %s/%s)", run_id, retry_count, _MAX_RUN_RETRIES)
await self._execute_task(task, scheduled_for=scheduled_for, manual=False, run_row=run)
async def _fail_run(
self,
task: dict[str, Any],
run_id: str,
scheduled_for: datetime,
*,
error: str,
retry_count: int = 0,
) -> None:
"""Mark a run failed and reflect the outcome on the parent task."""
result = _build_result(task, scheduled_for=scheduled_for, content=f"Scheduled task failed: {error}")
await self.store.update_run(
run_id,
{"status": "failed", "finished_at": datetime.now(UTC), "error": error, "result": result, "retry_count": retry_count},
)
await self.store.update_task(task["task_id"], task["user_id"], {"last_run_at": datetime.now(UTC), "last_status": "failed"})
async def _execute_task(
self,
task: dict[str, Any],
*,
scheduled_for: datetime,
manual: bool,
run_row: dict[str, Any] | None = None,
) -> None:
# Each run uses its own throwaway thread so message history cannot accumulate
# across runs and blow past the model's input window. The per-user
# task["scheduler_thread_id"] is no longer used to drive the LLM call —
# we keep it on the task row only for backwards compatibility.
runtime_thread_id = str(uuid.uuid4())
run_row = await self.store.create_run_once(
{
"id": str(uuid.uuid4()),
"task_id": task["task_id"],
"user_id": task["user_id"],
"scheduler_thread_id": runtime_thread_id,
"scheduled_for": scheduled_for,
"status": "running",
"started_at": datetime.now(UTC),
}
)
is_recovery = run_row is not None
if run_row is None:
return
run_row = await self.store.create_run_once(
{
"id": str(uuid.uuid4()),
"task_id": task["task_id"],
"user_id": task["user_id"],
"scheduler_thread_id": str(uuid.uuid4()),
"scheduled_for": scheduled_for,
"status": "running",
"started_at": datetime.now(UTC),
}
)
if run_row is None:
return
else:
# Recovery path: re-claim an existing (orphaned) run row.
await self.store.update_run(run_row["id"], {"status": "running", "started_at": datetime.now(UTC)})
if not manual:
# Advancing the schedule only happens once, when the run is first
# created — recovery retries reuse the original scheduled_for.
if not manual and not is_recovery:
next_run_at = _next_run_at(task["cron_expr"], task.get("timezone") or "UTC", base=scheduled_for)
await self.store.update_task(task["task_id"], task["user_id"], {"next_run_at": next_run_at, "last_status": "running"})
try:
await self._ensure_thread_meta(task["user_id"], runtime_thread_id)
async with self._agent_run_lock:
output_snapshot = self._snapshot_result_files(runtime_thread_id, task["user_id"])
record = await self._start_agent_run(task, scheduled_for=scheduled_for, runtime_thread_id=runtime_thread_id)
await self.store.update_run(run_row["id"], {"agent_run_id": record.run_id})
if record.task is not None:
await record.task
result_text = await self._extract_latest_ai_text(runtime_thread_id)
result_files = await self._extract_result_files(
runtime_thread_id,
task["user_id"],
run_id=run_row["id"],
output_snapshot=output_snapshot,
run_id = run_row["id"]
retry_count = int(run_row.get("retry_count") or 0)
while True:
runtime_thread_id = str(uuid.uuid4())
try:
result = await asyncio.wait_for(
self._run_agent_attempt(task, runtime_thread_id=runtime_thread_id, scheduled_for=scheduled_for, run_id=run_id),
timeout=_RUN_ATTEMPT_TIMEOUT_SECONDS,
)
result = _build_result(task, scheduled_for=scheduled_for, content=result_text, files=result_files)
except TimeoutError:
# The attempt overran the 30-minute cap. Retry a bounded number
# of times, then give up and mark the run failed.
if retry_count >= _MAX_RUN_RETRIES:
await self._fail_run(
task,
run_id,
scheduled_for,
error=f"执行超时:单次执行超过 {_RUN_ATTEMPT_TIMEOUT_SECONDS // 60} 分钟,重试 {_MAX_RUN_RETRIES} 次后仍失败",
retry_count=retry_count,
)
return
retry_count += 1
logger.warning("Scheduled task %s timed out; retry %s/%s", task["task_id"], retry_count, _MAX_RUN_RETRIES)
await self.store.update_run(
run_id,
{
"retry_count": retry_count,
"status": "running",
"started_at": datetime.now(UTC),
"error": f"执行超时,正在重试(第 {retry_count}/{_MAX_RUN_RETRIES} 次)",
},
)
continue
except Exception as exc:
logger.exception("Scheduled task %s failed", task["task_id"])
await self._fail_run(task, run_id, scheduled_for, error=str(exc), retry_count=retry_count)
return
await self.store.update_run(
run_row["id"],
{"status": "succeeded", "finished_at": datetime.now(UTC), "result": result},
run_id,
{"status": "succeeded", "finished_at": datetime.now(UTC), "result": result, "retry_count": retry_count, "error": None},
)
await self.store.update_task(task["task_id"], task["user_id"], {"last_run_at": datetime.now(UTC), "last_status": "succeeded"})
await self._deliver_element_result(task, result["content"])
except Exception as exc:
logger.exception("Scheduled task %s failed", task["task_id"])
result = _build_result(
task,
scheduled_for=scheduled_for,
content=f"Scheduled task failed: {exc}",
return
async def _run_agent_attempt(
self,
task: dict[str, Any],
*,
runtime_thread_id: str,
scheduled_for: datetime,
run_id: str,
) -> dict[str, Any]:
"""Run the agent once and build the result payload.
Wrapped in :func:`asyncio.wait_for` by the caller so a hung agent loop
cannot hold the shared run lock — and the whole scheduler — forever.
"""
await self._ensure_thread_meta(task["user_id"], runtime_thread_id)
async with self._agent_run_lock:
output_snapshot = self._snapshot_result_files(runtime_thread_id, task["user_id"])
record = await self._start_agent_run(task, scheduled_for=scheduled_for, runtime_thread_id=runtime_thread_id)
await self.store.update_run(run_id, {"agent_run_id": record.run_id})
if record.task is not None:
await record.task
result_text = await self._extract_latest_ai_text(runtime_thread_id)
result_files = await self._extract_result_files(
runtime_thread_id,
task["user_id"],
run_id=run_id,
output_snapshot=output_snapshot,
)
await self.store.update_run(
run_row["id"],
{"status": "failed", "finished_at": datetime.now(UTC), "error": str(exc), "result": result},
)
await self.store.update_task(task["task_id"], task["user_id"], {"last_run_at": datetime.now(UTC), "last_status": "failed"})
return _build_result(task, scheduled_for=scheduled_for, content=result_text, files=result_files)
async def _ensure_thread_meta(self, user_id: str, thread_id: str) -> None:
token = set_current_user(_RuntimeUser(id=user_id))
@@ -17,6 +17,7 @@ import pytest
from app.gateway.routers.skills import (
SkillDetailRequest,
SkillResponse,
_can_view_skill_source,
_skill_to_response,
)
from deerflow.config import paths as paths_module
@@ -160,3 +161,29 @@ def test_skill_to_response_detail_none_without_ownership(isolated_paths) -> None
resp = _skill_to_response(skill, None)
assert resp.detail is None
assert resp.builtin is True
# ---------------------------------------------------------------------------
# SKILL.md source visibility — admin / publisher only
# ---------------------------------------------------------------------------
def test_admin_can_view_any_skill_source() -> None:
# Admin sees a built-in skill's source (no owner).
assert _can_view_skill_source(SkillCategory.PUBLIC, None, "user-a", is_admin=True) is True
# ...and another user's custom skill.
assert _can_view_skill_source(SkillCategory.CUSTOM, "user-b", "user-a", is_admin=True) is True
def test_publisher_can_view_own_custom_skill_source() -> None:
assert _can_view_skill_source(SkillCategory.CUSTOM, "user-a", "user-a", is_admin=False) is True
def test_non_owner_cannot_view_custom_skill_source() -> None:
# A different user — even if the skill is published to them — cannot read source.
assert _can_view_skill_source(SkillCategory.CUSTOM, "user-b", "user-a", is_admin=False) is False
def test_non_admin_cannot_view_builtin_skill_source() -> None:
# Built-in / public skills have no publisher, so non-admins never see source.
assert _can_view_skill_source(SkillCategory.PUBLIC, None, "user-a", is_admin=False) is False