import asyncio import logging import mimetypes import os import re import tempfile 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 starlette.background import BackgroundTask 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.config.system_settings import generated_page_logo_path, load_system_settings from deerflow.html_runtime.addresses import html_map_tile_url, html_runtime_base_url from deerflow.html_runtime.offline_bundle import build_html_offline_bundle 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 _remove_temporary_bundle(path: str) -> None: try: Path(path).unlink(missing_ok=True) except OSError: logger.warning("Failed to remove temporary HTML offline bundle: %s", path, exc_info=True) async def build_html_offline_bundle_response(actual_path: Path) -> FileResponse: """Build and stream a portable ZIP for one generated HTML artifact.""" descriptor, temporary_name = tempfile.mkstemp(prefix="deerflow-html-offline-", suffix=".zip") os.close(descriptor) temporary_path = Path(temporary_name) try: branding = load_system_settings().generated_page_branding logo_path = generated_page_logo_path(branding) await file_io.run_until_complete( build_html_offline_bundle, actual_path, temporary_path, runtime_base_url=html_runtime_base_url(), map_tile_url=html_map_tile_url(), generated_logo_path=logo_path, ) except BaseException: temporary_path.unlink(missing_ok=True) raise download_name = f"{actual_path.stem}-离线包.zip" return FileResponse( temporary_path, filename=download_name, media_type="application/zip", headers={"Cache-Control": "private, no-store"}, background=BackgroundTask(_remove_temporary_bundle, str(temporary_path)), ) 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. HTML files can be exported with their DeerFlow " "runtime dependencies as a portable ZIP." ), ) @require_permission("threads", "read", owner_check=True) async def get_artifact( thread_id: str, path: str, request: Request, download: bool = False, offline_bundle: 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. offline_bundle (bool): If true for an HTML/XHTML file, returns a ZIP containing ``index.html`` plus the packaged DeerFlow JS/CSS/image runtime dependencies. 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 offline_bundle: if actual_path.suffix.lower() not in {".htm", ".html", ".xhtml"}: raise HTTPException(status_code=400, detail="Offline bundles are available for HTML files only") return await build_html_offline_bundle_response(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")