1153 lines
44 KiB
Python
1153 lines
44 KiB
Python
import asyncio
|
||
import logging
|
||
import mimetypes
|
||
import re
|
||
import zipfile
|
||
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.operation_timing import OperationTiming
|
||
from app.gateway.path_utils import aresolve_thread_virtual_path
|
||
from app.gateway.utils import sanitize_log_param
|
||
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
|
||
from deerflow.utils.file_io import file_io
|
||
|
||
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_INDEX_TTL_SECONDS = 300.0
|
||
# Filesystem reconciliation is deliberately low-concurrency: it is a background
|
||
# correctness sweep, not work that should compete with active model runs.
|
||
_ARTIFACT_LIBRARY_SCAN_CONCURRENCY = 2
|
||
_ARTIFACT_LIBRARY_THREAD_BATCH_SIZE = 500
|
||
_ARTIFACT_LIBRARY_SCAN_BATCH_SIZE = 100
|
||
_STRATEGY_PROJECT_BATCH_SIZE = 200
|
||
_STRATEGY_PROJECT_ID_PREFIX = "strategy-selection-"
|
||
_STRATEGY_PROJECT_SNAPSHOT_KIND = "strategy-selection-project"
|
||
_STRATEGY_PROJECT_THREAD_TYPE = "strategy_selection"
|
||
|
||
|
||
class ArtifactLibraryScanIncomplete(RuntimeError):
|
||
"""Raised when a reconciliation cannot safely publish a complete generation."""
|
||
|
||
|
||
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]
|
||
sync_status: Literal["idle", "syncing", "ready", "error"] = "idle"
|
||
syncing: bool = False
|
||
indexed_at: str | None = None
|
||
sync_error: str | None = None
|
||
|
||
|
||
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,
|
||
*,
|
||
errors: list[str] | None = None,
|
||
) -> 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. Direct per-thread listings skip that transient
|
||
# entry; full-library reconciliation also records it and refuses to
|
||
# replace the last complete database generation.
|
||
if errors is not None:
|
||
errors.append(entry.name)
|
||
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 _artifact_library_store(request: Request):
|
||
state = getattr(getattr(request, "app", None), "state", None)
|
||
store = getattr(state, "artifact_library_store", None)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Artifact library index 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,
|
||
*,
|
||
user_id: str | None = None,
|
||
strict: bool = False,
|
||
) -> list[dict]:
|
||
"""Project completed syntheses into the otherwise file-backed library."""
|
||
if strategy_task_store is None:
|
||
return []
|
||
resolved_user_id = user_id or str(get_effective_user_id() or "default")
|
||
rows: list[dict] = []
|
||
offset = 0
|
||
try:
|
||
while True:
|
||
batch = await strategy_task_store.list_tasks(
|
||
resolved_user_id,
|
||
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 as exc:
|
||
logger.warning("Failed to load strategy project artifacts", exc_info=True)
|
||
if strict:
|
||
raise ArtifactLibraryScanIncomplete(
|
||
"Strategy artifact source could not be read completely"
|
||
) from exc
|
||
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,
|
||
*,
|
||
user_id: str | None = None,
|
||
) -> list[dict]:
|
||
"""Load every normal conversation owned by the current request user."""
|
||
resolved_user_id = user_id or str(get_effective_user_id() or "default")
|
||
rows: list[dict] = []
|
||
seen: set[str] = set()
|
||
offset = 0
|
||
while True:
|
||
batch = await thread_store.search(
|
||
limit=_ARTIFACT_LIBRARY_THREAD_BATCH_SIZE,
|
||
offset=offset,
|
||
user_id=resolved_user_id,
|
||
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,
|
||
*,
|
||
user_id: str | None = None,
|
||
strict: bool = False,
|
||
) -> list[dict]:
|
||
"""Scan thread outputs and completed strategy-project syntheses."""
|
||
explicit_user_id = user_id
|
||
resolved_user_id = user_id or str(get_effective_user_id() or "default")
|
||
rows = await _search_artifact_library_threads(thread_store, user_id=resolved_user_id)
|
||
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:
|
||
# Background indexing passes an explicit owner from the
|
||
# authenticated request, avoiding one DB owner lookup per
|
||
# historical thread. Direct/legacy callers retain the safe
|
||
# resolver fallback.
|
||
path_user_id = (
|
||
resolved_user_id
|
||
if explicit_user_id is not None
|
||
else await aresolve_path_user_id(thread_id)
|
||
)
|
||
outputs_dir = paths.sandbox_outputs_dir(
|
||
thread_id,
|
||
user_id=path_user_id,
|
||
)
|
||
scan_errors: list[str] = []
|
||
files = await file_io(
|
||
collect_outputs_listing,
|
||
outputs_dir,
|
||
errors=scan_errors,
|
||
)
|
||
if strict and scan_errors:
|
||
raise ArtifactLibraryScanIncomplete(
|
||
f"Thread artifact scan observed {len(scan_errors)} transient file error(s)"
|
||
)
|
||
except Exception as exc:
|
||
logger.warning("Failed to scan artifacts for library thread %s", thread_id, exc_info=True)
|
||
if strict:
|
||
if isinstance(exc, ArtifactLibraryScanIncomplete):
|
||
raise
|
||
raise ArtifactLibraryScanIncomplete(
|
||
"A thread artifact source could not be read completely"
|
||
) from exc
|
||
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
|
||
|
||
# Do not create one asyncio.Task per historical conversation. Batching keeps
|
||
# memory bounded for long-lived accounts while the semaphore limits disk IO.
|
||
items: list[dict] = []
|
||
for start in range(0, len(rows), _ARTIFACT_LIBRARY_SCAN_BATCH_SIZE):
|
||
nested = await asyncio.gather(
|
||
*(scan_thread(row) for row in rows[start : start + _ARTIFACT_LIBRARY_SCAN_BATCH_SIZE])
|
||
)
|
||
items.extend(item for group in nested for item in group)
|
||
await asyncio.sleep(0)
|
||
items.extend(
|
||
await _scan_strategy_project_artifacts(
|
||
strategy_task_store,
|
||
user_id=resolved_user_id,
|
||
strict=strict,
|
||
)
|
||
)
|
||
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 _resolve_artifact_library_agent_names(
|
||
items: list[dict],
|
||
agent_store,
|
||
*,
|
||
user_id: str,
|
||
) -> list[dict]:
|
||
"""Resolve agent display names once during indexing, never on every page read."""
|
||
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 and agent_store is not None:
|
||
try:
|
||
agents = await agent_store.list_by_ids(agent_ids, user_id)
|
||
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)
|
||
return [
|
||
{
|
||
**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
|
||
]
|
||
|
||
|
||
async def _run_artifact_library_sync(app, *, user_id: str, generation: str) -> None:
|
||
"""Reconcile filesystem/task metadata into the persisted index off-request."""
|
||
store = app.state.artifact_library_store
|
||
try:
|
||
gate = getattr(app.state, "artifact_library_sync_semaphore", None)
|
||
|
||
async def _scan_and_publish() -> bool:
|
||
items = await _scan_artifact_library(
|
||
app.state.thread_store,
|
||
getattr(app.state, "parallel_agent_task_store", None),
|
||
user_id=user_id,
|
||
strict=True,
|
||
)
|
||
enriched = await _resolve_artifact_library_agent_names(
|
||
items,
|
||
getattr(app.state, "agent_store", None),
|
||
user_id=user_id,
|
||
)
|
||
published = await store.finish_sync(user_id, generation, enriched)
|
||
if published:
|
||
logger.info(
|
||
"Artifact library index refreshed for user %s (%d items)",
|
||
user_id,
|
||
len(enriched),
|
||
)
|
||
return published
|
||
|
||
if gate is None:
|
||
published = await _scan_and_publish()
|
||
else:
|
||
async with gate:
|
||
published = await _scan_and_publish()
|
||
if published:
|
||
return
|
||
except asyncio.CancelledError:
|
||
try:
|
||
await asyncio.shield(
|
||
store.fail_sync(user_id, generation, "Artifact library refresh cancelled")
|
||
)
|
||
except Exception:
|
||
logger.debug("Failed to record cancelled artifact refresh", exc_info=True)
|
||
raise
|
||
except Exception:
|
||
logger.exception("Artifact library background refresh failed for user %s", user_id)
|
||
try:
|
||
await store.fail_sync(
|
||
user_id,
|
||
generation,
|
||
"资料索引同步失败，请稍后重试",
|
||
)
|
||
except Exception:
|
||
logger.exception("Failed to persist artifact library refresh failure")
|
||
|
||
|
||
def _schedule_artifact_library_sync(app, *, user_id: str, generation: str) -> None:
|
||
"""Start one tracked refresh task after its cross-worker DB lease is claimed."""
|
||
tasks = getattr(app.state, "artifact_library_sync_tasks", None)
|
||
if not isinstance(tasks, dict):
|
||
tasks = {}
|
||
app.state.artifact_library_sync_tasks = tasks
|
||
current = tasks.get(user_id)
|
||
if current is not None and not current.done():
|
||
# A newly claimed generation means the previous task outlived its DB
|
||
# lease. Cancel that stale worker; finish_sync is generation-fenced, so
|
||
# it cannot overwrite the replacement even if cancellation arrives late.
|
||
current.cancel()
|
||
task = asyncio.create_task(
|
||
_run_artifact_library_sync(app, user_id=user_id, generation=generation),
|
||
name=f"artifact-library:{user_id}",
|
||
)
|
||
tasks[user_id] = task
|
||
|
||
def _discard(done: asyncio.Task) -> None:
|
||
if tasks.get(user_id) is done:
|
||
tasks.pop(user_id, None)
|
||
|
||
task.add_done_callback(_discard)
|
||
|
||
|
||
@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 DB-indexed page immediately and refresh that index in background.
|
||
|
||
Generated files remain authoritative on disk. A low-concurrency reconciler
|
||
asynchronously projects their metadata (plus completed strategy snapshots)
|
||
into the database. Reads therefore stay bounded to one ``COUNT`` and one
|
||
paginated query even for accounts with years of conversation history.
|
||
"""
|
||
store = _artifact_library_store(request)
|
||
user_id = _request_user_id(request)
|
||
local_tasks = getattr(request.app.state, "artifact_library_sync_tasks", {})
|
||
local_task = local_tasks.get(user_id) if isinstance(local_tasks, dict) else None
|
||
generation = None
|
||
if local_task is None or local_task.done():
|
||
generation = await store.claim_sync(
|
||
user_id,
|
||
force=refresh,
|
||
stale_after_seconds=_ARTIFACT_LIBRARY_INDEX_TTL_SECONDS,
|
||
lease_seconds=1800.0,
|
||
)
|
||
if generation is not None:
|
||
_schedule_artifact_library_sync(
|
||
request.app,
|
||
user_id=user_id,
|
||
generation=generation,
|
||
)
|
||
|
||
# A refresh can commit between the page query and the state query. Compare
|
||
# the surrounding state snapshots and re-read once when that happens so a
|
||
# response never advertises "ready" while carrying the previous generation.
|
||
state_before = await store.get_sync_state(user_id)
|
||
total, page = await store.list_items(
|
||
user_id,
|
||
query=query,
|
||
file_kind=file_kind,
|
||
source_type=source_type,
|
||
offset=offset,
|
||
limit=limit,
|
||
)
|
||
state_after = await store.get_sync_state(user_id)
|
||
if state_before.get("completed_at") != state_after.get("completed_at"):
|
||
total, page = await store.list_items(
|
||
user_id,
|
||
query=query,
|
||
file_kind=file_kind,
|
||
source_type=source_type,
|
||
offset=offset,
|
||
limit=limit,
|
||
)
|
||
sync_status = str(state_after.get("status") or "idle")
|
||
if sync_status not in {"idle", "syncing", "ready", "error"}:
|
||
sync_status = "error"
|
||
return ArtifactLibraryResponse(
|
||
total=total,
|
||
offset=offset,
|
||
limit=limit,
|
||
items=[ArtifactLibraryItem.model_validate(item) for item in page],
|
||
sync_status=sync_status,
|
||
syncing=sync_status == "syncing",
|
||
indexed_at=state_after.get("completed_at"),
|
||
sync_error=state_after.get("last_error") if sync_status == "error" else None,
|
||
)
|
||
|
||
|
||
@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, response: Response = None) -> dict: # type: ignore[assignment]
|
||
"""列出 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）。
|
||
"""
|
||
with OperationTiming(
|
||
response,
|
||
logger,
|
||
"thread_artifacts",
|
||
thread_id=sanitize_log_param(thread_id),
|
||
) as timing:
|
||
with timing.phase("owner_resolve"):
|
||
user_id = await aresolve_path_user_id(thread_id)
|
||
outputs_dir = get_paths().sandbox_outputs_dir(thread_id, user_id=user_id)
|
||
# Probe outside the shared artifact-files pool so a background library
|
||
# scan cannot stall an empty hello-chat listing for several seconds.
|
||
with timing.phase("dir_probe"):
|
||
exists = await asyncio.to_thread(lambda: outputs_dir.exists() and outputs_dir.is_dir())
|
||
if not exists:
|
||
return {"thread_id": thread_id, "files": []}
|
||
with timing.phase("dir_scan"):
|
||
files = await file_io(collect_outputs_listing, outputs_dir)
|
||
return {"thread_id": thread_id, "files": files}
|
||
|
||
|
||
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 file_io(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 file_io(
|
||
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 file_io(_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 = await file_io(_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 await file_io(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.")
|
||
|
||
await file_io(actual_path.write_text, body.content, encoding="utf-8")
|
||
return PlainTextResponse(content=body.content, media_type=mime_type or "text/plain")
|