Merge branch 'main' into rayhpeng/persistence-scaffold

# Conflicts:
#	config.example.yaml

backend/app/channels/base.py

@@ -106,3 +106,21 @@ class Channel(ABC):
                         logger.warning("[%s] file upload skipped for %s", self.name, attachment.filename)
                 except Exception:
                     logger.exception("[%s] failed to upload file %s", self.name, attachment.filename)
+
+    async def receive_file(self, msg: InboundMessage, thread_id: str) -> InboundMessage:
+        """
+        Optionally process and materialize inbound file attachments for this channel.
+
+        By default, this method does nothing and simply returns the original message.
+        Subclasses (e.g. FeishuChannel) may override this to download files (images, documents, etc)
+        referenced in msg.files, save them to the sandbox, and update msg.text to include
+        the sandbox file paths for downstream model consumption.
+
+        Args:
+            msg: The inbound message, possibly containing file metadata in msg.files.
+            thread_id: The resolved DeerFlow thread ID for sandbox path context.
+
+        Returns:
+            The (possibly modified) InboundMessage, with text and/or files updated as needed.
+        """
+        return msg

backend/app/channels/feishu.py

@@ -5,12 +5,15 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+import re
 import threading
-from typing import Any
+from typing import Any, Literal
 
 from app.channels.base import Channel
 from app.channels.commands import KNOWN_CHANNEL_COMMANDS
-from app.channels.message_bus import InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment
+from app.channels.message_bus import InboundMessage, InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment
+from deerflow.config.paths import VIRTUAL_PATH_PREFIX, get_paths
+from deerflow.sandbox.sandbox_provider import get_sandbox_provider
 
 logger = logging.getLogger(__name__)
 
@@ -56,6 +59,8 @@ class FeishuChannel(Channel):
         self._CreateFileRequestBody = None
         self._CreateImageRequest = None
         self._CreateImageRequestBody = None
+        self._GetMessageResourceRequest = None
+        self._thread_lock = threading.Lock()
 
     async def start(self) -> None:
         if self._running:
@@ -73,6 +78,7 @@ class FeishuChannel(Channel):
                 CreateMessageRequest,
                 CreateMessageRequestBody,
                 Emoji,
+                GetMessageResourceRequest,
                 PatchMessageRequest,
                 PatchMessageRequestBody,
                 ReplyMessageRequest,
@@ -96,6 +102,7 @@ class FeishuChannel(Channel):
         self._CreateFileRequestBody = CreateFileRequestBody
         self._CreateImageRequest = CreateImageRequest
         self._CreateImageRequestBody = CreateImageRequestBody
+        self._GetMessageResourceRequest = GetMessageResourceRequest
 
         app_id = self.config.get("app_id", "")
         app_secret = self.config.get("app_secret", "")
@@ -275,6 +282,112 @@ class FeishuChannel(Channel):
             raise RuntimeError(f"Feishu file upload failed: code={response.code}, msg={response.msg}")
         return response.data.file_key
 
