From 362a8268e85193cb49a1eb312681088e0e94dcc3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Thu, 4 Jun 2026 14:27:54 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=94=AF=E6=8C=81=E5=88=A0=E9=99=A4?= =?UTF-8?q?=E5=8D=95=E6=9D=A1=E8=81=8A=E5=A4=A9=E6=B6=88=E6=81=AF=EF=BC=88?= =?UTF-8?q?=E7=94=A8=E6=88=B7=E6=8F=90=E9=97=AE=E4=B8=8E=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E5=9B=9E=E7=AD=94=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 后端新增 DELETE /api/threads/{thread_id}/messages/{message_id}，从最新 checkpoint 的 messages 通道过滤掉指定 id 并写回新 checkpoint - 前端 useDeleteMessage 钩子 + DeleteMessageButton（带确认框）组件 - 在用户问题气泡与助手回答工具栏上各加删除按钮，仅对已落库消息显示 - MessageList 本地隐藏已删消息并在切换会话时重置；只读/分享场景禁用删除 - 新增 deleteMessage 相关中英文案 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../messages/delete-message-button.tsx | 95 ++ .../workspace/messages/message-list-item.tsx | 39 +- .../workspace/messages/message-list.tsx | 38 +- frontend-web/src/core/i18n/locales/en-US.ts | 3 + frontend-web/src/core/i18n/locales/types.ts | 3 + frontend-web/src/core/i18n/locales/zh-CN.ts | 3 + frontend-web/src/core/threads/hooks.ts | 47 +- .../backend/app/gateway/routers/threads.py | 1234 +++++++++-------- 8 files changed, 868 insertions(+), 594 deletions(-) create mode 100644 frontend-web/src/components/workspace/messages/delete-message-button.tsx diff --git a/frontend-web/src/components/workspace/messages/delete-message-button.tsx b/frontend-web/src/components/workspace/messages/delete-message-button.tsx new file mode 100644 index 00000000..ea81bff5 --- /dev/null +++ b/frontend-web/src/components/workspace/messages/delete-message-button.tsx @@ -0,0 +1,95 @@ +import { Trash2Icon } from "lucide-react"; +import { useState } from "react"; +import { toast } from "sonner"; + +import { Button } from "@/components/ui/button"; +import { + Dialog, + DialogContent, + DialogDescription, + DialogFooter, + DialogHeader, + DialogTitle, +} from "@/components/ui/dialog"; +import { useI18n } from "@/core/i18n/hooks"; +import { useDeleteMessage } from "@/core/threads/hooks"; + +import { Tooltip } from "../tooltip"; + +/** + * Hover affordance that deletes a single message (a user question or an + * assistant answer) after a confirmation step. The actual removal is persisted + * server-side; `onDeleted` lets the parent hide the bubble immediately without + * waiting for the stream cache to refresh. + */ +export function DeleteMessageButton({ + threadId, + messageId, + onDeleted, + className, +}: { + threadId: string; + messageId: string; + onDeleted?: () => void; + className?: string; +}) { + const { t } = useI18n(); + const [open, setOpen] = useState(false); + const deleteMessage = useDeleteMessage(); + + const handleConfirm = async () => { + try { + await deleteMessage.mutateAsync({ threadId, messageId }); + onDeleted?.(); + setOpen(false); + } catch (error) { + toast.error( + error instanceof Error ? error.message : "删除消息失败。", + ); + } + }; + + return ( + <> + + + + + + + {t.inputBox.deleteMessageConfirmTitle} + + {t.inputBox.deleteMessageConfirmDescription} + + + + + + + + + + ); +} diff --git a/frontend-web/src/components/workspace/messages/message-list-item.tsx b/frontend-web/src/components/workspace/messages/message-list-item.tsx index 2deec268..1be1d34e 100644 --- a/frontend-web/src/components/workspace/messages/message-list-item.tsx +++ b/frontend-web/src/components/workspace/messages/message-list-item.tsx @@ -57,6 +57,7 @@ import { CopyButton } from '../copy-button'; import { Tooltip } from '../tooltip'; import { SkillDisplayBlock } from '../skill-display-block'; import { ThreadContext } from './context'; +import { DeleteMessageButton } from './delete-message-button'; import { DownloadWordButton } from './download-word-button'; import { MarkdownContent, type MarkdownContentProps } from './markdown-content'; import { MessageTokenUsage } from './message-token-usage'; @@ -135,6 +136,7 @@ export function MessageListItem({ displaySources = [], extraRehypePlugins, onEditResubmit, + onMessageDeleted, }: { className?: string; fitRowToContent?: boolean; @@ -145,12 +147,18 @@ export function MessageListItem({ displaySources?: RuntimeDisplaySource[]; extraRehypePlugins?: ReturnType; onEditResubmit?: (text: string) => void; + onMessageDeleted?: (messageId: string) => void; }) { const isHuman = message.type === 'human'; const messageMarkdown = extractContentFromMessage(message) ?? extractReasoningContentFromMessage(message) ?? ''; + // Deletion is offered only for messages that already exist server-side + // (real, non-optimistic ids). Optimistic placeholders (`opt-*`) have no + // checkpoint entry to remove. + const canDelete = + !!onMessageDeleted && !!message.id && !message.id.startsWith('opt-'); return ( {/* Human messages carry their own inline copy/edit toolbar inside @@ -191,6 +200,13 @@ export function MessageListItem({ fileName="message" /> )} + {canDelete && ( + onMessageDeleted?.(message.id!)} + /> + )} )} @@ -214,6 +230,7 @@ function HumanMessageBlock({ messageTime, components, onEditResubmit, + deleteButton, }: { text: string; pageContext?: EmbeddedPageContext | null; @@ -221,6 +238,7 @@ function HumanMessageBlock({ messageTime: { label: string; full: string } | null; components: MarkdownContentProps["components"]; onEditResubmit?: (text: string) => void; + deleteButton?: React.ReactNode; }) { const { t } = useI18n(); const [editing, setEditing] = useState(false); @@ -298,10 +316,10 @@ function HumanMessageBlock({ )} {pageContext && } - {text && ( + {(text || deleteButton) && (
- - {onEditResubmit && ( + {text && } + {text && onEditResubmit && (
)} @@ -360,6 +379,7 @@ function MessageContent_({ displaySources = [], extraRehypePlugins, onEditResubmit, + onMessageDeleted, }: { className?: string; message: Message; @@ -369,6 +389,7 @@ function MessageContent_({ displaySources?: RuntimeDisplaySource[]; extraRehypePlugins?: ReturnType; onEditResubmit?: (text: string) => void; + onMessageDeleted?: (messageId: string) => void; }) { const baseRehypePlugins = useRehypeSplitWordsIntoSpans(isLoading); @@ -500,6 +521,8 @@ function MessageContent_({ } if (isHuman) { + const canDelete = + !!onMessageDeleted && !!message.id && !message.id.startsWith('opt-'); return ( onMessageDeleted?.(message.id!)} + className="size-6" + /> + ) : null + } /> ); } diff --git a/frontend-web/src/components/workspace/messages/message-list.tsx b/frontend-web/src/components/workspace/messages/message-list.tsx index a8c9952b..4ecefac0 100644 --- a/frontend-web/src/components/workspace/messages/message-list.tsx +++ b/frontend-web/src/components/workspace/messages/message-list.tsx @@ -1,7 +1,7 @@ import type { Message } from "@langchain/langgraph-sdk"; import type { BaseStream } from "@langchain/langgraph-sdk/react"; import { CheckIcon, SendIcon, XIcon } from "lucide-react"; -import { useCallback, useEffect, useMemo, useState } from "react"; +import { useCallback, useContext, useEffect, useMemo, useState } from "react"; import { Conversation, @@ -37,6 +37,7 @@ import { resolveAllowMultiple } from "@/roundtable-planning/lib/clarification"; import { ArtifactFileList } from "../artifacts/artifact-file-list"; import { StreamingIndicator } from "../streaming-indicator"; +import { ThreadContext } from "./context"; import { MarkdownContent } from "./markdown-content"; import { MessageGroup } from "./message-group"; import { MessageListItem } from "./message-list-item"; @@ -88,7 +89,37 @@ export function MessageList({ ); const updateSubtask = useUpdateSubtask(); const { t } = useI18n(); - const messages = thread.messages; + const threadContext = useContext(ThreadContext); + // Read-only contexts (public share, mock previews) must not offer deletion. + const allowDelete = !threadContext?.publicShareCode && !threadContext?.isMock; + + // Locally hide messages the user just deleted. The server-side checkpoint is + // trimmed immediately, but the live `useStream` cache still holds the message + // until the next refetch — filtering by id here makes the removal instant and + // keeps it hidden for the rest of the session (a reload reads the trimmed + // state). Reset whenever the active thread changes. + const [hiddenMessageIds, setHiddenMessageIds] = useState>( + () => new Set(), + ); + useEffect(() => { + setHiddenMessageIds(new Set()); + }, [threadId]); + const handleMessageDeleted = useCallback((messageId: string) => { + setHiddenMessageIds((prev) => { + const next = new Set(prev); + next.add(messageId); + return next; + }); + }, []); + + const allMessages = thread.messages; + const messages = useMemo( + () => + hiddenMessageIds.size === 0 + ? allMessages + : allMessages.filter((m) => !m.id || !hiddenMessageIds.has(m.id)), + [allMessages, hiddenMessageIds], + ); // Stamp a wall-clock time onto every question/answer the first time it is // seen, in a single batched commit. Questions sent this session already carry @@ -259,6 +290,9 @@ export function MessageList({ displaySources={displaySources} extraRehypePlugins={extraRehypePlugins} onEditResubmit={onEditResubmit} + onMessageDeleted={ + allowDelete ? handleMessageDeleted : undefined + } /> ); }); diff --git a/frontend-web/src/core/i18n/locales/en-US.ts b/frontend-web/src/core/i18n/locales/en-US.ts index 22fccf76..b6e750e2 100644 --- a/frontend-web/src/core/i18n/locales/en-US.ts +++ b/frontend-web/src/core/i18n/locales/en-US.ts @@ -82,6 +82,9 @@ export const enUS: Translations = { inputBox: { placeholder: "How can I assist you today?", editQuestion: "Edit question", + deleteMessage: "Delete message", + deleteMessageConfirmTitle: "Delete this message?", + deleteMessageConfirmDescription: "This action cannot be undone.", createSkillPrompt: "We're going to build a new skill step by step with `skill-creator`. To start, what do you want this skill to do?", addAttachments: "Add attachments", diff --git a/frontend-web/src/core/i18n/locales/types.ts b/frontend-web/src/core/i18n/locales/types.ts index 5545fa3b..e05856bc 100644 --- a/frontend-web/src/core/i18n/locales/types.ts +++ b/frontend-web/src/core/i18n/locales/types.ts @@ -79,6 +79,9 @@ export interface Translations { inputBox: { placeholder: string; editQuestion: string; + deleteMessage: string; + deleteMessageConfirmTitle: string; + deleteMessageConfirmDescription: string; createSkillPrompt: string; addAttachments: string; mode: string; diff --git a/frontend-web/src/core/i18n/locales/zh-CN.ts b/frontend-web/src/core/i18n/locales/zh-CN.ts index 9d605ce7..b884439e 100644 --- a/frontend-web/src/core/i18n/locales/zh-CN.ts +++ b/frontend-web/src/core/i18n/locales/zh-CN.ts @@ -82,6 +82,9 @@ export const zhCN: Translations = { inputBox: { placeholder: "今天我能为你做些什么？", editQuestion: "编辑问题", + deleteMessage: "删除消息", + deleteMessageConfirmTitle: "删除这条消息？", + deleteMessageConfirmDescription: "删除后将无法恢复。", createSkillPrompt: "我们一起用 skill-creator 技能来创建一个技能吧。先问问我希望这个技能能做什么。", addAttachments: "添加附件", diff --git a/frontend-web/src/core/threads/hooks.ts b/frontend-web/src/core/threads/hooks.ts index 4ab0b2c2..573a1b27 100644 --- a/frontend-web/src/core/threads/hooks.ts +++ b/frontend-web/src/core/threads/hooks.ts @@ -350,12 +350,12 @@ export function useThreadStream({ // agent thread URL — otherwise fall back to the global new chat. const hash = typeof window !== "undefined" ? window.location.hash : ""; const agentMatch = hash.match(/#\/page\/workspace\/agents\/([^/]+)\//); - const iframeMatch = hash.match(/#\/page\/workspace\/chats_iframe\//); - const target = agentMatch - ? `/page/workspace/agents/${agentMatch[1]}/chats/new` - : iframeMatch - ? "/page/workspace/chats_iframe/new" - : "/page/workspace/chats/new"; + const iframeMatch = hash.match(/#\/page\/workspace\/chats_iframe\//); + const target = agentMatch + ? `/page/workspace/agents/${agentMatch[1]}/chats/new` + : iframeMatch + ? "/page/workspace/chats_iframe/new" + : "/page/workspace/chats/new"; // First navigate to a safe URL so the post-reload page lands on a // valid route. Then full-reload so TanStack Query cache, context // providers and any other state captured under the previous user's @@ -804,6 +804,41 @@ export function useDeleteThread() { }); } +/** + * Delete a single message (a user question or an assistant answer) from a + * thread. Hits the Gateway endpoint that filters the message out of the latest + * checkpoint. The live `useStream` cache is not auto-refreshed by this call — + * callers hide the deleted id locally (see `MessageList`) so the removal shows + * immediately; a reload then reads the persisted, already-trimmed state. + */ +export function useDeleteMessage() { + return useMutation({ + mutationFn: async ({ + threadId, + messageId, + }: { + threadId: string; + messageId: string; + }) => { + const response = await apiFetch( + `${getBackendBaseURL()}/api/threads/${encodeURIComponent( + threadId, + )}/messages/${encodeURIComponent(messageId)}`, + { + method: "DELETE", + }, + ); + + if (!response.ok) { + const error = await response + .json() + .catch(() => ({ detail: "删除消息失败。" })); + throw new Error(error.detail ?? "删除消息失败。"); + } + }, + }); +} + export function useRenameThread() { const queryClient = useQueryClient(); const apiClient = getAPIClient(); diff --git a/offline-backend-20260512/backend/app/gateway/routers/threads.py b/offline-backend-20260512/backend/app/gateway/routers/threads.py index b688e476..3007388a 100644 --- a/offline-backend-20260512/backend/app/gateway/routers/threads.py +++ b/offline-backend-20260512/backend/app/gateway/routers/threads.py @@ -1,24 +1,24 @@ -"""Thread CRUD, state, and history endpoints. - -Combines the existing thread-local filesystem cleanup with LangGraph -Platform-compatible thread management backed by the checkpointer. - -Channel values returned in state responses are serialized through -:func:`deerflow.runtime.serialization.serialize_channel_values` to -ensure LangChain message objects are converted to JSON-safe dicts -matching the LangGraph Platform wire format expected by the -``useStream`` React hook. -""" - -from __future__ import annotations - +"""Thread CRUD, state, and history endpoints. + +Combines the existing thread-local filesystem cleanup with LangGraph +Platform-compatible thread management backed by the checkpointer. + +Channel values returned in state responses are serialized through +:func:`deerflow.runtime.serialization.serialize_channel_values` to +ensure LangChain message objects are converted to JSON-safe dicts +matching the LangGraph Platform wire format expected by the +``useStream`` React hook. +""" + +from __future__ import annotations + import logging import json import os from pathlib import Path import uuid from typing import Any - + from fastapi import APIRouter, HTTPException, Request from langgraph.checkpoint.base import empty_checkpoint from pydantic import BaseModel, Field, field_validator @@ -40,21 +40,21 @@ from deerflow.runtime.references import ( ) from deerflow.runtime.user_context import get_effective_user_id from deerflow.utils.time import coerce_iso, now_iso - + logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/threads", tags=["threads"]) _RAG_DEBUG_LOGGER_NAME = "rag_citation_debug" - - -# Metadata keys that the server controls; clients are not allowed to set -# them. Pydantic ``@field_validator("metadata")`` strips them on every -# inbound model below so a malicious client cannot reflect a forged -# owner identity through the API surface. Defense-in-depth — the -# row-level invariant is still ``threads_meta.user_id`` populated from -# the auth contextvar; this list closes the metadata-blob echo gap. -_SERVER_RESERVED_METADATA_KEYS: frozenset[str] = frozenset({"owner_id", "user_id"}) - - + + +# Metadata keys that the server controls; clients are not allowed to set +# them. Pydantic ``@field_validator("metadata")`` strips them on every +# inbound model below so a malicious client cannot reflect a forged +# owner identity through the API surface. Defense-in-depth — the +# row-level invariant is still ``threads_meta.user_id`` populated from +# the auth contextvar; this list closes the metadata-blob echo gap. +_SERVER_RESERVED_METADATA_KEYS: frozenset[str] = frozenset({"owner_id", "user_id"}) + + def _strip_reserved_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]: """Return ``metadata`` with server-controlled keys removed.""" if not metadata: @@ -135,91 +135,91 @@ async def _load_persisted_skill_displays() -> dict[str, SkillDisplayConfig]: # --------------------------------------------------------------------------- # Response / request models # --------------------------------------------------------------------------- - - -class ThreadDeleteResponse(BaseModel): - """Response model for thread cleanup.""" - - success: bool - message: str - - -class ThreadResponse(BaseModel): - """Response model for a single thread.""" - - thread_id: str = Field(description="Unique thread identifier") - status: str = Field(default="idle", description="Thread status: idle, busy, interrupted, error") - created_at: str = Field(default="", description="ISO timestamp") - updated_at: str = Field(default="", description="ISO timestamp") - metadata: dict[str, Any] = Field(default_factory=dict, description="Thread metadata") - values: dict[str, Any] = Field(default_factory=dict, description="Current state channel values") - interrupts: dict[str, Any] = Field(default_factory=dict, description="Pending interrupts") - - -class ThreadCreateRequest(BaseModel): - """Request body for creating a thread.""" - - thread_id: str | None = Field(default=None, description="Optional thread ID (auto-generated if omitted)") - assistant_id: str | None = Field(default=None, description="Associate thread with an assistant") - metadata: dict[str, Any] = Field(default_factory=dict, description="Initial metadata") - - _strip_reserved = field_validator("metadata")(classmethod(lambda cls, v: _strip_reserved_metadata(v))) - - -class ThreadSearchRequest(BaseModel): - """Request body for searching threads.""" - - metadata: dict[str, Any] = Field(default_factory=dict, description="Metadata filter (exact match)") - limit: int = Field(default=100, ge=1, le=1000, description="Maximum results") - offset: int = Field(default=0, ge=0, description="Pagination offset") - status: str | None = Field(default=None, description="Filter by thread status") - - -class ThreadStateResponse(BaseModel): - """Response model for thread state.""" - - values: dict[str, Any] = Field(default_factory=dict, description="Current channel values") - next: list[str] = Field(default_factory=list, description="Next tasks to execute") - metadata: dict[str, Any] = Field(default_factory=dict, description="Checkpoint metadata") - checkpoint: dict[str, Any] = Field(default_factory=dict, description="Checkpoint info") - checkpoint_id: str | None = Field(default=None, description="Current checkpoint ID") - parent_checkpoint_id: str | None = Field(default=None, description="Parent checkpoint ID") - created_at: str | None = Field(default=None, description="Checkpoint timestamp") - tasks: list[dict[str, Any]] = Field(default_factory=list, description="Interrupted task details") - - -class ThreadPatchRequest(BaseModel): - """Request body for patching thread metadata.""" - - metadata: dict[str, Any] = Field(default_factory=dict, description="Metadata to merge") - - _strip_reserved = field_validator("metadata")(classmethod(lambda cls, v: _strip_reserved_metadata(v))) - - -class ThreadStateUpdateRequest(BaseModel): - """Request body for updating thread state (human-in-the-loop resume).""" - - values: dict[str, Any] | None = Field(default=None, description="Channel values to merge") - checkpoint_id: str | None = Field(default=None, description="Checkpoint to branch from") - checkpoint: dict[str, Any] | None = Field(default=None, description="Full checkpoint object") - as_node: str | None = Field(default=None, description="Node identity for the update") - - -class HistoryEntry(BaseModel): - """Single checkpoint history entry.""" - - checkpoint_id: str - parent_checkpoint_id: str | None = None - metadata: dict[str, Any] = Field(default_factory=dict) - values: dict[str, Any] = Field(default_factory=dict) - created_at: str | None = None - next: list[str] = Field(default_factory=list) - - + + +class ThreadDeleteResponse(BaseModel): + """Response model for thread cleanup.""" + + success: bool + message: str + + +class ThreadResponse(BaseModel): + """Response model for a single thread.""" + + thread_id: str = Field(description="Unique thread identifier") + status: str = Field(default="idle", description="Thread status: idle, busy, interrupted, error") + created_at: str = Field(default="", description="ISO timestamp") + updated_at: str = Field(default="", description="ISO timestamp") + metadata: dict[str, Any] = Field(default_factory=dict, description="Thread metadata") + values: dict[str, Any] = Field(default_factory=dict, description="Current state channel values") + interrupts: dict[str, Any] = Field(default_factory=dict, description="Pending interrupts") + + +class ThreadCreateRequest(BaseModel): + """Request body for creating a thread.""" + + thread_id: str | None = Field(default=None, description="Optional thread ID (auto-generated if omitted)") + assistant_id: str | None = Field(default=None, description="Associate thread with an assistant") + metadata: dict[str, Any] = Field(default_factory=dict, description="Initial metadata") + + _strip_reserved = field_validator("metadata")(classmethod(lambda cls, v: _strip_reserved_metadata(v))) + + +class ThreadSearchRequest(BaseModel): + """Request body for searching threads.""" + + metadata: dict[str, Any] = Field(default_factory=dict, description="Metadata filter (exact match)") + limit: int = Field(default=100, ge=1, le=1000, description="Maximum results") + offset: int = Field(default=0, ge=0, description="Pagination offset") + status: str | None = Field(default=None, description="Filter by thread status") + + +class ThreadStateResponse(BaseModel): + """Response model for thread state.""" + + values: dict[str, Any] = Field(default_factory=dict, description="Current channel values") + next: list[str] = Field(default_factory=list, description="Next tasks to execute") + metadata: dict[str, Any] = Field(default_factory=dict, description="Checkpoint metadata") + checkpoint: dict[str, Any] = Field(default_factory=dict, description="Checkpoint info") + checkpoint_id: str | None = Field(default=None, description="Current checkpoint ID") + parent_checkpoint_id: str | None = Field(default=None, description="Parent checkpoint ID") + created_at: str | None = Field(default=None, description="Checkpoint timestamp") + tasks: list[dict[str, Any]] = Field(default_factory=list, description="Interrupted task details") + + +class ThreadPatchRequest(BaseModel): + """Request body for patching thread metadata.""" + + metadata: dict[str, Any] = Field(default_factory=dict, description="Metadata to merge") + + _strip_reserved = field_validator("metadata")(classmethod(lambda cls, v: _strip_reserved_metadata(v))) + + +class ThreadStateUpdateRequest(BaseModel): + """Request body for updating thread state (human-in-the-loop resume).""" + + values: dict[str, Any] | None = Field(default=None, description="Channel values to merge") + checkpoint_id: str | None = Field(default=None, description="Checkpoint to branch from") + checkpoint: dict[str, Any] | None = Field(default=None, description="Full checkpoint object") + as_node: str | None = Field(default=None, description="Node identity for the update") + + +class HistoryEntry(BaseModel): + """Single checkpoint history entry.""" + + checkpoint_id: str + parent_checkpoint_id: str | None = None + metadata: dict[str, Any] = Field(default_factory=dict) + values: dict[str, Any] = Field(default_factory=dict) + created_at: str | None = None + next: list[str] = Field(default_factory=list) + + class ThreadHistoryRequest(BaseModel): - """Request body for checkpoint history.""" - - limit: int = Field(default=10, ge=1, le=100, description="Maximum entries") + """Request body for checkpoint history.""" + + limit: int = Field(default=10, ge=1, le=100, description="Maximum entries") before: str | None = Field(default=None, description="Cursor for pagination") @@ -257,48 +257,48 @@ class ReferenceDebugLogRequest(BaseModel): event: str thread_id: str | None = None payload: dict[str, Any] = Field(default_factory=dict) - - -# --------------------------------------------------------------------------- -# Helpers -# --------------------------------------------------------------------------- - - -def _delete_thread_data(thread_id: str, paths: Paths | None = None, *, user_id: str | None = None) -> ThreadDeleteResponse: - """Delete local persisted filesystem data for a thread.""" - path_manager = paths or get_paths() - try: - path_manager.delete_thread_dir(thread_id, user_id=user_id) - except ValueError as exc: - raise HTTPException(status_code=422, detail=str(exc)) from exc - except FileNotFoundError: - # Not critical — thread data may not exist on disk - logger.debug("No local thread data to delete for %s", sanitize_log_param(thread_id)) - return ThreadDeleteResponse(success=True, message=f"No local data for {thread_id}") - except Exception as exc: - logger.exception("Failed to delete thread data for %s", sanitize_log_param(thread_id)) - raise HTTPException(status_code=500, detail="Failed to delete local thread data.") from exc - - logger.info("Deleted local thread data for %s", sanitize_log_param(thread_id)) - return ThreadDeleteResponse(success=True, message=f"Deleted local thread data for {thread_id}") - - + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _delete_thread_data(thread_id: str, paths: Paths | None = None, *, user_id: str | None = None) -> ThreadDeleteResponse: + """Delete local persisted filesystem data for a thread.""" + path_manager = paths or get_paths() + try: + path_manager.delete_thread_dir(thread_id, user_id=user_id) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from exc + except FileNotFoundError: + # Not critical — thread data may not exist on disk + logger.debug("No local thread data to delete for %s", sanitize_log_param(thread_id)) + return ThreadDeleteResponse(success=True, message=f"No local data for {thread_id}") + except Exception as exc: + logger.exception("Failed to delete thread data for %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to delete local thread data.") from exc + + logger.info("Deleted local thread data for %s", sanitize_log_param(thread_id)) + return ThreadDeleteResponse(success=True, message=f"Deleted local thread data for {thread_id}") + + def _derive_thread_status(checkpoint_tuple) -> str: - """Derive thread status from checkpoint metadata.""" - if checkpoint_tuple is None: - return "idle" - pending_writes = getattr(checkpoint_tuple, "pending_writes", None) or [] - - # Check for error in pending writes - for pw in pending_writes: - if len(pw) >= 2 and pw[1] == "__error__": - return "error" - - # Check for pending next tasks (indicates interrupt) - tasks = getattr(checkpoint_tuple, "tasks", None) - if tasks: - return "interrupted" - + """Derive thread status from checkpoint metadata.""" + if checkpoint_tuple is None: + return "idle" + pending_writes = getattr(checkpoint_tuple, "pending_writes", None) or [] + + # Check for error in pending writes + for pw in pending_writes: + if len(pw) >= 2 and pw[1] == "__error__": + return "error" + + # Check for pending next tasks (indicates interrupt) + tasks = getattr(checkpoint_tuple, "tasks", None) + if tasks: + return "interrupted" + return "idle" @@ -544,281 +544,281 @@ def _build_reference_batches(messages: list[Any]) -> list[ReferenceBatchResponse mid = _message_id(assistant_message) or str(index) batches.append(ReferenceBatchResponse(id=mid, assistant_message_id=mid, sources=sources)) return batches - - -# --------------------------------------------------------------------------- -# Endpoints -# --------------------------------------------------------------------------- - - -@router.delete("/{thread_id}", response_model=ThreadDeleteResponse) -@require_permission("threads", "delete", owner_check=True, require_existing=True) -async def delete_thread_data(thread_id: str, request: Request) -> ThreadDeleteResponse: - """Delete local persisted filesystem data for a thread. - - Cleans DeerFlow-managed thread directories, removes checkpoint data, - and removes the thread_meta row from the configured ThreadMetaStore - (sqlite or memory). - """ - from app.gateway.deps import get_thread_store - - # Clean local filesystem - response = _delete_thread_data(thread_id, user_id=get_effective_user_id()) - - # Remove checkpoints (best-effort) - checkpointer = getattr(request.app.state, "checkpointer", None) - if checkpointer is not None: - try: - if hasattr(checkpointer, "adelete_thread"): - await checkpointer.adelete_thread(thread_id) - except Exception: - logger.debug("Could not delete checkpoints for thread %s (not critical)", sanitize_log_param(thread_id)) - - # Remove thread_meta row (best-effort) — required for sqlite backend - # so the deleted thread no longer appears in /threads/search. - try: - thread_store = get_thread_store(request) - await thread_store.delete(thread_id) - except Exception: - logger.debug("Could not delete thread_meta for %s (not critical)", sanitize_log_param(thread_id)) - - return response - - -@router.post("", response_model=ThreadResponse) -async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadResponse: - """Create a new thread. - - Writes a thread_meta record (so the thread appears in /threads/search) - and an empty checkpoint (so state endpoints work immediately). - Idempotent: returns the existing record when ``thread_id`` already exists. - """ - from app.gateway.deps import get_thread_store - - checkpointer = get_checkpointer(request) - thread_store = get_thread_store(request) - thread_id = body.thread_id or str(uuid.uuid4()) - now = now_iso() - # ``body.metadata`` is already stripped of server-reserved keys by - # ``ThreadCreateRequest._strip_reserved`` — see the model definition. - - # Idempotency: return existing record when already present - existing_record = await thread_store.get(thread_id) - if existing_record is not None: - return ThreadResponse( - thread_id=thread_id, - status=existing_record.get("status", "idle"), - created_at=coerce_iso(existing_record.get("created_at", "")), - updated_at=coerce_iso(existing_record.get("updated_at", "")), - metadata=existing_record.get("metadata", {}), - ) - - # Write thread_meta so the thread appears in /threads/search immediately - try: - await thread_store.create( - thread_id, - assistant_id=getattr(body, "assistant_id", None), - metadata=body.metadata, - ) - except Exception: - logger.exception("Failed to write thread_meta for %s", sanitize_log_param(thread_id)) - raise HTTPException(status_code=500, detail="Failed to create thread") - - # Write an empty checkpoint so state endpoints work immediately - config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} - try: - ckpt_metadata = { - "step": -1, - "source": "input", - "writes": None, - "parents": {}, - **body.metadata, - "created_at": now, - } - await checkpointer.aput(config, empty_checkpoint(), ckpt_metadata, {}) - except Exception: - logger.exception("Failed to create checkpoint for thread %s", sanitize_log_param(thread_id)) - raise HTTPException(status_code=500, detail="Failed to create thread") - - logger.info("Thread created: %s", sanitize_log_param(thread_id)) - return ThreadResponse( - thread_id=thread_id, - status="idle", - created_at=now, - updated_at=now, - metadata=body.metadata, - ) - - -@router.post("/search", response_model=list[ThreadResponse]) -async def search_threads(body: ThreadSearchRequest, request: Request) -> list[ThreadResponse]: - """Search and list threads. - - Delegates to the configured ThreadMetaStore implementation - (SQL-backed for sqlite/postgres, Store-backed for memory mode). - """ - from app.gateway.deps import get_thread_store - - repo = get_thread_store(request) - rows = await repo.search( - metadata=body.metadata or None, - status=body.status, - limit=body.limit, - offset=body.offset, - ) - return [ - ThreadResponse( - thread_id=r["thread_id"], - status=r.get("status", "idle"), - # ``coerce_iso`` heals legacy unix-second values that - # ``MemoryThreadMetaStore`` historically wrote with ``time.time()``; - # SQL-backed rows already arrive as ISO strings and pass through. - created_at=coerce_iso(r.get("created_at", "")), - updated_at=coerce_iso(r.get("updated_at", "")), - metadata=r.get("metadata", {}), - values={"title": r["display_name"]} if r.get("display_name") else {}, - interrupts={}, - ) - for r in rows - ] - - -@router.patch("/{thread_id}", response_model=ThreadResponse) -@require_permission("threads", "write", owner_check=True, require_existing=True) -async def patch_thread(thread_id: str, body: ThreadPatchRequest, request: Request) -> ThreadResponse: - """Merge metadata into a thread record.""" - from app.gateway.deps import get_thread_store - - thread_store = get_thread_store(request) - record = await thread_store.get(thread_id) - if record is None: - raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") - - # ``body.metadata`` already stripped by ``ThreadPatchRequest._strip_reserved``. - try: - await thread_store.update_metadata(thread_id, body.metadata) - except Exception: - logger.exception("Failed to patch thread %s", sanitize_log_param(thread_id)) - raise HTTPException(status_code=500, detail="Failed to update thread") - - # Re-read to get the merged metadata + refreshed updated_at - record = await thread_store.get(thread_id) or record - return ThreadResponse( - thread_id=thread_id, - status=record.get("status", "idle"), - created_at=coerce_iso(record.get("created_at", "")), - updated_at=coerce_iso(record.get("updated_at", "")), - metadata=record.get("metadata", {}), - ) - - -@router.get("/{thread_id}", response_model=ThreadResponse) -@require_permission("threads", "read", owner_check=True) -async def get_thread(thread_id: str, request: Request) -> ThreadResponse: - """Get thread info. - - Reads metadata from the ThreadMetaStore and derives the accurate - execution status from the checkpointer. Falls back to the checkpointer - alone for threads that pre-date ThreadMetaStore adoption (backward compat). - """ - from app.gateway.deps import get_thread_store - - thread_store = get_thread_store(request) - checkpointer = get_checkpointer(request) - - record: dict | None = await thread_store.get(thread_id) - - # Derive accurate status from the checkpointer - config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} - try: - checkpoint_tuple = await checkpointer.aget_tuple(config) - except Exception: - logger.exception("Failed to get checkpoint for thread %s", sanitize_log_param(thread_id)) - raise HTTPException(status_code=500, detail="Failed to get thread") - - if record is None and checkpoint_tuple is None: - raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") - - # If the thread exists in the checkpointer but not in thread_meta (e.g. - # legacy data created before thread_meta adoption), synthesize a minimal - # record from the checkpoint metadata. - if record is None and checkpoint_tuple is not None: - ckpt_meta = getattr(checkpoint_tuple, "metadata", {}) or {} - record = { - "thread_id": thread_id, - "status": "idle", - "created_at": coerce_iso(ckpt_meta.get("created_at", "")), - "updated_at": coerce_iso(ckpt_meta.get("updated_at", ckpt_meta.get("created_at", ""))), - "metadata": {k: v for k, v in ckpt_meta.items() if k not in ("created_at", "updated_at", "step", "source", "writes", "parents")}, - } - - if record is None: - raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") - - status = _derive_thread_status(checkpoint_tuple) if checkpoint_tuple is not None else record.get("status", "idle") - checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {} if checkpoint_tuple is not None else {} - channel_values = checkpoint.get("channel_values", {}) - - return ThreadResponse( - thread_id=thread_id, - status=status, - created_at=coerce_iso(record.get("created_at", "")), - updated_at=coerce_iso(record.get("updated_at", "")), - metadata=record.get("metadata", {}), - values=serialize_channel_values(channel_values), - ) - - -# --------------------------------------------------------------------------- + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + + +@router.delete("/{thread_id}", response_model=ThreadDeleteResponse) +@require_permission("threads", "delete", owner_check=True, require_existing=True) +async def delete_thread_data(thread_id: str, request: Request) -> ThreadDeleteResponse: + """Delete local persisted filesystem data for a thread. + + Cleans DeerFlow-managed thread directories, removes checkpoint data, + and removes the thread_meta row from the configured ThreadMetaStore + (sqlite or memory). + """ + from app.gateway.deps import get_thread_store + + # Clean local filesystem + response = _delete_thread_data(thread_id, user_id=get_effective_user_id()) + + # Remove checkpoints (best-effort) + checkpointer = getattr(request.app.state, "checkpointer", None) + if checkpointer is not None: + try: + if hasattr(checkpointer, "adelete_thread"): + await checkpointer.adelete_thread(thread_id) + except Exception: + logger.debug("Could not delete checkpoints for thread %s (not critical)", sanitize_log_param(thread_id)) + + # Remove thread_meta row (best-effort) — required for sqlite backend + # so the deleted thread no longer appears in /threads/search. + try: + thread_store = get_thread_store(request) + await thread_store.delete(thread_id) + except Exception: + logger.debug("Could not delete thread_meta for %s (not critical)", sanitize_log_param(thread_id)) + + return response + + +@router.post("", response_model=ThreadResponse) +async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadResponse: + """Create a new thread. + + Writes a thread_meta record (so the thread appears in /threads/search) + and an empty checkpoint (so state endpoints work immediately). + Idempotent: returns the existing record when ``thread_id`` already exists. + """ + from app.gateway.deps import get_thread_store + + checkpointer = get_checkpointer(request) + thread_store = get_thread_store(request) + thread_id = body.thread_id or str(uuid.uuid4()) + now = now_iso() + # ``body.metadata`` is already stripped of server-reserved keys by + # ``ThreadCreateRequest._strip_reserved`` — see the model definition. + + # Idempotency: return existing record when already present + existing_record = await thread_store.get(thread_id) + if existing_record is not None: + return ThreadResponse( + thread_id=thread_id, + status=existing_record.get("status", "idle"), + created_at=coerce_iso(existing_record.get("created_at", "")), + updated_at=coerce_iso(existing_record.get("updated_at", "")), + metadata=existing_record.get("metadata", {}), + ) + + # Write thread_meta so the thread appears in /threads/search immediately + try: + await thread_store.create( + thread_id, + assistant_id=getattr(body, "assistant_id", None), + metadata=body.metadata, + ) + except Exception: + logger.exception("Failed to write thread_meta for %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to create thread") + + # Write an empty checkpoint so state endpoints work immediately + config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} + try: + ckpt_metadata = { + "step": -1, + "source": "input", + "writes": None, + "parents": {}, + **body.metadata, + "created_at": now, + } + await checkpointer.aput(config, empty_checkpoint(), ckpt_metadata, {}) + except Exception: + logger.exception("Failed to create checkpoint for thread %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to create thread") + + logger.info("Thread created: %s", sanitize_log_param(thread_id)) + return ThreadResponse( + thread_id=thread_id, + status="idle", + created_at=now, + updated_at=now, + metadata=body.metadata, + ) + + +@router.post("/search", response_model=list[ThreadResponse]) +async def search_threads(body: ThreadSearchRequest, request: Request) -> list[ThreadResponse]: + """Search and list threads. + + Delegates to the configured ThreadMetaStore implementation + (SQL-backed for sqlite/postgres, Store-backed for memory mode). + """ + from app.gateway.deps import get_thread_store + + repo = get_thread_store(request) + rows = await repo.search( + metadata=body.metadata or None, + status=body.status, + limit=body.limit, + offset=body.offset, + ) + return [ + ThreadResponse( + thread_id=r["thread_id"], + status=r.get("status", "idle"), + # ``coerce_iso`` heals legacy unix-second values that + # ``MemoryThreadMetaStore`` historically wrote with ``time.time()``; + # SQL-backed rows already arrive as ISO strings and pass through. + created_at=coerce_iso(r.get("created_at", "")), + updated_at=coerce_iso(r.get("updated_at", "")), + metadata=r.get("metadata", {}), + values={"title": r["display_name"]} if r.get("display_name") else {}, + interrupts={}, + ) + for r in rows + ] + + +@router.patch("/{thread_id}", response_model=ThreadResponse) +@require_permission("threads", "write", owner_check=True, require_existing=True) +async def patch_thread(thread_id: str, body: ThreadPatchRequest, request: Request) -> ThreadResponse: + """Merge metadata into a thread record.""" + from app.gateway.deps import get_thread_store + + thread_store = get_thread_store(request) + record = await thread_store.get(thread_id) + if record is None: + raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") + + # ``body.metadata`` already stripped by ``ThreadPatchRequest._strip_reserved``. + try: + await thread_store.update_metadata(thread_id, body.metadata) + except Exception: + logger.exception("Failed to patch thread %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to update thread") + + # Re-read to get the merged metadata + refreshed updated_at + record = await thread_store.get(thread_id) or record + return ThreadResponse( + thread_id=thread_id, + status=record.get("status", "idle"), + created_at=coerce_iso(record.get("created_at", "")), + updated_at=coerce_iso(record.get("updated_at", "")), + metadata=record.get("metadata", {}), + ) + + +@router.get("/{thread_id}", response_model=ThreadResponse) +@require_permission("threads", "read", owner_check=True) +async def get_thread(thread_id: str, request: Request) -> ThreadResponse: + """Get thread info. + + Reads metadata from the ThreadMetaStore and derives the accurate + execution status from the checkpointer. Falls back to the checkpointer + alone for threads that pre-date ThreadMetaStore adoption (backward compat). + """ + from app.gateway.deps import get_thread_store + + thread_store = get_thread_store(request) + checkpointer = get_checkpointer(request) + + record: dict | None = await thread_store.get(thread_id) + + # Derive accurate status from the checkpointer + config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} + try: + checkpoint_tuple = await checkpointer.aget_tuple(config) + except Exception: + logger.exception("Failed to get checkpoint for thread %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to get thread") + + if record is None and checkpoint_tuple is None: + raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") + + # If the thread exists in the checkpointer but not in thread_meta (e.g. + # legacy data created before thread_meta adoption), synthesize a minimal + # record from the checkpoint metadata. + if record is None and checkpoint_tuple is not None: + ckpt_meta = getattr(checkpoint_tuple, "metadata", {}) or {} + record = { + "thread_id": thread_id, + "status": "idle", + "created_at": coerce_iso(ckpt_meta.get("created_at", "")), + "updated_at": coerce_iso(ckpt_meta.get("updated_at", ckpt_meta.get("created_at", ""))), + "metadata": {k: v for k, v in ckpt_meta.items() if k not in ("created_at", "updated_at", "step", "source", "writes", "parents")}, + } + + if record is None: + raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") + + status = _derive_thread_status(checkpoint_tuple) if checkpoint_tuple is not None else record.get("status", "idle") + checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {} if checkpoint_tuple is not None else {} + channel_values = checkpoint.get("channel_values", {}) + + return ThreadResponse( + thread_id=thread_id, + status=status, + created_at=coerce_iso(record.get("created_at", "")), + updated_at=coerce_iso(record.get("updated_at", "")), + metadata=record.get("metadata", {}), + values=serialize_channel_values(channel_values), + ) + + +# --------------------------------------------------------------------------- @router.get("/{thread_id}/state", response_model=ThreadStateResponse) @require_permission("threads", "read", owner_check=True) async def get_thread_state(thread_id: str, request: Request) -> ThreadStateResponse: - """Get the latest state snapshot for a thread. - - Channel values are serialized to ensure LangChain message objects - are converted to JSON-safe dicts. - """ - checkpointer = get_checkpointer(request) - - config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} - try: - checkpoint_tuple = await checkpointer.aget_tuple(config) - except Exception: - logger.exception("Failed to get state for thread %s", sanitize_log_param(thread_id)) - raise HTTPException(status_code=500, detail="Failed to get thread state") - - if checkpoint_tuple is None: - raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") - - checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {} - metadata = getattr(checkpoint_tuple, "metadata", {}) or {} - checkpoint_id = None - ckpt_config = getattr(checkpoint_tuple, "config", {}) - if ckpt_config: - checkpoint_id = ckpt_config.get("configurable", {}).get("checkpoint_id") - - channel_values = checkpoint.get("channel_values", {}) - - parent_config = getattr(checkpoint_tuple, "parent_config", None) - parent_checkpoint_id = None - if parent_config: - parent_checkpoint_id = parent_config.get("configurable", {}).get("checkpoint_id") - - tasks_raw = getattr(checkpoint_tuple, "tasks", []) or [] - next_tasks = [t.name for t in tasks_raw if hasattr(t, "name")] - tasks = [{"id": getattr(t, "id", ""), "name": getattr(t, "name", "")} for t in tasks_raw] - - values = serialize_channel_values(channel_values) - - return ThreadStateResponse( - values=values, - next=next_tasks, - metadata=metadata, - checkpoint={"id": checkpoint_id, "ts": coerce_iso(metadata.get("created_at", ""))}, - checkpoint_id=checkpoint_id, - parent_checkpoint_id=parent_checkpoint_id, - created_at=coerce_iso(metadata.get("created_at", "")), - tasks=tasks, + """Get the latest state snapshot for a thread. + + Channel values are serialized to ensure LangChain message objects + are converted to JSON-safe dicts. + """ + checkpointer = get_checkpointer(request) + + config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} + try: + checkpoint_tuple = await checkpointer.aget_tuple(config) + except Exception: + logger.exception("Failed to get state for thread %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to get thread state") + + if checkpoint_tuple is None: + raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") + + checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {} + metadata = getattr(checkpoint_tuple, "metadata", {}) or {} + checkpoint_id = None + ckpt_config = getattr(checkpoint_tuple, "config", {}) + if ckpt_config: + checkpoint_id = ckpt_config.get("configurable", {}).get("checkpoint_id") + + channel_values = checkpoint.get("channel_values", {}) + + parent_config = getattr(checkpoint_tuple, "parent_config", None) + parent_checkpoint_id = None + if parent_config: + parent_checkpoint_id = parent_config.get("configurable", {}).get("checkpoint_id") + + tasks_raw = getattr(checkpoint_tuple, "tasks", []) or [] + next_tasks = [t.name for t in tasks_raw if hasattr(t, "name")] + tasks = [{"id": getattr(t, "id", ""), "name": getattr(t, "name", "")} for t in tasks_raw] + + values = serialize_channel_values(channel_values) + + return ThreadStateResponse( + values=values, + next=next_tasks, + metadata=metadata, + checkpoint={"id": checkpoint_id, "ts": coerce_iso(metadata.get("created_at", ""))}, + checkpoint_id=checkpoint_id, + parent_checkpoint_id=parent_checkpoint_id, + created_at=coerce_iso(metadata.get("created_at", "")), + tasks=tasks, ) @@ -922,162 +922,230 @@ async def write_reference_debug_log(body: ReferenceDebugLogRequest) -> dict[str, @router.post("/{thread_id}/state", response_model=ThreadStateResponse) @require_permission("threads", "write", owner_check=True, require_existing=True) async def update_thread_state(thread_id: str, body: ThreadStateUpdateRequest, request: Request) -> ThreadStateResponse: - """Update thread state (e.g. for human-in-the-loop resume or title rename). - - Writes a new checkpoint that merges *body.values* into the latest - channel values, then syncs any updated ``title`` field through the - ThreadMetaStore abstraction so that ``/threads/search`` reflects the - change immediately in both sqlite and memory backends. - """ - from app.gateway.deps import get_thread_store - - checkpointer = get_checkpointer(request) - thread_store = get_thread_store(request) - - # checkpoint_ns must be present in the config for aput — default to "" - # (the root graph namespace). checkpoint_id is optional; omitting it - # fetches the latest checkpoint for the thread. - read_config: dict[str, Any] = { - "configurable": { - "thread_id": thread_id, - "checkpoint_ns": "", - } - } - if body.checkpoint_id: - read_config["configurable"]["checkpoint_id"] = body.checkpoint_id - - try: - checkpoint_tuple = await checkpointer.aget_tuple(read_config) - except Exception: - logger.exception("Failed to get state for thread %s", sanitize_log_param(thread_id)) - raise HTTPException(status_code=500, detail="Failed to get thread state") - - if checkpoint_tuple is None: - raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") - - # Work on mutable copies so we don't accidentally mutate cached objects. - checkpoint: dict[str, Any] = dict(getattr(checkpoint_tuple, "checkpoint", {}) or {}) - metadata: dict[str, Any] = dict(getattr(checkpoint_tuple, "metadata", {}) or {}) - channel_values: dict[str, Any] = dict(checkpoint.get("channel_values", {})) - - if body.values: - channel_values.update(body.values) - - checkpoint["channel_values"] = channel_values - metadata["updated_at"] = now_iso() - - if body.as_node: - metadata["source"] = "update" - metadata["step"] = metadata.get("step", 0) + 1 - metadata["writes"] = {body.as_node: body.values} - - # aput requires checkpoint_ns in the config — use the same config used for the - # read (which always includes checkpoint_ns=""). Do NOT include checkpoint_id - # so that aput generates a fresh checkpoint ID for the new snapshot. - write_config: dict[str, Any] = { - "configurable": { - "thread_id": thread_id, - "checkpoint_ns": "", - } - } - try: - new_config = await checkpointer.aput(write_config, checkpoint, metadata, {}) - except Exception: - logger.exception("Failed to update state for thread %s", sanitize_log_param(thread_id)) - raise HTTPException(status_code=500, detail="Failed to update thread state") - - new_checkpoint_id: str | None = None - if isinstance(new_config, dict): - new_checkpoint_id = new_config.get("configurable", {}).get("checkpoint_id") - - # Sync title changes through the ThreadMetaStore abstraction so /threads/search - # reflects them immediately in both sqlite and memory backends. - if body.values and "title" in body.values: - new_title = body.values["title"] - if new_title: # Skip empty strings and None - try: - await thread_store.update_display_name(thread_id, new_title) - except Exception: - logger.debug("Failed to sync title to thread_meta for %s (non-fatal)", sanitize_log_param(thread_id)) - - return ThreadStateResponse( - values=serialize_channel_values(channel_values), - next=[], - metadata=metadata, - checkpoint_id=new_checkpoint_id, - created_at=coerce_iso(metadata.get("created_at", "")), - ) - - -@router.post("/{thread_id}/history", response_model=list[HistoryEntry]) -@require_permission("threads", "read", owner_check=True) -async def get_thread_history(thread_id: str, body: ThreadHistoryRequest, request: Request) -> list[HistoryEntry]: - """Get checkpoint history for a thread. - - Messages are read from the checkpointer's channel values (the - authoritative source) and serialized via - :func:`~deerflow.runtime.serialization.serialize_channel_values`. - Only the latest (first) checkpoint carries the ``messages`` key to - avoid duplicating them across every entry. - """ - checkpointer = get_checkpointer(request) - - config: dict[str, Any] = {"configurable": {"thread_id": thread_id}} - if body.before: - config["configurable"]["checkpoint_id"] = body.before - - entries: list[HistoryEntry] = [] - is_latest_checkpoint = True - try: - async for checkpoint_tuple in checkpointer.alist(config, limit=body.limit): - ckpt_config = getattr(checkpoint_tuple, "config", {}) - parent_config = getattr(checkpoint_tuple, "parent_config", None) - metadata = getattr(checkpoint_tuple, "metadata", {}) or {} - checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {} - - checkpoint_id = ckpt_config.get("configurable", {}).get("checkpoint_id", "") - parent_id = None - if parent_config: - parent_id = parent_config.get("configurable", {}).get("checkpoint_id") - - channel_values = checkpoint.get("channel_values", {}) - - # Build values from checkpoint channel_values - values: dict[str, Any] = {} - if title := channel_values.get("title"): - values["title"] = title - if thread_data := channel_values.get("thread_data"): - values["thread_data"] = thread_data - - # Attach messages only to the latest checkpoint entry. - if is_latest_checkpoint: - messages = channel_values.get("messages") - if messages: - values["messages"] = serialize_channel_values({"messages": messages}).get("messages", []) - is_latest_checkpoint = False - - # Derive next tasks - tasks_raw = getattr(checkpoint_tuple, "tasks", []) or [] - next_tasks = [t.name for t in tasks_raw if hasattr(t, "name")] - - # Strip LangGraph internal keys from metadata - user_meta = {k: v for k, v in metadata.items() if k not in ("created_at", "updated_at", "step", "source", "writes", "parents")} - # Keep step for ordering context - if "step" in metadata: - user_meta["step"] = metadata["step"] - - entries.append( - HistoryEntry( - checkpoint_id=checkpoint_id, - parent_checkpoint_id=parent_id, - metadata=user_meta, - values=values, - created_at=coerce_iso(metadata.get("created_at", "")), - next=next_tasks, - ) - ) - except Exception: - logger.exception("Failed to get history for thread %s", sanitize_log_param(thread_id)) - raise HTTPException(status_code=500, detail="Failed to get thread history") - - return entries + """Update thread state (e.g. for human-in-the-loop resume or title rename). + + Writes a new checkpoint that merges *body.values* into the latest + channel values, then syncs any updated ``title`` field through the + ThreadMetaStore abstraction so that ``/threads/search`` reflects the + change immediately in both sqlite and memory backends. + """ + from app.gateway.deps import get_thread_store + + checkpointer = get_checkpointer(request) + thread_store = get_thread_store(request) + + # checkpoint_ns must be present in the config for aput — default to "" + # (the root graph namespace). checkpoint_id is optional; omitting it + # fetches the latest checkpoint for the thread. + read_config: dict[str, Any] = { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": "", + } + } + if body.checkpoint_id: + read_config["configurable"]["checkpoint_id"] = body.checkpoint_id + + try: + checkpoint_tuple = await checkpointer.aget_tuple(read_config) + except Exception: + logger.exception("Failed to get state for thread %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to get thread state") + + if checkpoint_tuple is None: + raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") + + # Work on mutable copies so we don't accidentally mutate cached objects. + checkpoint: dict[str, Any] = dict(getattr(checkpoint_tuple, "checkpoint", {}) or {}) + metadata: dict[str, Any] = dict(getattr(checkpoint_tuple, "metadata", {}) or {}) + channel_values: dict[str, Any] = dict(checkpoint.get("channel_values", {})) + + if body.values: + channel_values.update(body.values) + + checkpoint["channel_values"] = channel_values + metadata["updated_at"] = now_iso() + + if body.as_node: + metadata["source"] = "update" + metadata["step"] = metadata.get("step", 0) + 1 + metadata["writes"] = {body.as_node: body.values} + + # aput requires checkpoint_ns in the config — use the same config used for the + # read (which always includes checkpoint_ns=""). Do NOT include checkpoint_id + # so that aput generates a fresh checkpoint ID for the new snapshot. + write_config: dict[str, Any] = { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": "", + } + } + try: + new_config = await checkpointer.aput(write_config, checkpoint, metadata, {}) + except Exception: + logger.exception("Failed to update state for thread %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to update thread state") + + new_checkpoint_id: str | None = None + if isinstance(new_config, dict): + new_checkpoint_id = new_config.get("configurable", {}).get("checkpoint_id") + + # Sync title changes through the ThreadMetaStore abstraction so /threads/search + # reflects them immediately in both sqlite and memory backends. + if body.values and "title" in body.values: + new_title = body.values["title"] + if new_title: # Skip empty strings and None + try: + await thread_store.update_display_name(thread_id, new_title) + except Exception: + logger.debug("Failed to sync title to thread_meta for %s (non-fatal)", sanitize_log_param(thread_id)) + + return ThreadStateResponse( + values=serialize_channel_values(channel_values), + next=[], + metadata=metadata, + checkpoint_id=new_checkpoint_id, + created_at=coerce_iso(metadata.get("created_at", "")), + ) + + +@router.delete("/{thread_id}/messages/{message_id}") +@require_permission("threads", "write", owner_check=True, require_existing=True) +async def delete_thread_message(thread_id: str, message_id: str, request: Request) -> dict[str, Any]: + """Delete a single message (a user question or an assistant answer) from a + thread's latest checkpoint. + + Filters the message whose ``id`` matches ``message_id`` out of the + ``messages`` channel and writes a fresh checkpoint. The remaining messages + keep their original ``BaseMessage`` objects (mirroring + :func:`update_thread_state`), so the graph can still resume the thread. + """ + checkpointer = get_checkpointer(request) + + read_config: dict[str, Any] = { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": "", + } + } + try: + checkpoint_tuple = await checkpointer.aget_tuple(read_config) + except Exception: + logger.exception("Failed to get state for thread %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to get thread state") + + if checkpoint_tuple is None: + raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found") + + checkpoint: dict[str, Any] = dict(getattr(checkpoint_tuple, "checkpoint", {}) or {}) + metadata: dict[str, Any] = dict(getattr(checkpoint_tuple, "metadata", {}) or {}) + channel_values: dict[str, Any] = dict(checkpoint.get("channel_values", {})) + + messages = channel_values.get("messages") + if not isinstance(messages, list): + raise HTTPException(status_code=404, detail="Thread has no messages") + + filtered = [m for m in messages if _message_id(m) != message_id] + if len(filtered) == len(messages): + raise HTTPException(status_code=404, detail=f"Message {message_id} not found") + + channel_values["messages"] = filtered + checkpoint["channel_values"] = channel_values + metadata["updated_at"] = now_iso() + + write_config: dict[str, Any] = { + "configurable": { + "thread_id": thread_id, + "checkpoint_ns": "", + } + } + try: + await checkpointer.aput(write_config, checkpoint, metadata, {}) + except Exception: + logger.exception( + "Failed to delete message %s from thread %s", + sanitize_log_param(message_id), + sanitize_log_param(thread_id), + ) + raise HTTPException(status_code=500, detail="Failed to delete message") + + logger.info( + "Deleted message %s from thread %s", + sanitize_log_param(message_id), + sanitize_log_param(thread_id), + ) + return {"ok": True, "deleted": message_id, "remaining": len(filtered)} + + +@router.post("/{thread_id}/history", response_model=list[HistoryEntry]) +@require_permission("threads", "read", owner_check=True) +async def get_thread_history(thread_id: str, body: ThreadHistoryRequest, request: Request) -> list[HistoryEntry]: + """Get checkpoint history for a thread. + + Messages are read from the checkpointer's channel values (the + authoritative source) and serialized via + :func:`~deerflow.runtime.serialization.serialize_channel_values`. + Only the latest (first) checkpoint carries the ``messages`` key to + avoid duplicating them across every entry. + """ + checkpointer = get_checkpointer(request) + + config: dict[str, Any] = {"configurable": {"thread_id": thread_id}} + if body.before: + config["configurable"]["checkpoint_id"] = body.before + + entries: list[HistoryEntry] = [] + is_latest_checkpoint = True + try: + async for checkpoint_tuple in checkpointer.alist(config, limit=body.limit): + ckpt_config = getattr(checkpoint_tuple, "config", {}) + parent_config = getattr(checkpoint_tuple, "parent_config", None) + metadata = getattr(checkpoint_tuple, "metadata", {}) or {} + checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {} + + checkpoint_id = ckpt_config.get("configurable", {}).get("checkpoint_id", "") + parent_id = None + if parent_config: + parent_id = parent_config.get("configurable", {}).get("checkpoint_id") + + channel_values = checkpoint.get("channel_values", {}) + + # Build values from checkpoint channel_values + values: dict[str, Any] = {} + if title := channel_values.get("title"): + values["title"] = title + if thread_data := channel_values.get("thread_data"): + values["thread_data"] = thread_data + + # Attach messages only to the latest checkpoint entry. + if is_latest_checkpoint: + messages = channel_values.get("messages") + if messages: + values["messages"] = serialize_channel_values({"messages": messages}).get("messages", []) + is_latest_checkpoint = False + + # Derive next tasks + tasks_raw = getattr(checkpoint_tuple, "tasks", []) or [] + next_tasks = [t.name for t in tasks_raw if hasattr(t, "name")] + + # Strip LangGraph internal keys from metadata + user_meta = {k: v for k, v in metadata.items() if k not in ("created_at", "updated_at", "step", "source", "writes", "parents")} + # Keep step for ordering context + if "step" in metadata: + user_meta["step"] = metadata["step"] + + entries.append( + HistoryEntry( + checkpoint_id=checkpoint_id, + parent_checkpoint_id=parent_id, + metadata=user_meta, + values=values, + created_at=coerce_iso(metadata.get("created_at", "")), + next=next_tasks, + ) + ) + except Exception: + logger.exception("Failed to get history for thread %s", sanitize_log_param(thread_id)) + raise HTTPException(status_code=500, detail="Failed to get thread history") + + return entries