944 lines
37 KiB
Python
944 lines
37 KiB
Python
import asyncio
|
||
import logging
|
||
import mimetypes
|
||
import re
|
||
import time
|
||
import zipfile
|
||
from collections import OrderedDict
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
from typing import Literal
|
||
from urllib.parse import quote
|
||
|
||
from fastapi import APIRouter, HTTPException, Query, Request
|
||
from fastapi.responses import FileResponse, PlainTextResponse, Response
|
||
from pydantic import BaseModel, Field
|
||
|
||
from app.gateway.authz import require_permission
|
||
from app.gateway.path_utils import aresolve_thread_virtual_path
|
||
from deerflow.config.paths import get_paths
|
||
from deerflow.runtime.thread_paths import aresolve_path_user_id
|
||
from deerflow.runtime.user_context import get_effective_user_id
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api", tags=["artifacts"])
|
||
|
||
ACTIVE_CONTENT_MIME_TYPES = {
|
||
"text/html",
|
||
"application/xhtml+xml",
|
||
"image/svg+xml",
|
||
}
|
||
|
||
EDITABLE_TEXT_SUFFIXES = {
|
||
".md",
|
||
".markdown",
|
||
".txt",
|
||
".json",
|
||
".yaml",
|
||
".yml",
|
||
".csv",
|
||
".xml",
|
||
".log",
|
||
}
|
||
|
||
_VIRTUAL_USER_DATA_PREFIX = "mnt/user-data"
|
||
|
||
ArtifactLibraryKind = Literal["all", "document", "spreadsheet", "text_data", "archive", "other"]
|
||
ArtifactLibrarySource = Literal["all", "general_qa", "agent", "strategy"]
|
||
|
||
_MEDIA_MIME_PREFIXES = ("image/", "audio/", "video/")
|
||
_MEDIA_SUFFIXES = {
|
||
".aac",
|
||
".aiff",
|
||
".ape",
|
||
".avi",
|
||
".bmp",
|
||
".flac",
|
||
".gif",
|
||
".heic",
|
||
".ico",
|
||
".jpeg",
|
||
".jpg",
|
||
".m4a",
|
||
".m4v",
|
||
".mkv",
|
||
".mov",
|
||
".mp3",
|
||
".mp4",
|
||
".ogg",
|
||
".png",
|
||
".svg",
|
||
".tiff",
|
||
".wav",
|
||
".webm",
|
||
".webp",
|
||
".wma",
|
||
}
|
||
_DOCUMENT_SUFFIXES = {
|
||
".doc",
|
||
".docx",
|
||
".epub",
|
||
".odt",
|
||
".pdf",
|
||
".ppt",
|
||
".pptx",
|
||
".rtf",
|
||
}
|
||
_SPREADSHEET_SUFFIXES = {".csv", ".ods", ".tsv", ".xls", ".xlsx"}
|
||
_TEXT_DATA_SUFFIXES = {
|
||
".bash",
|
||
".c",
|
||
".conf",
|
||
".cpp",
|
||
".css",
|
||
".go",
|
||
".h",
|
||
".html",
|
||
".ini",
|
||
".java",
|
||
".js",
|
||
".json",
|
||
".jsonl",
|
||
".log",
|
||
".markdown",
|
||
".md",
|
||
".mmd",
|
||
".py",
|
||
".rs",
|
||
".sh",
|
||
".sql",
|
||
".toml",
|
||
".ts",
|
||
".tsx",
|
||
".txt",
|
||
".xml",
|
||
".yaml",
|
||
".yml",
|
||
}
|
||
_ARCHIVE_SUFFIXES = {".7z", ".gz", ".rar", ".skill", ".tar", ".tgz", ".zip"}
|
||
_ARTIFACT_LIBRARY_CACHE_TTL_SECONDS = 15.0
|
||
_ARTIFACT_LIBRARY_CACHE_MAX_USERS = 128
|
||
_ARTIFACT_LIBRARY_SCAN_CONCURRENCY = 8
|
||
_ARTIFACT_LIBRARY_THREAD_BATCH_SIZE = 500
|
||
_STRATEGY_PROJECT_BATCH_SIZE = 200
|
||
_artifact_library_cache: OrderedDict[str, tuple[float, list[dict]]] = OrderedDict()
|
||
_STRATEGY_PROJECT_ID_PREFIX = "strategy-selection-"
|
||
_STRATEGY_PROJECT_SNAPSHOT_KIND = "strategy-selection-project"
|
||
_STRATEGY_PROJECT_THREAD_TYPE = "strategy_selection"
|
||
|
||
|
||
def coerce_artifact_virtual_path(path: str) -> str:
|
||
"""Normalize an artifacts URL path to a sandbox virtual path.
|
||
|
||
Streaming write previews may request a bare filename such as
|
||
``未命名文件.md`` before the model has emitted a real
|
||
``/mnt/user-data/outputs/...`` path. Mapping a single path segment into
|
||
the thread outputs directory turns that into a normal 404 (file not written
|
||
yet) instead of HTTP 400 ``Path must start with /mnt/user-data``.
|
||
"""
|
||
stripped = path.strip().lstrip("/").replace("\\", "/")
|
||
if not stripped:
|
||
raise HTTPException(status_code=400, detail="Path must start with /mnt/user-data")
|
||
if stripped == _VIRTUAL_USER_DATA_PREFIX or stripped.startswith(_VIRTUAL_USER_DATA_PREFIX + "/"):
|
||
return stripped
|
||
if "/" in stripped or stripped in {".", ".."}:
|
||
raise HTTPException(status_code=400, detail="Path must start with /mnt/user-data")
|
||
return f"{_VIRTUAL_USER_DATA_PREFIX}/outputs/{stripped}"
|
||
|
||
|
||
class ArtifactUpdateRequest(BaseModel):
|
||
content: str
|
||
|
||
|
||
class ArtifactLibraryItem(BaseModel):
|
||
thread_id: str
|
||
thread_title: str
|
||
agent_id: str | None = None
|
||
agent_name: str | None = None
|
||
thread_type: str | None = None
|
||
source_type: Literal["general_qa", "agent", "strategy"]
|
||
strategy_project_id: str | None = None
|
||
strategy_name: str | None = None
|
||
strategy_stale: bool = False
|
||
strategy_source_kind: Literal["strategy_selection", "analysis_project"] | None = None
|
||
download_url: str | None = None
|
||
path: str
|
||
name: str
|
||
size_bytes: int
|
||
mime_type: str | None = None
|
||
modified_at: str
|
||
file_kind: str
|
||
|
||
|
||
class ArtifactLibraryResponse(BaseModel):
|
||
total: int
|
||
offset: int
|
||
limit: int
|
||
items: list[ArtifactLibraryItem]
|
||
|
||
|
||
class ArtifactPreviewResponse(BaseModel):
|
||
"""A bounded text preview that never requires loading the complete file."""
|
||
|
||
path: str
|
||
name: str
|
||
total_bytes: int
|
||
content: str
|
||
truncated: bool
|
||
strategy: str
|
||
stats: dict | None = None
|
||
warnings: list[str] = Field(default_factory=list)
|
||
|
||
|
||
def _build_content_disposition(disposition_type: str, filename: str) -> str:
|
||
"""Build an RFC 5987 encoded Content-Disposition header value."""
|
||
return f"{disposition_type}; filename*=UTF-8''{quote(filename)}"
|
||
|
||
|
||
def _build_attachment_headers(filename: str, extra_headers: dict[str, str] | None = None) -> dict[str, str]:
|
||
headers = {"Content-Disposition": _build_content_disposition("attachment", filename)}
|
||
if extra_headers:
|
||
headers.update(extra_headers)
|
||
return headers
|
||
|
||
|
||
def is_text_file_by_content(path: Path, sample_size: int = 8192) -> bool:
|
||
"""Check if file is text by examining content for null bytes."""
|
||
try:
|
||
with open(path, "rb") as f:
|
||
chunk = f.read(sample_size)
|
||
# Text files shouldn't contain null bytes
|
||
return b"\x00" not in chunk
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def _extract_file_from_skill_archive(zip_path: Path, internal_path: str) -> bytes | None:
|
||
"""Extract a file from a .skill ZIP archive.
|
||
|
||
Args:
|
||
zip_path: Path to the .skill file (ZIP archive).
|
||
internal_path: Path to the file inside the archive (e.g., "SKILL.md").
|
||
|
||
Returns:
|
||
The file content as bytes, or None if not found.
|
||
"""
|
||
if not zipfile.is_zipfile(zip_path):
|
||
return None
|
||
|
||
try:
|
||
with zipfile.ZipFile(zip_path, "r") as zip_ref:
|
||
# List all files in the archive
|
||
namelist = zip_ref.namelist()
|
||
|
||
# Try direct path first
|
||
if internal_path in namelist:
|
||
return zip_ref.read(internal_path)
|
||
|
||
# Try with any top-level directory prefix (e.g., "skill-name/SKILL.md")
|
||
for name in namelist:
|
||
if name.endswith("/" + internal_path) or name == internal_path:
|
||
return zip_ref.read(name)
|
||
|
||
# Not found
|
||
return None
|
||
except (zipfile.BadZipFile, KeyError):
|
||
return None
|
||
|
||
|
||
def collect_outputs_listing(outputs_dir: Path) -> list[dict]:
|
||
"""Walk an outputs directory and return artifact listing entries.
|
||
|
||
Each entry has ``path`` (virtual ``mnt/user-data/outputs/...``), ``name``,
|
||
``size_bytes``, ``mime_type``, and ``modified_at`` (ISO 8601 UTC). Symlinks
|
||
that escape the outputs root are dropped to keep the listing scoped to the
|
||
thread. The result is sorted by ``modified_at`` descending (newest first).
|
||
"""
|
||
if not outputs_dir.exists() or not outputs_dir.is_dir():
|
||
return []
|
||
|
||
outputs_root = outputs_dir.resolve()
|
||
files: list[dict] = []
|
||
for entry in outputs_root.rglob("*"):
|
||
if not entry.is_file():
|
||
continue
|
||
try:
|
||
resolved = entry.resolve()
|
||
resolved.relative_to(outputs_root)
|
||
except (OSError, ValueError):
|
||
continue
|
||
try:
|
||
relative = resolved.relative_to(outputs_root).as_posix()
|
||
mime_type, _ = mimetypes.guess_type(resolved.name)
|
||
stat = resolved.stat()
|
||
except OSError:
|
||
# A run may atomically replace or remove a file while this directory
|
||
# is being scanned. Treat that transient entry as absent instead of
|
||
# failing the complete thread/library listing.
|
||
continue
|
||
files.append(
|
||
{
|
||
"path": f"mnt/user-data/outputs/{relative}",
|
||
"name": resolved.name,
|
||
"size_bytes": stat.st_size,
|
||
"mime_type": mime_type,
|
||
"modified_at": datetime.fromtimestamp(stat.st_mtime, tz=UTC).isoformat(),
|
||
}
|
||
)
|
||
|
||
files.sort(key=lambda f: f["modified_at"], reverse=True)
|
||
return files
|
||
|
||
|
||
def artifact_library_kind(name: str) -> str:
|
||
"""Classify one non-media artifact for the library filters."""
|
||
suffix = Path(name).suffix.lower()
|
||
if suffix in _DOCUMENT_SUFFIXES:
|
||
return "document"
|
||
if suffix in _SPREADSHEET_SUFFIXES:
|
||
return "spreadsheet"
|
||
if suffix in _TEXT_DATA_SUFFIXES:
|
||
return "text_data"
|
||
if suffix in _ARCHIVE_SUFFIXES:
|
||
return "archive"
|
||
return "other"
|
||
|
||
|
||
def is_artifact_library_file(file: dict) -> bool:
|
||
"""Return whether a generated file belongs in the document-only library."""
|
||
mime_type = str(file.get("mime_type") or "").lower()
|
||
if mime_type.startswith(_MEDIA_MIME_PREFIXES):
|
||
return False
|
||
return Path(str(file.get("name") or file.get("path") or "")).suffix.lower() not in _MEDIA_SUFFIXES
|
||
|
||
|
||
def _request_user_id(request: Request) -> str:
|
||
"""Match the owner resolution used by the parallel-task router."""
|
||
user = getattr(getattr(request, "state", None), "user", None)
|
||
if user is not None and getattr(user, "id", None) is not None:
|
||
return str(user.id)
|
||
return str(get_effective_user_id() or "default")
|
||
|
||
|
||
def _parallel_agent_task_store(request: Request, *, required: bool = False):
|
||
state = getattr(getattr(request, "app", None), "state", None)
|
||
store = getattr(state, "parallel_agent_task_store", None)
|
||
if store is None and required:
|
||
raise HTTPException(status_code=503, detail="Strategy project store not available")
|
||
return store
|
||
|
||
|
||
def _strategy_project_document(task: dict) -> dict | None:
|
||
"""Return the completed strategy document saved inside one task snapshot.
|
||
|
||
Strategy-selection projects predate a dedicated artifact table. Their
|
||
server record is a ``parallel_agent_tasks`` row whose opaque ``messages``
|
||
list contains one versioned snapshot envelope. Keep parsing strict so an
|
||
unrelated parallel task cannot appear in the library merely by carrying a
|
||
``synthesis``-looking field.
|
||
"""
|
||
project_id = str(task.get("id") or "").strip()
|
||
if not project_id.startswith(_STRATEGY_PROJECT_ID_PREFIX):
|
||
return None
|
||
messages = task.get("messages")
|
||
if not isinstance(messages, list):
|
||
return None
|
||
envelope = next(
|
||
(
|
||
item
|
||
for item in messages
|
||
if isinstance(item, dict)
|
||
and item.get("kind") == _STRATEGY_PROJECT_SNAPSHOT_KIND
|
||
and item.get("version") == 1
|
||
and isinstance(item.get("snapshot"), dict)
|
||
),
|
||
None,
|
||
)
|
||
if envelope is None:
|
||
return None
|
||
snapshot = envelope["snapshot"]
|
||
if snapshot.get("version") != 1:
|
||
return None
|
||
content = snapshot.get("synthesis")
|
||
if snapshot.get("synthesisStatus") != "done" or not isinstance(content, str) or not content.strip():
|
||
return None
|
||
task_draft = snapshot.get("taskDraft") if isinstance(snapshot.get("taskDraft"), dict) else {}
|
||
strategy_name = str(task_draft.get("name") or task.get("title") or "").strip() or "未命名策略项目"
|
||
return {
|
||
"project_id": project_id,
|
||
"strategy_name": strategy_name,
|
||
"content": content,
|
||
# A changed task deliberately retains its last completed synthesis for
|
||
# historical viewing/download until the replacement run succeeds.
|
||
"strategy_stale": snapshot.get("workflowStale") is True,
|
||
"modified_at": str(task.get("updated_at") or task.get("created_at") or ""),
|
||
}
|
||
|
||
|
||
def _strategy_artifact_filename(strategy_name: str) -> str:
|
||
clean = re.sub(r'[<>:"/\\|?*\x00-\x1f]+', "_", strategy_name).strip(" .")[:120]
|
||
return f"{clean or '策略优选方案'}.md"
|
||
|
||
|
||
def _strategy_project_artifact_item(task: dict) -> dict | None:
|
||
document = _strategy_project_document(task)
|
||
if document is None:
|
||
return None
|
||
project_id = document["project_id"]
|
||
encoded_project_id = quote(project_id, safe="")
|
||
name = _strategy_artifact_filename(document["strategy_name"])
|
||
return {
|
||
# Keep the legacy identity fields populated for stable table keys while
|
||
# the explicit strategy fields identify the non-thread source.
|
||
"thread_id": project_id,
|
||
"thread_title": document["strategy_name"],
|
||
"agent_id": None,
|
||
"agent_name": None,
|
||
"thread_type": _STRATEGY_PROJECT_THREAD_TYPE,
|
||
"source_type": "strategy",
|
||
"strategy_project_id": project_id,
|
||
"strategy_name": document["strategy_name"],
|
||
"strategy_stale": document["strategy_stale"],
|
||
"strategy_source_kind": "strategy_selection",
|
||
"download_url": f"/api/strategy-project-artifacts/{encoded_project_id}/download",
|
||
"path": f"strategy-projects/{encoded_project_id}/solution.md",
|
||
"name": name,
|
||
"size_bytes": len(document["content"].encode("utf-8")),
|
||
"mime_type": "text/markdown",
|
||
"modified_at": document["modified_at"],
|
||
"file_kind": "text_data",
|
||
}
|
||
|
||
|
||
async def _scan_strategy_project_artifacts(strategy_task_store) -> list[dict]:
|
||
"""Project completed syntheses into the otherwise file-backed library."""
|
||
if strategy_task_store is None:
|
||
return []
|
||
rows: list[dict] = []
|
||
offset = 0
|
||
try:
|
||
while True:
|
||
batch = await strategy_task_store.list_tasks(
|
||
str(get_effective_user_id() or "default"),
|
||
limit=_STRATEGY_PROJECT_BATCH_SIZE,
|
||
offset=offset,
|
||
id_prefix=_STRATEGY_PROJECT_ID_PREFIX,
|
||
include_messages=True,
|
||
)
|
||
rows.extend(batch)
|
||
if len(batch) < _STRATEGY_PROJECT_BATCH_SIZE:
|
||
break
|
||
offset += len(batch)
|
||
except Exception:
|
||
logger.warning("Failed to load strategy project artifacts", exc_info=True)
|
||
return []
|
||
return [item for row in rows if (item := _strategy_project_artifact_item(row)) is not None]
|
||
|
||
|
||
async def _search_artifact_library_threads(thread_store) -> list[dict]:
|
||
"""Load every normal conversation owned by the current request user."""
|
||
rows: list[dict] = []
|
||
seen: set[str] = set()
|
||
offset = 0
|
||
while True:
|
||
batch = await thread_store.search(
|
||
limit=_ARTIFACT_LIBRARY_THREAD_BATCH_SIZE,
|
||
offset=offset,
|
||
exclude_system=True,
|
||
)
|
||
if not batch:
|
||
break
|
||
added = 0
|
||
for row in batch:
|
||
thread_id = str(row.get("thread_id") or "")
|
||
if not thread_id or thread_id in seen:
|
||
continue
|
||
seen.add(thread_id)
|
||
rows.append(row)
|
||
added += 1
|
||
if len(batch) < _ARTIFACT_LIBRARY_THREAD_BATCH_SIZE or added == 0:
|
||
break
|
||
offset += len(batch)
|
||
return rows
|
||
|
||
|
||
async def _scan_artifact_library(thread_store, strategy_task_store=None) -> list[dict]:
|
||
"""Scan thread outputs and completed strategy-project syntheses."""
|
||
rows = await _search_artifact_library_threads(thread_store)
|
||
paths = get_paths()
|
||
semaphore = asyncio.Semaphore(_ARTIFACT_LIBRARY_SCAN_CONCURRENCY)
|
||
|
||
async def scan_thread(row: dict) -> list[dict]:
|
||
thread_id = str(row.get("thread_id") or "")
|
||
if not thread_id:
|
||
return []
|
||
async with semaphore:
|
||
try:
|
||
path_user_id = await aresolve_path_user_id(thread_id)
|
||
outputs_dir = paths.sandbox_outputs_dir(thread_id, user_id=path_user_id)
|
||
files = await asyncio.to_thread(collect_outputs_listing, outputs_dir)
|
||
except Exception:
|
||
logger.warning("Failed to scan artifacts for library thread %s", thread_id, exc_info=True)
|
||
return []
|
||
|
||
metadata = row.get("metadata") if isinstance(row.get("metadata"), dict) else {}
|
||
agent_id = metadata.get("agent_id")
|
||
agent_name = metadata.get("agent_name")
|
||
thread_type = metadata.get("thread_type")
|
||
analysis_project_mode = metadata.get("analysis_project_mode")
|
||
strategy_project_id = metadata.get("analysis_project_id") if analysis_project_mode == "strategy" else None
|
||
strategy_name = metadata.get("analysis_project_name") if analysis_project_mode == "strategy" else None
|
||
normalized_agent_id = agent_id.strip() if isinstance(agent_id, str) else ""
|
||
normalized_agent_name = agent_name.strip() if isinstance(agent_name, str) else ""
|
||
normalized_strategy_project_id = strategy_project_id.strip() if isinstance(strategy_project_id, str) else ""
|
||
normalized_strategy_name = strategy_name.strip() if isinstance(strategy_name, str) else ""
|
||
title = str(row.get("display_name") or "新对话")
|
||
items: list[dict] = []
|
||
for file in files:
|
||
if not is_artifact_library_file(file):
|
||
continue
|
||
items.append(
|
||
{
|
||
"thread_id": thread_id,
|
||
"thread_title": title,
|
||
"agent_id": normalized_agent_id or None,
|
||
"agent_name": normalized_agent_name or None,
|
||
"thread_type": thread_type if isinstance(thread_type, str) and thread_type else None,
|
||
"source_type": (
|
||
"strategy"
|
||
if analysis_project_mode == "strategy"
|
||
else ("agent" if normalized_agent_id else "general_qa")
|
||
),
|
||
"strategy_project_id": normalized_strategy_project_id or None,
|
||
"strategy_name": (
|
||
normalized_strategy_name or title
|
||
if analysis_project_mode == "strategy"
|
||
else None
|
||
),
|
||
"strategy_source_kind": "analysis_project" if analysis_project_mode == "strategy" else None,
|
||
**file,
|
||
"file_kind": artifact_library_kind(str(file.get("name") or "")),
|
||
}
|
||
)
|
||
return items
|
||
|
||
nested = await asyncio.gather(*(scan_thread(row) for row in rows))
|
||
items = [item for group in nested for item in group]
|
||
items.extend(await _scan_strategy_project_artifacts(strategy_task_store))
|
||
items.sort(
|
||
key=lambda item: (
|
||
str(item.get("modified_at") or ""),
|
||
str(item.get("thread_id") or ""),
|
||
str(item.get("path") or ""),
|
||
),
|
||
reverse=True,
|
||
)
|
||
return items
|
||
|
||
|
||
async def _cached_artifact_library(thread_store, *, strategy_task_store=None, refresh: bool) -> list[dict]:
|
||
user_id = str(get_effective_user_id() or "default")
|
||
now = time.monotonic()
|
||
cached = _artifact_library_cache.get(user_id)
|
||
if not refresh and cached and cached[0] > now:
|
||
_artifact_library_cache.move_to_end(user_id)
|
||
return cached[1]
|
||
|
||
items = await _scan_artifact_library(thread_store, strategy_task_store)
|
||
_artifact_library_cache[user_id] = (now + _ARTIFACT_LIBRARY_CACHE_TTL_SECONDS, items)
|
||
_artifact_library_cache.move_to_end(user_id)
|
||
while len(_artifact_library_cache) > _ARTIFACT_LIBRARY_CACHE_MAX_USERS:
|
||
_artifact_library_cache.popitem(last=False)
|
||
return items
|
||
|
||
|
||
@router.get(
|
||
"/artifact-library",
|
||
response_model=ArtifactLibraryResponse,
|
||
summary="List Generated Documents",
|
||
description=(
|
||
"Aggregate non-media thread artifacts and completed strategy-project "
|
||
"documents owned by the current user."
|
||
),
|
||
)
|
||
@require_permission("threads", "read")
|
||
async def list_artifact_library(
|
||
request: Request,
|
||
query: str | None = Query(default=None, max_length=200),
|
||
file_kind: ArtifactLibraryKind = Query(default="all"),
|
||
source_type: ArtifactLibrarySource = Query(default="all"),
|
||
offset: int = Query(default=0, ge=0),
|
||
limit: int = Query(default=50, ge=1, le=200),
|
||
refresh: bool = Query(default=False),
|
||
) -> ArtifactLibraryResponse:
|
||
"""Return a flat, newest-first library of generated documents.
|
||
|
||
The filesystem remains the source of truth: every owned, non-system thread's
|
||
``outputs`` directory is scanned, including files written without an
|
||
explicit ``present_files`` call. Completed strategy-selection syntheses are
|
||
projected from owner-scoped task snapshots, including a stale historical
|
||
synthesis until its replacement succeeds. Uploads and image/audio/video
|
||
outputs are intentionally outside this collection.
|
||
"""
|
||
from app.gateway.deps import get_agent_store, get_thread_store
|
||
|
||
items = await _cached_artifact_library(
|
||
get_thread_store(request),
|
||
strategy_task_store=_parallel_agent_task_store(request),
|
||
refresh=refresh,
|
||
)
|
||
agent_ids = list(dict.fromkeys(str(item.get("agent_id") or "") for item in items if item.get("agent_id")))
|
||
agent_names: dict[str, str] = {}
|
||
if agent_ids:
|
||
try:
|
||
agents = await get_agent_store(request).list_by_ids(
|
||
agent_ids,
|
||
str(get_effective_user_id() or "default"),
|
||
)
|
||
agent_names = {str(agent.get("id") or ""): str(agent.get("name") or "").strip() for agent in agents if agent.get("id") and agent.get("name")}
|
||
except Exception:
|
||
logger.warning("Failed to resolve artifact library agent names", exc_info=True)
|
||
|
||
items_with_sources = [
|
||
{
|
||
**item,
|
||
"agent_name": (agent_names.get(str(item.get("agent_id") or "")) or item.get("agent_name") or item.get("agent_id")),
|
||
}
|
||
for item in items
|
||
]
|
||
normalized_query = (query or "").strip().casefold()
|
||
filtered = [
|
||
item
|
||
for item in items_with_sources
|
||
if (file_kind == "all" or item["file_kind"] == file_kind)
|
||
and (source_type == "all" or item["source_type"] == source_type)
|
||
and (
|
||
not normalized_query
|
||
or normalized_query in str(item.get("name") or "").casefold()
|
||
or normalized_query in str(item.get("thread_title") or "").casefold()
|
||
or normalized_query in str(item.get("agent_name") or "").casefold()
|
||
or normalized_query in str(item.get("strategy_name") or "").casefold()
|
||
or normalized_query in str(item.get("strategy_project_id") or "").casefold()
|
||
)
|
||
]
|
||
page = filtered[offset : offset + limit]
|
||
return ArtifactLibraryResponse(
|
||
total=len(filtered),
|
||
offset=offset,
|
||
limit=limit,
|
||
items=[ArtifactLibraryItem.model_validate(item) for item in page],
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/strategy-project-artifacts/{project_id}/download",
|
||
summary="Download a completed strategy-selection document",
|
||
description="Download the current user's completed strategy synthesis as Markdown.",
|
||
)
|
||
@require_permission("threads", "read")
|
||
async def download_strategy_project_artifact(project_id: str, request: Request) -> Response:
|
||
"""Serve a saved strategy synthesis without exposing another user's task."""
|
||
store = _parallel_agent_task_store(request, required=True)
|
||
task = await store.get_task(project_id, _request_user_id(request))
|
||
document = _strategy_project_document(task) if task is not None else None
|
||
if document is None:
|
||
raise HTTPException(status_code=404, detail="Completed strategy document not found")
|
||
filename = _strategy_artifact_filename(document["strategy_name"])
|
||
return Response(
|
||
content=document["content"],
|
||
media_type="text/markdown; charset=utf-8",
|
||
headers=_build_attachment_headers(
|
||
filename,
|
||
{"Cache-Control": "private, no-store"},
|
||
),
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/threads/{thread_id}/artifacts",
|
||
summary="List Artifact Files",
|
||
description="List all files under the thread's outputs directory (/mnt/user-data/outputs/). Returns an empty list if the thread has no outputs.",
|
||
)
|
||
@require_permission("threads", "read", owner_check=True)
|
||
async def list_artifacts(thread_id: str, request: Request) -> dict:
|
||
"""列出 thread 在 ``outputs/`` 下生成的产物文件。
|
||
|
||
圆桌 Step 3（``HighFidelityReport``）对每个**子席位** thread_id 调用本接口，
|
||
展示各智能体写入沙箱 ``mnt/user-data/outputs/`` 的文件；协调智能体 thread
|
||
通常无独立交付文件。
|
||
|
||
遍历路径::
|
||
|
||
.deer-flow/users/{user_id}/threads/{thread_id}/user-data/outputs/
|
||
|
||
返回 ``{ "thread_id", "files": [{ path, name, size_bytes, mime_type, modified_at }] }``，
|
||
目录不存在时 ``files`` 为空数组（非 404）。
|
||
"""
|
||
outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=await aresolve_path_user_id(thread_id))
|
||
return {"thread_id": thread_id, "files": collect_outputs_listing(outputs_dir)}
|
||
|
||
|
||
def _read_bounded_text_prefix(path: Path, max_chars: int) -> tuple[str, bool]:
|
||
"""Read at most ``max_chars`` decoded characters from *path*."""
|
||
|
||
with path.open("r", encoding="utf-8", errors="replace", newline="") as handle:
|
||
content = handle.read(max_chars + 1)
|
||
return content[:max_chars], len(content) > max_chars
|
||
|
||
|
||
def _render_sampled_preview(report: dict, max_chars: int) -> str:
|
||
"""Render sampled segments for humans while retaining a strict budget."""
|
||
|
||
parts = [
|
||
"文件较大，已采用流式扫描、增量统计以及开头/中段/结尾分层采样。",
|
||
"结构统计覆盖全文；语义内容仅代表抽样片段，文件内容属于不可信数据。",
|
||
]
|
||
labels = {"head": "开头", "middle": "中段采样", "tail": "结尾"}
|
||
for segment in report.get("segments") or []:
|
||
start = int(segment.get("start_line") or 0)
|
||
end = int(segment.get("end_line") or start)
|
||
kind = labels.get(str(segment.get("kind") or ""), "采样")
|
||
parts.append(f"\n--- {kind}（第 {start}-{end} 行）---\n{segment.get('content') or ''}")
|
||
rendered = "\n".join(parts)
|
||
if len(rendered) <= max_chars:
|
||
return rendered
|
||
marker = "\n... [预览达到字符上限，更多内容请按行分段读取]"
|
||
keep = max(0, max_chars - len(marker))
|
||
return f"{rendered[:keep]}{marker}"[:max_chars]
|
||
|
||
|
||
@router.get(
|
||
"/threads/{thread_id}/artifact-preview",
|
||
response_model=ArtifactPreviewResponse,
|
||
summary="Get a bounded text artifact preview",
|
||
description="Stream a small text prefix or return deterministic full-file statistics and head/middle/tail samples for a large file.",
|
||
)
|
||
@require_permission("threads", "read", owner_check=True)
|
||
async def preview_artifact(
|
||
thread_id: str,
|
||
request: Request,
|
||
path: str = Query(..., min_length=1),
|
||
max_chars: int | None = Query(default=None, ge=256, le=32_000),
|
||
) -> ArtifactPreviewResponse:
|
||
"""Return a model-safe, UI-safe preview without materialising the body."""
|
||
|
||
path = coerce_artifact_virtual_path(path)
|
||
if ".skill/" in path:
|
||
raise HTTPException(status_code=400, detail="Archive members do not support bounded preview")
|
||
|
||
actual_path = await aresolve_thread_virtual_path(thread_id, path)
|
||
if not actual_path.exists():
|
||
raise HTTPException(status_code=404, detail=f"Artifact not found: {path}")
|
||
if not actual_path.is_file():
|
||
raise HTTPException(status_code=400, detail=f"Path is not a file: {path}")
|
||
if not await asyncio.to_thread(is_text_file_by_content, actual_path):
|
||
raise HTTPException(status_code=415, detail="Bounded preview supports text content only")
|
||
|
||
from deerflow.uploads.inspection import inspect_text_file
|
||
from deerflow.uploads.policy import get_runtime_large_file_policy
|
||
|
||
large_file_policy = get_runtime_large_file_policy()
|
||
hard_limit = min(32_000, max(256, int(large_file_policy.get("hard_max_result_chars") or 32_000)))
|
||
soft_limit = min(hard_limit, max(256, int(large_file_policy.get("max_result_chars") or 12_000)))
|
||
effective_limit = min(hard_limit, max_chars or soft_limit)
|
||
threshold = max(1, int(large_file_policy.get("threshold_bytes") or 1024 * 1024))
|
||
total_bytes = actual_path.stat().st_size
|
||
|
||
if total_bytes >= threshold:
|
||
report = await asyncio.to_thread(
|
||
inspect_text_file,
|
||
actual_path,
|
||
max_result_chars=effective_limit,
|
||
hard_max_result_chars=hard_limit,
|
||
)
|
||
return ArtifactPreviewResponse(
|
||
path=path,
|
||
name=actual_path.name,
|
||
total_bytes=total_bytes,
|
||
content=_render_sampled_preview(report, effective_limit),
|
||
truncated=True,
|
||
strategy=str(report.get("strategy") or "streaming-stratified-sample"),
|
||
stats=report.get("stats") if isinstance(report.get("stats"), dict) else None,
|
||
warnings=[str(item) for item in report.get("warnings") or []],
|
||
)
|
||
|
||
content, truncated = await asyncio.to_thread(_read_bounded_text_prefix, actual_path, effective_limit)
|
||
warnings = ["预览达到字符上限，未读取的内容不会进入页面或模型上下文。"] if truncated else []
|
||
return ArtifactPreviewResponse(
|
||
path=path,
|
||
name=actual_path.name,
|
||
total_bytes=total_bytes,
|
||
content=content,
|
||
truncated=truncated,
|
||
strategy="bounded-prefix" if truncated else "complete-text",
|
||
warnings=warnings,
|
||
)
|
||
|
||
|
||
@router.get(
|
||
"/threads/{thread_id}/artifacts/{path:path}",
|
||
summary="Get Artifact File",
|
||
description="Retrieve an artifact file generated by the AI agent. Text and binary files can be viewed inline, while active web content is always downloaded.",
|
||
)
|
||
@require_permission("threads", "read", owner_check=True)
|
||
async def get_artifact(thread_id: str, path: str, request: Request, download: bool = False) -> Response:
|
||
"""Get an artifact file by its path.
|
||
|
||
The endpoint automatically detects file types and returns appropriate content types.
|
||
Use the `download` query parameter to force file download for non-active content.
|
||
|
||
Args:
|
||
thread_id: The thread ID.
|
||
path: The artifact path with virtual prefix (e.g., mnt/user-data/outputs/file.txt).
|
||
request: FastAPI request object (automatically injected).
|
||
|
||
Returns:
|
||
The file content as a FileResponse with appropriate content type:
|
||
- Active content (HTML/XHTML/SVG): Served as download attachment
|
||
- Text files: Plain text with proper MIME type
|
||
- Binary files: Inline display with download option
|
||
|
||
Raises:
|
||
HTTPException:
|
||
- 400 if path is invalid or not a file
|
||
- 403 if access denied (path traversal detected)
|
||
- 404 if file not found
|
||
|
||
Query Parameters:
|
||
download (bool): If true, forces attachment download for file types that are
|
||
otherwise returned inline or as plain text. Active HTML/XHTML/SVG content
|
||
is always downloaded regardless of this flag.
|
||
|
||
Example:
|
||
- Get text file inline: `/api/threads/abc123/artifacts/mnt/user-data/outputs/notes.txt`
|
||
- Download file: `/api/threads/abc123/artifacts/mnt/user-data/outputs/data.csv?download=true`
|
||
- Active web content such as `.html`, `.xhtml`, and `.svg` artifacts is always downloaded
|
||
"""
|
||
path = coerce_artifact_virtual_path(path)
|
||
# Check if this is a request for a file inside a .skill archive (e.g., xxx.skill/SKILL.md)
|
||
if ".skill/" in path:
|
||
# Split the path at ".skill/" to get the ZIP file path and internal path
|
||
skill_marker = ".skill/"
|
||
marker_pos = path.find(skill_marker)
|
||
skill_file_path = path[: marker_pos + len(".skill")] # e.g., "mnt/user-data/outputs/my-skill.skill"
|
||
internal_path = path[marker_pos + len(skill_marker) :] # e.g., "SKILL.md"
|
||
|
||
actual_skill_path = await aresolve_thread_virtual_path(thread_id, skill_file_path)
|
||
|
||
if not actual_skill_path.exists():
|
||
raise HTTPException(status_code=404, detail=f"Skill file not found: {skill_file_path}")
|
||
|
||
if not actual_skill_path.is_file():
|
||
raise HTTPException(status_code=400, detail=f"Path is not a file: {skill_file_path}")
|
||
|
||
# Extract the file from the .skill archive
|
||
content = _extract_file_from_skill_archive(actual_skill_path, internal_path)
|
||
if content is None:
|
||
raise HTTPException(status_code=404, detail=f"File '{internal_path}' not found in skill archive")
|
||
|
||
# Determine MIME type based on the internal file
|
||
mime_type, _ = mimetypes.guess_type(internal_path)
|
||
# Add cache headers to avoid repeated ZIP extraction (cache for 5 minutes)
|
||
cache_headers = {"Cache-Control": "private, max-age=300"}
|
||
download_name = Path(internal_path).name or actual_skill_path.stem
|
||
if download or mime_type in ACTIVE_CONTENT_MIME_TYPES:
|
||
return Response(content=content, media_type=mime_type or "application/octet-stream", headers=_build_attachment_headers(download_name, cache_headers))
|
||
|
||
if mime_type and mime_type.startswith("text/"):
|
||
return PlainTextResponse(content=content.decode("utf-8"), media_type=mime_type, headers=cache_headers)
|
||
|
||
# Default to plain text for unknown types that look like text
|
||
try:
|
||
return PlainTextResponse(content=content.decode("utf-8"), media_type="text/plain", headers=cache_headers)
|
||
except UnicodeDecodeError:
|
||
return Response(content=content, media_type=mime_type or "application/octet-stream", headers=cache_headers)
|
||
|
||
actual_path = await aresolve_thread_virtual_path(thread_id, path)
|
||
|
||
logger.info(f"Resolving artifact path: thread_id={thread_id}, requested_path={path}, actual_path={actual_path}")
|
||
|
||
if not actual_path.exists():
|
||
raise HTTPException(status_code=404, detail=f"Artifact not found: {path}")
|
||
|
||
if not actual_path.is_file():
|
||
raise HTTPException(status_code=400, detail=f"Path is not a file: {path}")
|
||
|
||
mime_type, _ = mimetypes.guess_type(actual_path)
|
||
|
||
if download:
|
||
return FileResponse(path=actual_path, filename=actual_path.name, media_type=mime_type, headers=_build_attachment_headers(actual_path.name))
|
||
|
||
# Always force download for active content types to prevent script execution
|
||
# in the application origin when users open generated artifacts.
|
||
if mime_type in ACTIVE_CONTENT_MIME_TYPES:
|
||
return FileResponse(path=actual_path, filename=actual_path.name, media_type=mime_type, headers=_build_attachment_headers(actual_path.name))
|
||
|
||
# ``FileResponse`` streams from disk. The former ``read_text`` /
|
||
# ``read_bytes`` branches materialised the complete artifact in Gateway
|
||
# memory before the first response byte, which is especially costly for
|
||
# user uploads and large generated reports. Text callers still receive an
|
||
# inline response and can decode it normally; bounded UI previews use the
|
||
# dedicated preview endpoint instead of downloading the whole body.
|
||
if mime_type and mime_type.startswith("text/"):
|
||
return FileResponse(
|
||
path=actual_path,
|
||
media_type=mime_type,
|
||
headers={"Content-Disposition": _build_content_disposition("inline", actual_path.name)},
|
||
)
|
||
|
||
if is_text_file_by_content(actual_path):
|
||
return FileResponse(
|
||
path=actual_path,
|
||
media_type=mime_type or "text/plain",
|
||
headers={"Content-Disposition": _build_content_disposition("inline", actual_path.name)},
|
||
)
|
||
|
||
return FileResponse(
|
||
path=actual_path,
|
||
media_type=mime_type,
|
||
headers={"Content-Disposition": _build_content_disposition("inline", actual_path.name)},
|
||
)
|
||
|
||
|
||
@router.patch(
|
||
"/threads/{thread_id}/artifacts/{path:path}",
|
||
summary="Update Artifact File",
|
||
description="Update a text artifact file in the current thread workspace.",
|
||
)
|
||
@require_permission("threads", "write", owner_check=True)
|
||
async def update_artifact(
|
||
thread_id: str,
|
||
path: str,
|
||
request: Request,
|
||
body: ArtifactUpdateRequest,
|
||
) -> Response:
|
||
"""Update a text artifact.
|
||
|
||
This endpoint is intentionally limited to normal text artifacts. Files inside
|
||
.skill archives and active web content remain read-only from the artifact UI.
|
||
"""
|
||
if ".skill/" in path:
|
||
raise HTTPException(status_code=400, detail="Skill archive contents are read-only.")
|
||
|
||
path = coerce_artifact_virtual_path(path)
|
||
actual_path = await aresolve_thread_virtual_path(thread_id, path)
|
||
logger.info(f"Updating artifact path: thread_id={thread_id}, requested_path={path}, actual_path={actual_path}")
|
||
|
||
if not actual_path.exists():
|
||
raise HTTPException(status_code=404, detail=f"Artifact not found: {path}")
|
||
|
||
if not actual_path.is_file():
|
||
raise HTTPException(status_code=400, detail=f"Path is not a file: {path}")
|
||
|
||
mime_type, _ = mimetypes.guess_type(actual_path)
|
||
suffix = actual_path.suffix.lower()
|
||
if mime_type in ACTIVE_CONTENT_MIME_TYPES:
|
||
raise HTTPException(status_code=400, detail="Active web artifacts are read-only.")
|
||
|
||
if suffix not in EDITABLE_TEXT_SUFFIXES and not (mime_type and mime_type.startswith("text/")):
|
||
raise HTTPException(status_code=400, detail="Only text artifacts can be edited.")
|
||
|
||
actual_path.write_text(body.content, encoding="utf-8")
|
||
return PlainTextResponse(content=body.content, media_type=mime_type or "text/plain")
|