From 50d70a5050a53a9a43f820972c3d3c89d296a0d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Mon, 8 Jun 2026 18:00:39 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat:=20=E5=AF=B9=E8=AF=9D=E5=88=97?= =?UTF-8?q?=E8=A1=A8=E6=94=AF=E6=8C=81=E5=88=86=E9=A1=B5=E3=80=81=E6=94=B6?= =?UTF-8?q?=E8=97=8F=E3=80=81=E7=BD=AE=E9=A1=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 收藏(is_favorite)/置顶(is_pinned)作为布尔标记存入对话已有 metadata， 复用现有 PATCH /api/threads/{id} 与 search 往返，无需后端改动与迁移。 - hooks: 新增 useUpdateThreadMetadata + useToggleThreadPin/Favorite - utils: 新增 isThreadPinned/isThreadFavorite/sortThreadsPinnedFirst - 侧边栏: 置顶优先排序 + 下拉菜单收藏/置顶项 + 置顶图标 - 对话页: 置顶区 + 全部/收藏过滤 + 页码翻页(每页20) + 行内收藏/置顶按钮 - i18n: zh-CN/en-US/types 新增相关文案 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../components/workspace/recent-chat-list.tsx | 68 +++- frontend-web/src/core/i18n/locales/en-US.ts | 12 + frontend-web/src/core/i18n/locales/types.ts | 12 + frontend-web/src/core/i18n/locales/zh-CN.ts | 12 + frontend-web/src/core/threads/hooks.ts | 152 ++++++-- frontend-web/src/core/threads/utils.ts | 22 ++ frontend-web/src/pages/ChatsPage.tsx | 350 +++++++++++++----- 7 files changed, 489 insertions(+), 139 deletions(-) diff --git a/frontend-web/src/components/workspace/recent-chat-list.tsx b/frontend-web/src/components/workspace/recent-chat-list.tsx index 7def3260..46026eb8 100644 --- a/frontend-web/src/components/workspace/recent-chat-list.tsx +++ b/frontend-web/src/components/workspace/recent-chat-list.tsx @@ -7,7 +7,10 @@ import { FileText, MoreHorizontal, Pencil, + Pin, + PinOff, Share2, + Star, Trash2, } from "lucide-react"; import Link from "next/link"; @@ -54,9 +57,17 @@ import { useDeleteThread, useRenameThread, useThreads, + useToggleThreadFavorite, + useToggleThreadPin, } from "@/core/threads/hooks"; import type { AgentThread, AgentThreadState } from "@/core/threads/types"; -import { pathOfThread, titleOfThread } from "@/core/threads/utils"; +import { + isThreadFavorite, + isThreadPinned, + pathOfThread, + sortThreadsPinnedFirst, + titleOfThread, +} from "@/core/threads/utils"; import { env } from "@/env"; import { desensitize } from "@/lib/desensitize"; import { isIMEComposing } from "@/lib/ime"; @@ -82,6 +93,8 @@ export function RecentChatList() { const { data: threads = [] } = useThreads(); const { mutate: deleteThread } = useDeleteThread(); const { mutate: renameThread } = useRenameThread(); + const { toggle: togglePin } = useToggleThreadPin(); + const { toggle: toggleFavorite } = useToggleThreadFavorite(); // Rename dialog state const [renameDialogOpen, setRenameDialogOpen] = useState(false); @@ -160,11 +173,13 @@ export function RecentChatList() { [t], ); - const visibleThreads = threads.filter( - (thread) => - thread.metadata?.thread_type !== "scheduler" && - thread.metadata?.thread_type !== "bootstrap" && - thread.metadata?.thread_type !== "roundtable", + const visibleThreads = sortThreadsPinnedFirst( + threads.filter( + (thread) => + thread.metadata?.thread_type !== "scheduler" && + thread.metadata?.thread_type !== "bootstrap" && + thread.metadata?.thread_type !== "roundtable", + ), ); if (visibleThreads.length === 0) { @@ -195,6 +210,8 @@ export function RecentChatList() {
{visibleThreads.map((thread) => { const isActive = pathOfThread(thread) === pathname; + const pinned = isThreadPinned(thread); + const favorited = isThreadFavorite(thread); return ( - {desensitize(titleOfThread(thread, t.pages.newChat))} + {pinned && ( + + )} + + {desensitize(titleOfThread(thread, t.pages.newChat))} + {env.NEXT_PUBLIC_STATIC_WEBSITE_ONLY !== "true" && ( @@ -243,6 +265,36 @@ export function RecentChatList() { {t.common.rename} + + togglePin(thread.thread_id, !pinned) + } + > + {pinned ? ( + + ) : ( + + )} + {pinned ? t.common.unpin : t.common.pin} + + + toggleFavorite(thread.thread_id, !favorited) + } + > + + + {favorited + ? t.common.unfavorite + : t.common.favorite} + + handleShare(thread)} > diff --git a/frontend-web/src/core/i18n/locales/en-US.ts b/frontend-web/src/core/i18n/locales/en-US.ts index 01520e16..8e9c2475 100644 --- a/frontend-web/src/core/i18n/locales/en-US.ts +++ b/frontend-web/src/core/i18n/locales/en-US.ts @@ -25,6 +25,10 @@ export const enUS: Translations = { edit: "Edit", rename: "Rename", share: "Share", + pin: "Pin", + unpin: "Unpin", + favorite: "Favorite", + unfavorite: "Unfavorite", openInNewWindow: "Open in new window", close: "Close", more: "More", @@ -255,6 +259,14 @@ export const enUS: Translations = { // Chats chats: { searchChats: "Search chats", + pinnedSection: "Pinned", + allChats: "All", + favorites: "Favorites", + onlyFavorites: "Favorites only", + prevPage: "Previous", + nextPage: "Next", + noChats: "No conversations yet", + noFavorites: "No favorited conversations yet", }, // Conversation sharing diff --git a/frontend-web/src/core/i18n/locales/types.ts b/frontend-web/src/core/i18n/locales/types.ts index c3c374a3..c1e053f9 100644 --- a/frontend-web/src/core/i18n/locales/types.ts +++ b/frontend-web/src/core/i18n/locales/types.ts @@ -26,6 +26,10 @@ export interface Translations { edit: string; rename: string; share: string; + pin: string; + unpin: string; + favorite: string; + unfavorite: string; openInNewWindow: string; close: string; more: string; @@ -198,6 +202,14 @@ export interface Translations { // Chats chats: { searchChats: string; + pinnedSection: string; + allChats: string; + favorites: string; + onlyFavorites: string; + prevPage: string; + nextPage: string; + noChats: string; + noFavorites: string; }; // Conversation sharing diff --git a/frontend-web/src/core/i18n/locales/zh-CN.ts b/frontend-web/src/core/i18n/locales/zh-CN.ts index 50ecf6fe..c0606781 100644 --- a/frontend-web/src/core/i18n/locales/zh-CN.ts +++ b/frontend-web/src/core/i18n/locales/zh-CN.ts @@ -25,6 +25,10 @@ export const zhCN: Translations = { edit: "编辑", rename: "重命名", share: "分享", + pin: "置顶", + unpin: "取消置顶", + favorite: "收藏", + unfavorite: "取消收藏", openInNewWindow: "在新窗口打开", close: "关闭", more: "更多", @@ -242,6 +246,14 @@ export const zhCN: Translations = { // Chats chats: { searchChats: "搜索对话", + pinnedSection: "置顶", + allChats: "全部", + favorites: "收藏", + onlyFavorites: "只看收藏", + prevPage: "上一页", + nextPage: "下一页", + noChats: "暂无对话", + noFavorites: "暂无收藏的对话", }, // Conversation sharing diff --git a/frontend-web/src/core/threads/hooks.ts b/frontend-web/src/core/threads/hooks.ts index 983bbf3b..6a021e46 100644 --- a/frontend-web/src/core/threads/hooks.ts +++ b/frontend-web/src/core/threads/hooks.ts @@ -621,15 +621,15 @@ export function useThreadStream({ lockThreadRagMode(threadId); } - const mergedContext = { - ...context, - ...extraContext, - }; - - await thread.submit( - { - messages: [ - { + const mergedContext = { + ...context, + ...extraContext, + }; + + await thread.submit( + { + messages: [ + { type: "human", content: [ { @@ -647,33 +647,33 @@ export function useThreadStream({ }, ], }, - { - threadId: threadId, - streamSubgraphs: true, - streamResumable: true, - config: { - recursion_limit: 1000, - }, - context: { - ...mergedContext, - thinking_enabled: mergedContext.mode !== "flash", - is_plan_mode: - mergedContext.mode === "pro" || mergedContext.mode === "ultra", - subagent_enabled: mergedContext.mode === "ultra", - reasoning_effort: - mergedContext.reasoning_effort ?? - (mergedContext.mode === "ultra" - ? "high" - : mergedContext.mode === "pro" - ? "medium" - : mergedContext.mode === "thinking" - ? "low" - : undefined), - memory_injection_enabled: - mergedContext.memory_injection_enabled ?? true, - rag_mode_enabled: ragModeEnabled, - thread_id: threadId, - }, + { + threadId: threadId, + streamSubgraphs: true, + streamResumable: true, + config: { + recursion_limit: 1000, + }, + context: { + ...mergedContext, + thinking_enabled: mergedContext.mode !== "flash", + is_plan_mode: + mergedContext.mode === "pro" || mergedContext.mode === "ultra", + subagent_enabled: mergedContext.mode === "ultra", + reasoning_effort: + mergedContext.reasoning_effort ?? + (mergedContext.mode === "ultra" + ? "high" + : mergedContext.mode === "pro" + ? "medium" + : mergedContext.mode === "thinking" + ? "low" + : undefined), + memory_injection_enabled: + mergedContext.memory_injection_enabled ?? true, + rag_mode_enabled: ragModeEnabled, + thread_id: threadId, + }, }, ); void queryClient.invalidateQueries({ queryKey: ["threads", "search"] }); @@ -853,6 +853,86 @@ export function useDeleteMessage() { }); } +/** + * Merge metadata into a conversation's record via the Gateway + * `PATCH /api/threads/{id}` route. Used to persist per-conversation flags such + * as `is_pinned` (置顶) and `is_favorite` (收藏) — a conversation belongs to a + * single user, so these live directly on the thread metadata (no join table). + * Updates the `["threads", "search"]` cache optimistically so the UI reflects + * the toggle immediately, mirroring `useRenameThread`. + */ +export function useUpdateThreadMetadata() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: async ({ + threadId, + metadata, + }: { + threadId: string; + metadata: Record; + }) => { + const response = await apiFetch( + `${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}`, + { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ metadata }), + }, + ); + if (!response.ok) { + const error = await response + .json() + .catch(() => ({ detail: "更新对话失败。" })); + throw new Error(error.detail ?? "更新对话失败。"); + } + }, + onSuccess(_, { threadId, metadata }) { + queryClient.setQueriesData( + { + queryKey: ["threads", "search"], + exact: false, + }, + (oldData: Array | undefined) => { + if (oldData == null) { + return oldData; + } + return oldData.map((t) => + t.thread_id === threadId + ? { + ...t, + metadata: { + ...(t.metadata as Record | null), + ...metadata, + }, + } + : t, + ); + }, + ); + }, + }); +} + +/** Toggle a conversation's pinned (置顶) flag. */ +export function useToggleThreadPin() { + const { mutate, ...rest } = useUpdateThreadMetadata(); + return { + ...rest, + toggle: (threadId: string, pinned: boolean) => + mutate({ threadId, metadata: { is_pinned: pinned } }), + }; +} + +/** Toggle a conversation's favorite (收藏) flag. */ +export function useToggleThreadFavorite() { + const { mutate, ...rest } = useUpdateThreadMetadata(); + return { + ...rest, + toggle: (threadId: string, favorite: boolean) => + mutate({ threadId, metadata: { is_favorite: favorite } }), + }; +} + export function useRenameThread() { const queryClient = useQueryClient(); const apiClient = getAPIClient(); diff --git a/frontend-web/src/core/threads/utils.ts b/frontend-web/src/core/threads/utils.ts index 6148a239..7a3d8769 100644 --- a/frontend-web/src/core/threads/utils.ts +++ b/frontend-web/src/core/threads/utils.ts @@ -81,6 +81,28 @@ export function titleOfThread(thread: AgentThread, fallback = "Untitled") { return fallback; } +export function isThreadPinned(thread: AgentThread): boolean { + return (thread.metadata as Record | null)?.is_pinned === true; +} + +export function isThreadFavorite(thread: AgentThread): boolean { + return (thread.metadata as Record | null)?.is_favorite === true; +} + +/** + * Stable sort that lifts pinned (置顶) threads to the top while preserving the + * incoming order (typically `updated_at` desc) within both the pinned and the + * unpinned group. + */ +export function sortThreadsPinnedFirst(threads: AgentThread[]): AgentThread[] { + const pinned: AgentThread[] = []; + const rest: AgentThread[] = []; + for (const thread of threads) { + (isThreadPinned(thread) ? pinned : rest).push(thread); + } + return [...pinned, ...rest]; +} + export function sanitizeThreadTitleForDisplay(title: string | null | undefined) { if (!title) return "Untitled"; diff --git a/frontend-web/src/pages/ChatsPage.tsx b/frontend-web/src/pages/ChatsPage.tsx index d4ba237c..d10b3c52 100644 --- a/frontend-web/src/pages/ChatsPage.tsx +++ b/frontend-web/src/pages/ChatsPage.tsx @@ -1,95 +1,255 @@ -"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, - WorkspaceHeader, -} from "@/components/workspace/workspace-container"; -import { useI18n } from "@/core/i18n/hooks"; -import { useThreads } from "@/core/threads/hooks"; -import { usePathOfThread, titleOfThread } from "@/core/threads/utils"; -import { formatTimeAgo } from "@/core/utils/datetime"; -import { desensitize } from "@/lib/desensitize"; - -export default function ChatsPage() { - const pathOfThread = usePathOfThread(); - 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}`; - }, [t.pages.chats, t.pages.appName]); - - const filteredThreads = useMemo(() => { - return threads?.filter((thread) => { - return ( - thread.metadata?.thread_type !== "scheduler" && - thread.metadata?.thread_type !== "roundtable" && - titleOfThread(thread).toLowerCase().includes(search.toLowerCase()) - ); - }); - }, [threads, search]); - return ( - - - -
-
-
- -
- setSearch(e.target.value)} - /> -
-
- -
- {filteredThreads?.map((thread) => ( - -
-
-
{desensitize(titleOfThread(thread, t.pages.newChat))}
-
- {thread.updated_at && ( -
- {formatTimeAgo(thread.updated_at)} -
- )} -
- - ))} -
- -
-
- - - - ); -} +"use client"; + +import { + ChevronLeft, + ChevronRight, + DownloadIcon, + Pin, + PinOff, + Star, +} 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, + WorkspaceHeader, +} from "@/components/workspace/workspace-container"; +import { useI18n } from "@/core/i18n/hooks"; +import { + useThreads, + useToggleThreadFavorite, + useToggleThreadPin, +} from "@/core/threads/hooks"; +import type { AgentThread } from "@/core/threads/types"; +import { + isThreadFavorite, + isThreadPinned, + usePathOfThread, + titleOfThread, +} from "@/core/threads/utils"; +import { formatTimeAgo } from "@/core/utils/datetime"; +import { desensitize } from "@/lib/desensitize"; +import { cn } from "@/lib/utils"; + +const PAGE_SIZE = 20; + +export default function ChatsPage() { + const pathOfThread = usePathOfThread(); + const { t } = useI18n(); + const { data: threads } = useThreads(); + const { toggle: togglePin } = useToggleThreadPin(); + const { toggle: toggleFavorite } = useToggleThreadFavorite(); + const [search, setSearch] = useState(""); + const [onlyFavorites, setOnlyFavorites] = useState(false); + const [page, setPage] = useState(1); + const [importOpen, setImportOpen] = useState(false); + + useEffect(() => { + document.title = `${t.pages.chats} - ${t.pages.appName}`; + }, [t.pages.chats, t.pages.appName]); + + // Reset to the first page whenever the filtered result set changes. + useEffect(() => { + setPage(1); + }, [search, onlyFavorites]); + + const filteredThreads = useMemo(() => { + return (threads ?? []).filter((thread) => { + if ( + thread.metadata?.thread_type === "scheduler" || + thread.metadata?.thread_type === "roundtable" + ) { + return false; + } + if (onlyFavorites && !isThreadFavorite(thread)) { + return false; + } + return titleOfThread(thread) + .toLowerCase() + .includes(search.toLowerCase()); + }); + }, [threads, search, onlyFavorites]); + + // Pinned conversations are surfaced in their own section and excluded from + // the paginated list so they stay visible regardless of the current page. + const pinnedThreads = useMemo( + () => filteredThreads.filter(isThreadPinned), + [filteredThreads], + ); + const restThreads = useMemo( + () => filteredThreads.filter((t) => !isThreadPinned(t)), + [filteredThreads], + ); + + const totalPages = Math.max(1, Math.ceil(restThreads.length / PAGE_SIZE)); + const currentPage = Math.min(page, totalPages); + const pagedThreads = restThreads.slice( + (currentPage - 1) * PAGE_SIZE, + currentPage * PAGE_SIZE, + ); + + const renderThread = (thread: AgentThread) => { + const pinned = isThreadPinned(thread); + const favorited = isThreadFavorite(thread); + return ( +
+ +
+ {pinned && ( + + )} + + {desensitize(titleOfThread(thread, t.pages.newChat))} + +
+ {thread.updated_at && ( +
+ {formatTimeAgo(thread.updated_at)} +
+ )} + + {/* Row actions — kept out of the Link so clicks don't navigate. */} +
+ + +
+
+ ); + }; + + const isEmpty = filteredThreads.length === 0; + + return ( + + + +
+
+
+
+ + +
+ +
+ setSearch(e.target.value)} + /> +
+
+ +
+ {pinnedThreads.length > 0 && ( + <> +
+ {t.chats.pinnedSection} +
+ {pinnedThreads.map(renderThread)} + + )} + {pagedThreads.map(renderThread)} + {isEmpty && ( +
+ {onlyFavorites ? t.chats.noFavorites : t.chats.noChats} +
+ )} + {totalPages > 1 && ( +
+ + + {currentPage} / {totalPages} + + +
+ )} +
+ +
+
+ + + + ); +} From 40206dc94dd77da24b6cd2c7d19871c512772de7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Mon, 8 Jun 2026 18:01:55 +0800 Subject: [PATCH 2/2] =?UTF-8?q?fix:=20Postgres=20store=20=E8=87=AA?= =?UTF-8?q?=E6=84=88=E8=BF=9E=E6=8E=A5=E6=B1=A0=20+=20=E5=90=AF=E7=94=A8?= =?UTF-8?q?=20PG=20checkpointer=20+=20=E5=88=86=E6=9E=90=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E8=B6=85=E6=97=B6=E6=8C=89=E6=9C=8D=E5=8A=A1=E7=AB=AF=E9=80=82?= =?UTF-8?q?=E9=85=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - async_provider: AsyncPostgresStore 改用连接池(check_connection 校验)， 避免单条长连接在服务重启/空闲超时后永久失效；池大小可经 DEER_FLOW_PG_POOL_MIN/MAX 环境变量调整 - admin_users: 排行榜分析查询的每语句超时按服务端区分 (MySQL>=5.7.4 max_execution_time / MariaDB max_statement_time / MySQL<=5.6 跳过 / PostgreSQL statement_timeout) - config.yaml: checkpointer.use_postgres 置为 true Co-Authored-By: Claude Opus 4.8 (1M context) --- .../app/gateway/routers/admin_users.py | 31 ++++++++++++-- .../deerflow/runtime/store/async_provider.py | 40 ++++++++++++++++++- offline-backend-20260512/config.yaml | 2 +- 3 files changed, 66 insertions(+), 7 deletions(-) diff --git a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py index 0865fcdc..5b1e0e7c 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/admin_users.py +++ b/offline-backend-20260512/backend/app/gateway/routers/admin_users.py @@ -458,13 +458,36 @@ async def _try_set_session_value(session, sql: str) -> None: logger.debug("Failed to rollback after analytics session setting failure", exc_info=True) +def _is_mariadb(dialect) -> bool: + """SQLAlchemy reports ``dialect.name == 'mysql'`` for MariaDB too.""" + if getattr(dialect, "_is_mariadb", False) or getattr(dialect, "is_mariadb", False): + return True + version_info = getattr(dialect, "server_version_info", None) or () + return any("mariadb" in str(part).lower() for part in version_info) + + async def _set_analytics_read_timeout(session) -> None: + """Best-effort per-statement read timeout for the analytics queries. + + The right knob differs by server, and some servers have none at all: + - MySQL >= 5.7.4: ``max_execution_time`` (milliseconds) + - MariaDB: ``max_statement_time`` (seconds) + - MySQL <= 5.6: no per-statement timeout variable — skip it entirely + (5.6 raises "Unknown system variable" otherwise) + - PostgreSQL: ``statement_timeout`` (milliseconds) + """ conn = await session.connection() - dialect = conn.dialect.name - if dialect == "mysql": - await _try_set_session_value(session, f"SET SESSION max_execution_time = {_LEADERBOARD_QUERY_TIMEOUT_MS}") + dialect = conn.dialect + name = dialect.name + if name == "mysql": + if _is_mariadb(dialect): + # MariaDB: max_statement_time is in SECONDS (fractional ok). + await _try_set_session_value(session, f"SET SESSION max_statement_time = {_LEADERBOARD_QUERY_TIMEOUT_MS / 1000}") + elif (getattr(dialect, "server_version_info", None) or ()) >= (5, 7, 4): + await _try_set_session_value(session, f"SET SESSION max_execution_time = {_LEADERBOARD_QUERY_TIMEOUT_MS}") + # else: MySQL <= 5.6 has no statement-timeout variable — skip silently. await _try_set_session_value(session, f"SET SESSION innodb_lock_wait_timeout = {_LEADERBOARD_LOCK_WAIT_SECONDS}") - elif dialect == "postgresql": + elif name == "postgresql": await _try_set_session_value(session, f"SET LOCAL statement_timeout = {_LEADERBOARD_QUERY_TIMEOUT_MS}") diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/store/async_provider.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/store/async_provider.py index 25156b01..94966e0d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/store/async_provider.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/store/async_provider.py @@ -28,6 +28,42 @@ from deerflow.runtime.store.provider import POSTGRES_CONN_REQUIRED, POSTGRES_STO logger = logging.getLogger(__name__) + +def _pool_size_env(name: str, default: int) -> int: + import os + + raw = os.getenv(name, "").strip() + if not raw: + return default + try: + return max(0, int(raw)) + except ValueError: + logger.warning("Invalid %s=%r; using default %d", name, raw, default) + return default + + +def _postgres_pool_config() -> dict: + """Pool config for the Postgres store. + + A single long-lived connection (the default ``from_conn_string`` behavior) + dies permanently on a server restart / idle timeout / network blip — every + later store call then fails with "server closed the connection + unexpectedly". A pool with ``check=check_connection`` validates each + connection on checkout and replaces dead ones, so the runtime self-heals. + Extra keys (``check``/``max_idle``/``max_lifetime``) are passed straight + through to ``AsyncConnectionPool`` by ``from_conn_string``. + """ + from psycopg_pool import AsyncConnectionPool + + return { + "min_size": _pool_size_env("DEER_FLOW_PG_POOL_MIN", 1), + "max_size": _pool_size_env("DEER_FLOW_PG_POOL_MAX", 10), + "check": AsyncConnectionPool.check_connection, + "max_idle": 300.0, + "max_lifetime": 3600.0, + } + + # --------------------------------------------------------------------------- # Internal backend factory # --------------------------------------------------------------------------- @@ -71,9 +107,9 @@ async def _async_store(config) -> AsyncIterator[BaseStore]: if not config.connection_string: raise ValueError(POSTGRES_CONN_REQUIRED) - async with AsyncPostgresStore.from_conn_string(config.connection_string) as store: + async with AsyncPostgresStore.from_conn_string(config.connection_string, pool_config=_postgres_pool_config()) as store: await store.setup() - logger.info("Store: using AsyncPostgresStore") + logger.info("Store: using AsyncPostgresStore (pooled, self-healing)") yield store return diff --git a/offline-backend-20260512/config.yaml b/offline-backend-20260512/config.yaml index aa098e5c..751c0156 100644 --- a/offline-backend-20260512/config.yaml +++ b/offline-backend-20260512/config.yaml @@ -200,7 +200,7 @@ checkpointer: type: postgres # true: 新 checkpoint 写入 Postgres；false: 保留 PG 配置但继续使用 SQLite。 # 也可临时用环境变量 DEER_FLOW_USE_PG_CHECKPOINTER=0 覆盖。 - use_postgres: false + use_postgres: true connection_string: postgresql://postgres:postgres123@47.88.25.99:8432/deerflow sqlite_connection_string: backend/.deer-flow/data/deerflow.db dual_read_enabled: true