+    async def receive_file(self, msg: InboundMessage, thread_id: str) -> InboundMessage:
+        """Download a Feishu file into the thread uploads directory.
+
+        Returns the sandbox virtual path when the image is persisted successfully.
+        """
+        if not msg.thread_ts:
+            logger.warning("[Feishu] received file message without thread_ts, cannot associate with conversation: %s", msg)
+            return msg
+        files = msg.files
+        if not files:
+            logger.warning("[Feishu] received message with no files: %s", msg)
+            return msg
+        text = msg.text
+        for file in files:
+            if file.get("image_key"):
+                virtual_path = await self._receive_single_file(msg.thread_ts, file["image_key"], "image", thread_id)
+                text = text.replace("[image]", virtual_path, 1)
+            elif file.get("file_key"):
+                virtual_path = await self._receive_single_file(msg.thread_ts, file["file_key"], "file", thread_id)
+                text = text.replace("[file]", virtual_path, 1)
+        msg.text = text
+        return msg
+
+    async def _receive_single_file(self, message_id: str, file_key: str, type: Literal["image", "file"], thread_id: str) -> str:
+        request = self._GetMessageResourceRequest.builder().message_id(message_id).file_key(file_key).type(type).build()
+
+        def inner():
+            return self._api_client.im.v1.message_resource.get(request)
+
+        try:
+            response = await asyncio.to_thread(inner)
+        except Exception:
+            logger.exception("[Feishu] resource get request failed for resource_key=%s type=%s", file_key, type)
+            return f"Failed to obtain the [{type}]"
+
+        if not response.success():
+            logger.warning(
+                "[Feishu] resource get failed: resource_key=%s, type=%s, code=%s, msg=%s, log_id=%s ",
+                file_key,
+                type,
+                response.code,
+                response.msg,
+                response.get_log_id(),
+            )
+            return f"Failed to obtain the [{type}]"
+
+        image_stream = getattr(response, "file", None)
+        if image_stream is None:
+            logger.warning("[Feishu] resource get returned no file stream: resource_key=%s, type=%s", file_key, type)
+            return f"Failed to obtain the [{type}]"
+
+        try:
+            content: bytes = await asyncio.to_thread(image_stream.read)
+        except Exception:
+            logger.exception("[Feishu] failed to read resource stream: resource_key=%s, type=%s", file_key, type)
+            return f"Failed to obtain the [{type}]"
+
+        if not content:
+            logger.warning("[Feishu] empty resource content: resource_key=%s, type=%s", file_key, type)
+            return f"Failed to obtain the [{type}]"
+
+        paths = get_paths()
+        paths.ensure_thread_dirs(thread_id)
+        uploads_dir = paths.sandbox_uploads_dir(thread_id).resolve()
+
+        ext = "png" if type == "image" else "bin"
+        raw_filename = getattr(response, "file_name", "") or f"feishu_{file_key[-12:]}.{ext}"
+
+        # Sanitize filename: preserve extension, replace path chars in name part
+        if "." in raw_filename:
+            name_part, ext = raw_filename.rsplit(".", 1)
+            name_part = re.sub(r"[./\\]", "_", name_part)
+            filename = f"{name_part}.{ext}"
+        else:
+            filename = re.sub(r"[./\\]", "_", raw_filename)
+        resolved_target = uploads_dir / filename
+
+        def down_load():
+            # use thread_lock to avoid filename conflicts when writing
+            with self._thread_lock:
+                resolved_target.write_bytes(content)
+
+        try:
+            await asyncio.to_thread(down_load)
+        except Exception:
+            logger.exception("[Feishu] failed to persist downloaded resource: %s, type=%s", resolved_target, type)
+            return f"Failed to obtain the [{type}]"
+
+        virtual_path = f"{VIRTUAL_PATH_PREFIX}/uploads/{resolved_target.name}"
+
+        try:
+            sandbox_provider = get_sandbox_provider()
+            sandbox_id = sandbox_provider.acquire(thread_id)
+            if sandbox_id != "local":
+                sandbox = sandbox_provider.get(sandbox_id)
+                if sandbox is None:
+                    logger.warning("[Feishu] sandbox not found for thread_id=%s", thread_id)
+                    return f"Failed to obtain the [{type}]"
+                sandbox.update_file(virtual_path, content)
+        except Exception:
+            logger.exception("[Feishu] failed to sync resource into non-local sandbox: %s", virtual_path)
+            return f"Failed to obtain the [{type}]"
+
+        logger.info("[Feishu] downloaded resource mapped: file_key=%s -> %s", file_key, virtual_path)
+        return virtual_path
+
     # -- message formatting ------------------------------------------------
 
     @staticmethod
@@ -479,9 +592,28 @@ class FeishuChannel(Channel):
             # Parse message content
             content = json.loads(message.content)
 
+            # files_list store the any-file-key in feishu messages, which can be used to download the file content later
+            # In Feishu channel, image_keys are independent of file_keys.
+            # The file_key includes files, videos, and audio, but does not include stickers.
+            files_list = []
+
             if "text" in content:
                 # Handle plain text messages
                 text = content["text"]
+            elif "file_key" in content:
+                file_key = content.get("file_key")
+                if isinstance(file_key, str) and file_key:
+                    files_list.append({"file_key": file_key})
+                    text = "[file]"
+                else:
+                    text = ""
+            elif "image_key" in content:
+                image_key = content.get("image_key")
+                if isinstance(image_key, str) and image_key:
+                    files_list.append({"image_key": image_key})
+                    text = "[image]"
+                else:
+                    text = ""
             elif "content" in content and isinstance(content["content"], list):
                 # Handle rich-text messages with a top-level "content" list (e.g., topic groups/posts)
                 text_paragraphs: list[str] = []
@@ -495,6 +627,16 @@ class FeishuChannel(Channel):
                                     text_value = element.get("text", "")
                                     if text_value:
                                         paragraph_text_parts.append(text_value)
+                                elif element.get("tag") == "img":
+                                    image_key = element.get("image_key")
+                                    if isinstance(image_key, str) and image_key:
+                                        files_list.append({"image_key": image_key})
+                                        paragraph_text_parts.append("[image]")
+                                elif element.get("tag") in ("file", "media"):
+                                    file_key = element.get("file_key")
+                                    if isinstance(file_key, str) and file_key:
+                                        files_list.append({"file_key": file_key})
+                                        paragraph_text_parts.append("[file]")
                         if paragraph_text_parts:
                             # Join text segments within a paragraph with spaces to avoid "helloworld"
                             text_paragraphs.append(" ".join(paragraph_text_parts))
@@ -514,7 +656,7 @@ class FeishuChannel(Channel):
                 text[:100] if text else "",
             )
 
-            if not text:
+            if not (text or files_list):
                 logger.info("[Feishu] empty text, ignoring message")
                 return
 
@@ -534,6 +676,7 @@ class FeishuChannel(Channel):
                 text=text,
                 msg_type=msg_type,
                 thread_ts=msg_id,
+                files=files_list,
                 metadata={"message_id": msg_id, "root_id": root_id},
             )
             inbound.topic_id = topic_id

backend/app/channels/manager.py

@@ -675,6 +675,18 @@ class ChannelManager:
             thread_id = await self._create_thread(client, msg)
 
         assistant_id, run_config, run_context = self._resolve_run_params(msg, thread_id)
+
+        # If the inbound message contains file attachments, let the channel
+        # materialize (download) them and update msg.text to include sandbox file paths.
+        # This enables downstream models to access user-uploaded files by path.
+        # Channels that do not support file download will simply return the original message.
+        if msg.files:
+            from .service import get_channel_service
+
+            service = get_channel_service()
+            channel = service.get_channel(msg.channel_name) if service else None
+            logger.info("[Manager] preparing receive file context for %d attachments", len(msg.files))
+            msg = await channel.receive_file(msg, thread_id) if channel else msg
         if extra_context:
             run_context.update(extra_context)
 

backend/app/channels/service.py

@@ -6,6 +6,7 @@ import logging
 import os
 from typing import Any
 
+from app.channels.base import Channel
 from app.channels.manager import DEFAULT_GATEWAY_URL, DEFAULT_LANGGRAPH_URL, ChannelManager
 from app.channels.message_bus import MessageBus
 from app.channels.store import ChannelStore
@@ -164,6 +165,10 @@ class ChannelService:
             "channels": channels_status,
         }
 
+    def get_channel(self, name: str) -> Channel | None:
+        """Return a running channel instance by name when available."""
+        return self._channels.get(name)
+
 
 # -- singleton access -------------------------------------------------------
 

backend/app/gateway/routers/skills.py

@@ -1,14 +1,29 @@
 import json
 import logging
+import shutil
 from pathlib import Path
 
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel, Field
 
 from app.gateway.path_utils import resolve_thread_virtual_path
+from deerflow.agents.lead_agent.prompt import clear_skills_system_prompt_cache
 from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, get_extensions_config, reload_extensions_config
 from deerflow.skills import Skill, load_skills
 from deerflow.skills.installer import SkillAlreadyExistsError, install_skill_from_archive
+from deerflow.skills.manager import (
+    append_history,
+    atomic_write,
+    custom_skill_exists,
+    ensure_custom_skill_is_editable,
+    get_custom_skill_dir,
+    get_custom_skill_file,
+    get_skill_history_file,
+    read_custom_skill_content,
+    read_history,
+    validate_skill_markdown_content,
+)
+from deerflow.skills.security_scanner import scan_skill_content
 
 logger = logging.getLogger(__name__)
 
@@ -52,6 +67,22 @@ class SkillInstallResponse(BaseModel):
     message: str = Field(..., description="Installation result message")
 
 
+class CustomSkillContentResponse(SkillResponse):
+    content: str = Field(..., description="Raw SKILL.md content")
+
+
+class CustomSkillUpdateRequest(BaseModel):
+    content: str = Field(..., description="Replacement SKILL.md content")
+
+
+class CustomSkillHistoryResponse(BaseModel):
+    history: list[dict]
+
+
+class SkillRollbackRequest(BaseModel):
+    history_index: int = Field(default=-1, description="History entry index to restore from, defaulting to the latest change.")
+
+
 def _skill_to_response(skill: Skill) -> SkillResponse:
     """Convert a Skill object to a SkillResponse."""
     return SkillResponse(
@@ -78,6 +109,180 @@ async def list_skills() -> SkillsListResponse:
         raise HTTPException(status_code=500, detail=f"Failed to load skills: {str(e)}")
 
 
+@router.post(
+    "/skills/install",
+    response_model=SkillInstallResponse,
+    summary="Install Skill",
+    description="Install a skill from a .skill file (ZIP archive) located in the thread's user-data directory.",
+)
+async def install_skill(request: SkillInstallRequest) -> SkillInstallResponse:
+    try:
+        skill_file_path = resolve_thread_virtual_path(request.thread_id, request.path)
+        result = install_skill_from_archive(skill_file_path)
+        return SkillInstallResponse(**result)
+    except FileNotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except SkillAlreadyExistsError as e:
+        raise HTTPException(status_code=409, detail=str(e))
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Failed to install skill: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to install skill: {str(e)}")
+
+
+@router.get("/skills/custom", response_model=SkillsListResponse, summary="List Custom Skills")
+async def list_custom_skills() -> SkillsListResponse:
+    try:
+        skills = [skill for skill in load_skills(enabled_only=False) if skill.category == "custom"]
+        return SkillsListResponse(skills=[_skill_to_response(skill) for skill in skills])
+    except Exception as e:
+        logger.error("Failed to list custom skills: %s", e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to list custom skills: {str(e)}")
+
+
+@router.get("/skills/custom/{skill_name}", response_model=CustomSkillContentResponse, summary="Get Custom Skill Content")
+async def get_custom_skill(skill_name: str) -> CustomSkillContentResponse:
+    try:
+        skills = load_skills(enabled_only=False)
+        skill = next((s for s in skills if s.name == skill_name and s.category == "custom"), None)
+        if skill is None:
+            raise HTTPException(status_code=404, detail=f"Custom skill '{skill_name}' not found")
+        return CustomSkillContentResponse(**_skill_to_response(skill).model_dump(), content=read_custom_skill_content(skill_name))
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("Failed to get custom skill %s: %s", skill_name, e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to get custom skill: {str(e)}")
+
+
+@router.put("/skills/custom/{skill_name}", response_model=CustomSkillContentResponse, summary="Edit Custom Skill")
+async def update_custom_skill(skill_name: str, request: CustomSkillUpdateRequest) -> CustomSkillContentResponse:
+    try:
+        ensure_custom_skill_is_editable(skill_name)
+        validate_skill_markdown_content(skill_name, request.content)
+        scan = await scan_skill_content(request.content, executable=False, location=f"{skill_name}/SKILL.md")
+        if scan.decision == "block":
+            raise HTTPException(status_code=400, detail=f"Security scan blocked the edit: {scan.reason}")
+        skill_file = get_custom_skill_dir(skill_name) / "SKILL.md"
+        prev_content = skill_file.read_text(encoding="utf-8")
+        atomic_write(skill_file, request.content)
+        append_history(
+            skill_name,
+            {
+                "action": "human_edit",
+                "author": "human",
+                "thread_id": None,
+                "file_path": "SKILL.md",
+                "prev_content": prev_content,
+                "new_content": request.content,
+                "scanner": {"decision": scan.decision, "reason": scan.reason},
+            },
+        )
+        clear_skills_system_prompt_cache()
+        return await get_custom_skill(skill_name)
+    except HTTPException:
+        raise
+    except FileNotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:
+        logger.error("Failed to update custom skill %s: %s", skill_name, e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to update custom skill: {str(e)}")
+
+
+@router.delete("/skills/custom/{skill_name}", summary="Delete Custom Skill")
+async def delete_custom_skill(skill_name: str) -> dict[str, bool]:
+    try:
+        ensure_custom_skill_is_editable(skill_name)
+        skill_dir = get_custom_skill_dir(skill_name)
+        prev_content = read_custom_skill_content(skill_name)
+        append_history(
+            skill_name,
+            {
+                "action": "human_delete",
+                "author": "human",
+                "thread_id": None,
+                "file_path": "SKILL.md",
+                "prev_content": prev_content,
+                "new_content": None,
+                "scanner": {"decision": "allow", "reason": "Deletion requested."},
+            },
+        )
+        shutil.rmtree(skill_dir)
+        clear_skills_system_prompt_cache()
+        return {"success": True}
+    except FileNotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:
+        logger.error("Failed to delete custom skill %s: %s", skill_name, e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to delete custom skill: {str(e)}")
+
+
+@router.get("/skills/custom/{skill_name}/history", response_model=CustomSkillHistoryResponse, summary="Get Custom Skill History")
+async def get_custom_skill_history(skill_name: str) -> CustomSkillHistoryResponse:
+    try:
+        if not custom_skill_exists(skill_name) and not get_skill_history_file(skill_name).exists():
+            raise HTTPException(status_code=404, detail=f"Custom skill '{skill_name}' not found")
+        return CustomSkillHistoryResponse(history=read_history(skill_name))
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("Failed to read history for %s: %s", skill_name, e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to read history: {str(e)}")
+
+
+@router.post("/skills/custom/{skill_name}/rollback", response_model=CustomSkillContentResponse, summary="Rollback Custom Skill")
+async def rollback_custom_skill(skill_name: str, request: SkillRollbackRequest) -> CustomSkillContentResponse:
+    try:
+        if not custom_skill_exists(skill_name) and not get_skill_history_file(skill_name).exists():
+            raise HTTPException(status_code=404, detail=f"Custom skill '{skill_name}' not found")
+        history = read_history(skill_name)
+        if not history:
+            raise HTTPException(status_code=400, detail=f"Custom skill '{skill_name}' has no history")
+        record = history[request.history_index]
+        target_content = record.get("prev_content")
+        if target_content is None:
+            raise HTTPException(status_code=400, detail="Selected history entry has no previous content to roll back to")
+        validate_skill_markdown_content(skill_name, target_content)
+        scan = await scan_skill_content(target_content, executable=False, location=f"{skill_name}/SKILL.md")
+        skill_file = get_custom_skill_file(skill_name)
+        current_content = skill_file.read_text(encoding="utf-8") if skill_file.exists() else None
+        history_entry = {
+            "action": "rollback",
+            "author": "human",
+            "thread_id": None,
+            "file_path": "SKILL.md",
+            "prev_content": current_content,
+            "new_content": target_content,
+            "rollback_from_ts": record.get("ts"),
+            "scanner": {"decision": scan.decision, "reason": scan.reason},
+        }
+        if scan.decision == "block":
+            append_history(skill_name, history_entry)
+            raise HTTPException(status_code=400, detail=f"Rollback blocked by security scanner: {scan.reason}")
+        atomic_write(skill_file, target_content)
+        append_history(skill_name, history_entry)
+        clear_skills_system_prompt_cache()
+        return await get_custom_skill(skill_name)
+    except HTTPException:
+        raise
+    except IndexError:
+        raise HTTPException(status_code=400, detail="history_index is out of range")
+    except FileNotFoundError as e:
+        raise HTTPException(status_code=404, detail=str(e))
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:
+        logger.error("Failed to roll back custom skill %s: %s", skill_name, e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to roll back custom skill: {str(e)}")
+
+
 @router.get(
     "/skills/{skill_name}",
     response_model=SkillResponse,
@@ -147,27 +352,3 @@ async def update_skill(skill_name: str, request: SkillUpdateRequest) -> SkillRes
     except Exception as e:
         logger.error(f"Failed to update skill {skill_name}: {e}", exc_info=True)
         raise HTTPException(status_code=500, detail=f"Failed to update skill: {str(e)}")
-
-
-@router.post(
-    "/skills/install",
-    response_model=SkillInstallResponse,
-    summary="Install Skill",
-    description="Install a skill from a .skill file (ZIP archive) located in the thread's user-data directory.",
-)
-async def install_skill(request: SkillInstallRequest) -> SkillInstallResponse:
-    try:
-        skill_file_path = resolve_thread_virtual_path(request.thread_id, request.path)
-        result = install_skill_from_archive(skill_file_path)
-        return SkillInstallResponse(**result)
-    except FileNotFoundError as e:
-        raise HTTPException(status_code=404, detail=str(e))
-    except SkillAlreadyExistsError as e:
-        raise HTTPException(status_code=409, detail=str(e))
-    except ValueError as e:
-        raise HTTPException(status_code=400, detail=str(e))
-    except HTTPException:
-        raise
-    except Exception as e:
-        logger.error(f"Failed to install skill: {e}", exc_info=True)
-        raise HTTPException(status_code=500, detail=f"Failed to install skill: {str(e)}")