from __future__ import annotations

import copy
import io
import ipaddress
from functools import wraps
import hashlib
import json
import mimetypes
import os
from pathlib import Path, PurePosixPath
import re
import secrets
import shutil
import socket
import subprocess
import sys
import tempfile
import threading
import time
from typing import Any
from urllib.parse import urlparse
import uuid
import zipfile

import httpx

from services.config import BASE_DIR, DATA_DIR


MAX_ARCHIVE_BYTES = 25 * 1024 * 1024
MAX_FILE_BYTES = 8 * 1024 * 1024
MAX_FILES = 500
# An administrator-built export bundle carries every Skill together with its
# showcase media, so it is orders of magnitude larger than a Skill downloaded
# from a repository. Only the administrator-only import endpoint uses these
# relaxed limits; public installs keep the tight ones above.
MAX_IMPORT_ARCHIVE_BYTES = 256 * 1024 * 1024
MAX_IMPORT_FILES = 5000
MAX_RESOURCE_BYTES = 512 * 1024
MAX_SCRIPT_ARGS = 48
MAX_SCRIPT_ARG_CHARS = 2048
MAX_SCRIPT_OUTPUT_BYTES = 25 * 1024 * 1024
MEDIA_PARSER_OUTPUT_BYTES = max(
    MAX_SCRIPT_OUTPUT_BYTES,
    int(os.environ.get("CHATGPT2API_MEDIA_PARSER_MAX_BYTES", 512 * 1024 * 1024)),
)
MEDIA_PARSER_ARTIFACT_TTL_SECONDS = max(
    900,
    int(os.environ.get("CHATGPT2API_MEDIA_PARSER_TTL_SECONDS", 6 * 60 * 60)),
)
MEDIA_PARSER_OWNER_STORAGE_BYTES = max(
    MEDIA_PARSER_OUTPUT_BYTES,
    int(os.environ.get("CHATGPT2API_MEDIA_PARSER_OWNER_STORAGE_BYTES", 1024 * 1024 * 1024)),
)
MAX_SCRIPT_OUTPUT_FILES = 24
MAX_SCRIPT_STDIO_CHARS = 20000
DEFAULT_SCRIPT_TIMEOUT_SECONDS = 45
MAX_SCRIPT_TIMEOUT_SECONDS = 120
SKILLS_DIR = DATA_DIR / "skills"
BUNDLED_SKILLS_DIR = BASE_DIR / "bundled_skills"
BUNDLED_SKILL_VERSION_FILE = ".bundled-version"
SKILL_RUNS_DIR = DATA_DIR / "files" / "skills"
SAFE_NAME_RE = re.compile(r"[^a-zA-Z0-9._-]+")
FRONTMATTER_RE = re.compile(r"^---\s*\n(?P<body>.*?)\n---\s*(?:\n|$)", re.DOTALL)
TEXT_RESOURCE_EXTENSIONS = {
    ".md", ".mdx", ".txt", ".json", ".jsonl", ".yaml", ".yml", ".toml", ".ini",
    ".py", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".css", ".html", ".svg",
}
SKILL_PRESENTATION_FILE = "skill-ui.json"
SKILL_PRESENTATION_OVERRIDE_FILE = ".presentation-overrides.json"
SKILL_PRESENTATION_ASSET_DIR = ".presentation-assets"
# Updating a Skill replaces its directory wholesale. Before that happens the
# previous copy is kept so an administrator can roll back a bad release.
SKILL_BACKUP_DIR = ".backups"
MAX_SKILL_BACKUPS = 3
SKILL_EXPORT_FORMAT = "chatsam-skills-export"
SKILL_EXPORT_FORMAT_VERSION = 1
EXPORT_MANIFEST_FILE = "manifest.json"
# Administrator state that survives an overwrite: script permissions, the
# showcase overrides, uploaded showcase media and the user-facing visibility.
PRESERVED_STATE_FILES = (
    ".permissions.json",
    SKILL_PRESENTATION_OVERRIDE_FILE,
    ".visibility.json",
)
PRESERVED_STATE_DIRS = (SKILL_PRESENTATION_ASSET_DIR,)
SKILL_WORKFLOW_FILE = "skill-workflow.json"
PRESENTATION_SOURCE_FILES = {
    SKILL_PRESENTATION_FILE,
    SKILL_PRESENTATION_OVERRIDE_FILE,
    SKILL_WORKFLOW_FILE,
}
PRESENTATION_SUMMARY_KEYS = (
    "version",
    "title",
    "description",
    "category",
    "tags",
    "capabilities",
    "origin",
    "cover",
    "asset_count",
    "example_count",
    "input_count",
    "output_count",
    "workflow_count",
)
PRESENTATION_ASSET_EXTENSIONS = {
    ".png", ".jpg", ".jpeg", ".webp", ".gif", ".svg",
    ".html", ".htm", ".pdf", ".mp4", ".webm", ".mp3", ".wav",
}
PRESENTATION_MEDIA_ROOTS = {"assets", "examples", "example", "demo", "demos", "preview", "previews"}
SCRIPT_EXTENSIONS = {".py", ".js", ".mjs", ".cjs"}
USER_GUIDE_DOC_RE = re.compile(
    r"^(?:readme|usage|quickstart|getting[-_ ]started|user[-_ ]guide|how[-_ ]to)(?:[._-].*)?\.(?:md|mdx|txt)$",
    re.IGNORECASE,
)
GUIDE_CACHE_FILE = ".user-guide-cache.json"
SKILL_FAVORITES_FILE = "skill-favorites.json"
# Caches and build leftovers add nothing to an export bundle.
EXPORT_EXCLUDED_NAMES = {GUIDE_CACHE_FILE, ".DS_Store"}
EXPORT_EXCLUDED_PARTS = {"__pycache__"}
# A Skill package is not always a single directory. Repositories keep runnable
# code beside the Skill folder, or ship it in a monorepo package, and describe
# the layout in Markdown instead of in SKILL.md. These rules decide which
# documents are read while deciding what belongs to the package.
PACKAGE_DOC_RE = re.compile(
    r"^(?:readme|usage|quickstart|getting[-_ ]started|user[-_ ]guide)(?:[._-].*)?\.(?:md|mdx|txt)$",
    re.IGNORECASE,
)
PACKAGE_DOCUMENT_SUFFIXES = {".md", ".mdx", ".txt"}
MAX_PACKAGE_DOCUMENTS = 40
# ``scripts/`` already has a fixed meaning inside a Skill — it is where the
# runtime looks for controlled scripts. A repository-level ``scripts/`` holds
# development tooling, so re-homing it would produce ``scripts/scripts/`` and mix
# unrelated files into the Skill's own script list.
RESERVED_PACKAGE_ROOT_NAMES = {"scripts"}


class _KeyedRunLocks:
    """Serialize only duplicate idempotent runs inside this API process."""

    def __init__(self) -> None:
        self._guard = threading.Lock()
        self._locks: dict[str, tuple[threading.Lock, int]] = {}

    def acquire(self, key: str) -> threading.Lock:
        with self._guard:
            lock, references = self._locks.get(key, (threading.Lock(), 0))
            self._locks[key] = (lock, references + 1)
        lock.acquire()
        return lock

    def release(self, key: str, lock: threading.Lock) -> None:
        lock.release()
        with self._guard:
            current, references = self._locks.get(key, (lock, 1))
            if current is lock and references <= 1:
                self._locks.pop(key, None)
            elif current is lock:
                self._locks[key] = (lock, references - 1)


_SCRIPT_RUN_LOCKS = _KeyedRunLocks()


def _serialize_duplicate_runs(function):
    @wraps(function)
    def wrapped(self, skill_id, identity, *args, **kwargs):
        client_run_id = str(kwargs.get("client_run_id") or "").strip()
        if not client_run_id:
            return function(self, skill_id, identity, *args, **kwargs)
        owner = _safe_name(str(identity.get("id") or "user"), "user")
        key = hashlib.sha256(f"{owner}:{skill_id}:{client_run_id}".encode("utf-8")).hexdigest()
        lock = _SCRIPT_RUN_LOCKS.acquire(key)
        try:
            return function(self, skill_id, identity, *args, **kwargs)
        finally:
            _SCRIPT_RUN_LOCKS.release(key, lock)

    return wrapped


class SkillInstallError(ValueError):
    pass


def _safe_name(value: str, fallback: str = "skill") -> str:
    normalized = SAFE_NAME_RE.sub("-", value.strip()).strip("-._").lower()
    return (normalized or fallback)[:80]


def _frontmatter(skill_file: Path) -> dict[str, str]:
    try:
        text = skill_file.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as exc:
        raise SkillInstallError(f"无法读取 {skill_file.name}: {exc}") from exc
    match = FRONTMATTER_RE.match(text)
    values: dict[str, str] = {}
    if match:
        for raw_line in match.group("body").splitlines():
            key, separator, value = raw_line.partition(":")
            if separator and key.strip() in {
                "name", "description", "author", "version", "homepage",
                "repository", "source", "source_url",
            }:
                values[key.strip()] = value.strip().strip("\"'")
    values.setdefault("name", skill_file.parent.name)
    values.setdefault("description", "")
    return values


def _presentation_category(skill_id: str, name: str, description: str) -> str:
    text = f"{skill_id} {name} {description}".lower()
    if re.search(r"\b(?:image|photo|poster|banner|logo|art|draw|paint|gif|wallpaper|holographic)\b|视觉|图片|海报|作图|封面|壁纸|摄影|插画|卡片", text):
        return "视觉创作"
    if re.search(r"\b(?:ui|ux|frontend|website|web)\b|界面|交互|网页", text):
        return "界面设计"
    if re.search(r"\b(?:document|pdf|ppt|slide|sheet|excel)\b|文档|表格|演示", text):
        return "文档效率"
    if re.search(r"\b(?:search|research)\b|搜索|调研", text):
        return "搜索研究"
    return "专业工作流"


def _asset_kind(path: Path, mime_type: str) -> str:
    suffix = path.suffix.lower()
    if mime_type.startswith("image/"):
        return "image"
    if mime_type.startswith("video/"):
        return "video"
    if mime_type.startswith("audio/"):
        return "audio"
    if suffix in {".html", ".htm"}:
        return "html"
    if suffix == ".pdf":
        return "pdf"
    return "file"


def _string_list(value: Any, *, limit: int = 16) -> list[str]:
    if isinstance(value, str):
        values = [value]
    elif isinstance(value, list):
        values = [str(item) for item in value if isinstance(item, (str, int, float))]
    else:
        return []
    result: list[str] = []
    for value_item in values:
        clean = value_item.strip()
        if clean and clean not in result:
            result.append(clean[:160])
        if len(result) >= limit:
            break
    return result


def _http_url(value: object) -> str:
    clean = str(value or "").strip()
    parsed = urlparse(clean)
    return clean if parsed.scheme in {"http", "https"} and parsed.hostname else ""


def _require_public_remote_url(value: str) -> str:
    clean = _http_url(value)
    if not clean:
        raise SkillInstallError("安装说明地址必须是有效的 http/https URL")
    hostname = urlparse(clean).hostname or ""
    try:
        addresses = {item[4][0] for item in socket.getaddrinfo(hostname, None)}
    except OSError as exc:
        raise SkillInstallError("安装说明地址无法解析") from exc
    for address in addresses:
        ip = ipaddress.ip_address(address)
        if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified:
            raise SkillInstallError("安装说明地址不能指向本机或内网")
    return clean


def _skill_origin(skill_dir: Path, metadata: dict[str, str], manifest: dict[str, Any]) -> dict[str, Any]:
    install_metadata: dict[str, Any] = {}
    try:
        raw_install = json.loads((skill_dir / ".install.json").read_text(encoding="utf-8"))
        if isinstance(raw_install, dict):
            install_metadata = raw_install
    except (OSError, UnicodeError, ValueError):
        pass

    package: dict[str, Any] = {}
    try:
        raw_package = json.loads((skill_dir / "package.json").read_text(encoding="utf-8"))
        if isinstance(raw_package, dict):
            package = raw_package
    except (OSError, UnicodeError, ValueError):
        pass

    manifest_source = manifest.get("source")
    source_config = manifest_source if isinstance(manifest_source, dict) else {}
    package_repository = package.get("repository")
    if isinstance(package_repository, dict):
        package_repository = package_repository.get("url")
    installed_source = str(install_metadata.get("source") or "").strip()
    url = next((candidate for candidate in (
        _http_url(source_config.get("url")),
        _http_url(source_config.get("repository")),
        _http_url(metadata.get("source_url")),
        _http_url(metadata.get("repository")),
        _http_url(metadata.get("homepage")),
        _http_url(package_repository),
        _http_url(package.get("homepage")),
        _http_url(installed_source),
    ) if candidate), "")
    if url.startswith("git+"):
        url = url[4:]
    url = url.removesuffix(".git")

    package_author = package.get("author")
    if isinstance(package_author, dict):
        package_author = package_author.get("name")
    author = str(
        source_config.get("author")
        or manifest.get("author")
        or metadata.get("author")
        or package_author
        or ""
    ).strip()[:160]
    github_match = re.match(r"https?://github\.com/([^/]+)/([^/#?]+)", url, re.IGNORECASE)
    if not author and github_match:
        author = github_match.group(1)

    if github_match:
        source_type = "github"
        default_label = "GitHub 开源"
    elif installed_source.startswith("upload:"):
        source_type = "upload"
        default_label = "本地上传"
    elif url:
        source_type = "url"
        default_label = "外部来源"
    else:
        source_type = "local"
        default_label = "本地 Skill"
    return {
        "type": source_type,
        "label": str(source_config.get("label") or source_config.get("name") or default_label).strip()[:120],
        "url": url,
        "author": author,
        "version": str(source_config.get("version") or manifest.get("version") or metadata.get("version") or package.get("version") or "").strip()[:80],
        "open_source": bool(source_config.get("openSource", source_config.get("open_source", source_type == "github"))),
    }


def _safe_relative_path(value: str) -> Path:
    clean = str(value or "").strip().replace("\\", "/")
    path = PurePosixPath(clean)
    if not clean or path.is_absolute() or ".." in path.parts or any(not part for part in path.parts):
        raise SkillInstallError("Skill 文件路径无效")
    return Path(*path.parts)


def _safe_output_name(value: str) -> str:
    name = Path(str(value or "").replace("\\", "/")).name
    clean = re.sub(r"[^A-Za-z0-9._-]+", "-", name).strip(".-")
    if not clean:
        raise SkillInstallError("输出文件名无效")
    return clean[:160]


def _is_text_resource(path: Path) -> bool:
    return path.suffix.lower() in TEXT_RESOURCE_EXTENSIONS


def _script_kind(path: Path) -> str:
    suffix = path.suffix.lower()
    if suffix == ".py":
        return "python"
    if suffix in {".js", ".mjs", ".cjs"}:
        return "node"
    return ""


class _ScannedFile:
    """A single file discovered by one recursive walk of a Skill directory."""

    __slots__ = ("path", "relative", "name", "suffix", "size", "mtime_ns")

    def __init__(self, path: Path, relative: str, size: int, mtime_ns: int) -> None:
        self.path = path
        self.relative = relative
        self.name = path.name
        self.suffix = path.suffix.lower()
        self.size = size
        self.mtime_ns = mtime_ns

    @property
    def is_presentation_source(self) -> bool:
        return self.relative in PRESENTATION_SOURCE_FILES

    @property
    def is_user_document(self) -> bool:
        return bool(USER_GUIDE_DOC_RE.match(self.name.lower()))

    @property
    def is_presentation_asset(self) -> bool:
        return self.suffix in PRESENTATION_ASSET_EXTENSIONS


def _scan_skill_files(skill_dir: Path) -> list[_ScannedFile]:
    """Walk a Skill directory exactly once and reuse the result everywhere.

    ``Path.rglob`` is a Python-level recursive scan and every call re-stats each
    entry, so listing one Skill used to cost four full walks plus three reads of
    the same SKILL.md. ``os.scandir`` reuses the stat data the OS already handed
    back, and every consumer below shares this single result instead.
    """
    found: list[_ScannedFile] = []
    pending: list[Path] = [skill_dir]
    while pending:
        current = pending.pop()
        try:
            with os.scandir(current) as entries:
                for entry in entries:
                    try:
                        if entry.is_dir(follow_symlinks=False):
                            pending.append(Path(entry.path))
                            continue
                        if not entry.is_file():
                            continue
                        info = entry.stat()
                    except OSError:
                        continue
                    path = Path(entry.path)
                    found.append(_ScannedFile(
                        path,
                        path.relative_to(skill_dir).as_posix(),
                        info.st_size,
                        info.st_mtime_ns,
                    ))
        except OSError:
            continue
    # Preserve the historical ordering: the guide fingerprint must stay stable
    # for unchanged directories so existing cached guides remain valid.
    found.sort(key=lambda item: item.path)
    return found


def _guide_fingerprint(files: list[_ScannedFile]) -> str:
    """Cheap version of everything that can change a Skill's rendered surface."""
    digest = hashlib.sha256()
    for item in files:
        if not (
            item.relative == "SKILL.md"
            or item.is_user_document
            or item.is_presentation_source
            or item.is_presentation_asset
        ):
            continue
        digest.update(f"{item.relative}:{item.size}:{item.mtime_ns}\n".encode("utf-8"))
    return digest.hexdigest()[:20]


def _content_fingerprint(files: list[_ScannedFile]) -> str:
    """Version of every file in the directory, used to key the describe cache."""
    digest = hashlib.sha256()
    for item in files:
        digest.update(f"{item.relative}:{item.size}:{item.mtime_ns}\n".encode("utf-8"))
    return digest.hexdigest()[:20]


def _presentation_summary(presentation: dict[str, Any]) -> dict[str, Any]:
    return {
        key: copy.deepcopy(presentation[key])
        for key in PRESENTATION_SUMMARY_KEYS
        if key in presentation
    }


def _guide_document_score(path: str) -> tuple[int, str]:
    normalized = path.replace("\\", "/").lower()
    name = normalized.rsplit("/", 1)[-1]
    score = 20 if "/" in normalized else 0
    if re.search(r"zh[-_.]?(?:cn|hans)|chinese|中文", name):
        score -= 12
    if name.startswith("readme"):
        score -= 8
    if re.match(r"^(?:usage|quickstart|getting[-_ ]started|user[-_ ]guide)", name):
        score -= 6
    if normalized.startswith("_package-docs/"):
        score -= 4
    return score, normalized


def _ancestor_directories(skill_file: Path, search_root: Path) -> list[Path]:
    """Directories from the Skill folder up to and including the search root."""

    ancestors: list[Path] = []
    search_root_resolved = search_root.resolve()
    current = skill_file.parent.parent
    while True:
        try:
            current.resolve().relative_to(search_root_resolved)
        except ValueError:
            break
        ancestors.append(current)
        if current.resolve() == search_root_resolved:
            break
        current = current.parent
    return ancestors


def _package_documents(skill_file: Path, search_root: Path) -> list[Path]:
    """Return the documents that describe this Skill package.

    SKILL.md is the authoritative protocol file, but repositories also document
    how to run a Skill in ``references/*.md`` beside it and in the README above
    it. All of those can name directories the Skill needs at run time, so all of
    them are read when working out what the package contains.
    """

    documents = [skill_file]
    taken = 0
    for scanned in _scan_skill_files(skill_file.parent):
        if taken >= MAX_PACKAGE_DOCUMENTS:
            break
        if scanned.suffix not in PACKAGE_DOCUMENT_SUFFIXES:
            continue
        taken += 1
        documents.append(scanned.path)
    for ancestor in _ancestor_directories(skill_file, search_root):
        try:
            entries = sorted(ancestor.iterdir())
        except OSError:
            continue
        for candidate in entries:
            if candidate.is_file() and PACKAGE_DOC_RE.match(candidate.name):
                documents.append(candidate)
    return documents


def _referenced_package_roots(skill_file: Path, search_root: Path) -> list[Path]:
    """Find repository directories that belong to this Skill package.

    Some repositories keep SKILL.md in a small definition directory while the
    executable implementation lives beside it (for example
    ``media-parser/`` + ``astrbot_plugin_media_parser/``), and monorepo layouts
    ship a companion service in a sibling package (``skills/holo-card/`` +
    ``packages/holo-card-api/``). Installing only the SKILL.md parent produces a
    Skill that can explain commands but cannot run them, so every directory named
    as ``dir/...`` by the package documents is included when it really exists
    next to the Skill.

    Copying an ancestor would drag the Skill folder into its own ``scripts``
    directory, and copying a directory that holds another SKILL.md would install
    a second, competing Skill — both are skipped.
    """

    segments: list[str] = []
    for document in _package_documents(skill_file, search_root):
        try:
            content = document.read_text(encoding="utf-8")[:MAX_RESOURCE_BYTES]
        except (OSError, UnicodeError):
            continue
        for match in re.finditer(r"(?<![\w.:-])([A-Za-z0-9][A-Za-z0-9._-]*)/[A-Za-z0-9._-]+(?:/[A-Za-z0-9._-]+)*", content):
            segment = match.group(1)
            if segment not in segments and segment not in {skill_file.parent.name, "http", "https"}:
                segments.append(segment)
            if len(segments) >= 256:
                break
        if len(segments) >= 256:
            break

    skill_dir = skill_file.parent.resolve()
    roots: list[Path] = []
    search_root_resolved = search_root.resolve()
    for current in _ancestor_directories(skill_file, search_root):
        for segment in segments:
            candidate = (current / segment).resolve()
            try:
                candidate.relative_to(search_root_resolved)
            except ValueError:
                continue
            if not candidate.exists() or candidate in roots:
                continue
            if candidate.name in RESERVED_PACKAGE_ROOT_NAMES:
                continue
            if candidate == skill_dir or candidate in skill_dir.parents or skill_dir in candidate.parents:
                continue
            if next(candidate.rglob("SKILL.md"), None) is not None:
                continue
            roots.append(candidate)
    return roots


def _validate_zip_member(info: zipfile.ZipInfo) -> PurePosixPath:
    path = PurePosixPath(info.filename.replace("\\", "/"))
    if path.is_absolute() or ".." in path.parts:
        raise SkillInstallError(f"压缩包包含不安全路径：{info.filename}")
    mode = info.external_attr >> 16
    if mode and (mode & 0o170000) == 0o120000:
        raise SkillInstallError(f"压缩包不允许符号链接：{info.filename}")
    if info.file_size > MAX_FILE_BYTES:
        raise SkillInstallError(f"文件过大：{info.filename}")
    return path


def _is_export_excluded(relative: str) -> bool:
    parts = relative.split("/")
    if parts[-1] in EXPORT_EXCLUDED_NAMES or parts[-1].endswith(".pyc"):
        return True
    return any(part in EXPORT_EXCLUDED_PARTS for part in parts)


def _export_bundle_sources(extracted: Path) -> dict[str, dict[str, Any]]:
    """Return {skill_id: manifest entry} when the archive is an export bundle.

    A bundle produced by ``export_skills`` is the unit of migration between
    deployments: it carries each Skill's recorded origin so the importing
    deployment can keep updating it from the same place. Importing must
    therefore not relabel the Skill as a local upload — which only happens for
    archives that actually declare the bundle format.
    """

    path = extracted / EXPORT_MANIFEST_FILE
    if not path.is_file():
        return {}
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, ValueError):
        return {}
    if not isinstance(payload, dict) or payload.get("format") != SKILL_EXPORT_FORMAT:
        return {}
    entries: dict[str, dict[str, Any]] = {}
    for item in payload.get("skills") or []:
        if isinstance(item, dict) and str(item.get("id") or "").strip():
            entries[str(item["id"]).strip()] = item
    return entries


def _restore_preserved_state(target: Path, backup: Path) -> None:
    """Copy administrator state from an overwrite snapshot into the fresh install.

    A reinstall brings new code, but the administrator's decisions about the
    Skill — which scripts may run, how it is presented, which media was uploaded
    and whether end users see it — belong to the installation, not to the
    package, so they are carried across.
    """

    for name in PRESERVED_STATE_FILES:
        source = backup / name
        if source.is_file():
            shutil.copy2(source, target / name)
    for name in PRESERVED_STATE_DIRS:
        source = backup / name
        if not source.is_dir():
            continue
        destination = target / name
        if destination.exists():
            shutil.rmtree(destination, ignore_errors=True)
        shutil.copytree(source, destination)


def _format_size_limit(limit: int) -> str:
    return f"{limit // (1024 * 1024)}MB"


def _extract_archive(
    payload: bytes,
    destination: Path,
    *,
    subpath: str = "",
    max_bytes: int = MAX_ARCHIVE_BYTES,
    max_files: int = MAX_FILES,
) -> None:
    if len(payload) > max_bytes:
        raise SkillInstallError(f"ZIP 不能超过 {_format_size_limit(max_bytes)}")
    try:
        archive = zipfile.ZipFile(io.BytesIO(payload))
    except zipfile.BadZipFile as exc:
        raise SkillInstallError("上传内容不是有效 ZIP") from exc
    infos = archive.infolist()
    if subpath:
        wanted = PurePosixPath(subpath).parts
        infos = [
            info for info in infos
            if tuple(PurePosixPath(info.filename.replace("\\", "/")).parts[1:1 + len(wanted)]) == wanted
        ]
        if not infos:
            raise SkillInstallError(f"仓库中找不到技能目录：{subpath}")
    if len(infos) > max_files:
        raise SkillInstallError(f"ZIP 文件数量不能超过 {max_files}")
    total_size = sum(info.file_size for info in infos)
    if total_size > max_bytes:
        raise SkillInstallError(f"ZIP 解压后不能超过 {_format_size_limit(max_bytes)}")
    for info in infos:
        relative = _validate_zip_member(info)
        if not relative.parts:
            continue
        target = destination.joinpath(*relative.parts)
        if info.is_dir():
            target.mkdir(parents=True, exist_ok=True)
            continue
        target.parent.mkdir(parents=True, exist_ok=True)
        with archive.open(info) as source, target.open("wb") as output:
            shutil.copyfileobj(source, output)


def _github_download_candidates(url: str) -> list[tuple[str, str]]:
    parsed = urlparse(url)
    if parsed.netloc.lower() not in {"github.com", "www.github.com"}:
        return [(url, "")]
    parts = [part for part in parsed.path.split("/") if part]
    if len(parts) < 2:
        raise SkillInstallError("GitHub URL 必须包含 owner/repository")
    owner, repository = parts[0], parts[1].removesuffix(".git")
    if len(parts) >= 5 and parts[2] == "tree":
        ref = parts[3]
        subpath = "/".join(parts[4:])
        return [(f"https://codeload.github.com/{owner}/{repository}/zip/{ref}", subpath)]
    return [
        (f"https://codeload.github.com/{owner}/{repository}/zip/refs/heads/main", ""),
        (f"https://codeload.github.com/{owner}/{repository}/zip/refs/heads/master", ""),
    ]


class SkillService:
    def __init__(
        self,
        root: Path = SKILLS_DIR,
        runs_root: Path = SKILL_RUNS_DIR,
        favorites_path: Path | None = None,
    ) -> None:
        self.root = root
        self.runs_root = runs_root
        self.favorites_path = favorites_path or (
            DATA_DIR / SKILL_FAVORITES_FILE
            if root == SKILLS_DIR
            else root.parent / SKILL_FAVORITES_FILE
        )
        self._favorites_lock = threading.RLock()
        # Keyed by (resolved directory, guide version). The cache always holds the
        # full presentation; summaries are projected from it so the list view and
        # the detail view no longer build the same Skill twice.
        self._presentation_cache: dict[tuple[str, str], dict[str, Any]] = {}
        self._describe_cache: dict[tuple[str, str, bool], dict[str, Any]] = {}
        self._run_cleanup_lock = threading.RLock()
        self._last_run_cleanup_at = 0.0
        self._media_link_lock = threading.RLock()
        self._media_links: dict[str, dict[str, Any]] = {}

    def sync_bundled_skills(self, source_root: Path = BUNDLED_SKILLS_DIR) -> list[str]:
        """Install the Skills shipped with the application into runtime storage.

        Runtime storage may be a Hugging Face persistent volume, whereas the
        bundled directory lives in the immutable application image.  Refresh
        the application-owned files on every deployment while retaining the
        administrator-controlled visibility, permissions and presentation
        overrides already stored on the volume.
        """

        if not source_root.is_dir():
            return []
        self.root.mkdir(parents=True, exist_ok=True)
        synced: list[str] = []
        for source in sorted(path for path in source_root.iterdir() if path.is_dir()):
            skill_file = source / "SKILL.md"
            if not skill_file.is_file():
                continue
            metadata = _frontmatter(skill_file)
            skill_id = _safe_name(metadata["name"], _safe_name(source.name))
            target = self.root / skill_id
            digest = hashlib.sha256()
            for resource in sorted(path for path in source.rglob("*") if path.is_file()):
                relative = resource.relative_to(source).as_posix()
                digest.update(relative.encode("utf-8"))
                digest.update(b"\0")
                digest.update(resource.read_bytes())
                digest.update(b"\0")
            version = digest.hexdigest()
            version_file = target / BUNDLED_SKILL_VERSION_FILE
            if version_file.is_file() and version_file.read_text(encoding="utf-8").strip() == version:
                continue
            staging = self.root / f".{skill_id}.bundled-{os.getpid()}-{uuid.uuid4().hex[:6]}"
            previous = self.root / f".{skill_id}.previous-{os.getpid()}-{uuid.uuid4().hex[:6]}"
            try:
                shutil.copytree(source, staging)
                (staging / BUNDLED_SKILL_VERSION_FILE).write_text(version + "\n", encoding="utf-8")
                if target.is_dir():
                    _restore_preserved_state(staging, target)
                    os.replace(target, previous)
                os.replace(staging, target)
                shutil.rmtree(previous, ignore_errors=True)
                synced.append(skill_id)
            except Exception:
                shutil.rmtree(staging, ignore_errors=True)
                if previous.exists() and not target.exists():
                    os.replace(previous, target)
                raise
        self._presentation_cache.clear()
        self._describe_cache.clear()
        return synced

    def register_media_link(
        self,
        url: str,
        *,
        headers: dict[str, object] | None,
        name: str,
        media_type: str,
    ) -> dict[str, Any]:
        clean_url = _require_public_remote_url(url)
        allowed_headers = {"user-agent", "accept", "accept-language", "referer", "origin"}
        clean_headers = {
            str(key): str(value)
            for key, value in (headers or {}).items()
            if str(key).lower() in allowed_headers and str(value).strip()
        }
        token = secrets.token_urlsafe(32)
        expires_at = time.time() + MEDIA_PARSER_ARTIFACT_TTL_SECONDS
        with self._media_link_lock:
            current = time.time()
            self._media_links = {
                key: value
                for key, value in self._media_links.items()
                if float(value.get("expires_at") or 0) > current
            }
            self._media_links[token] = {
                "url": clean_url,
                "headers": clean_headers,
                "name": _safe_output_name(name),
                "media_type": str(media_type or "application/octet-stream"),
                "expires_at": expires_at,
            }
        return {
            "name": _safe_output_name(name),
            "type": str(media_type or "application/octet-stream"),
            "size": 0,
            "url": f"/api/skills/media/{token}",
            "expires_at": expires_at,
            "source_url": clean_url,
            "streamed": True,
        }

    def resolve_media_link(self, token: str) -> dict[str, Any]:
        clean_token = str(token or "").strip()
        if not re.fullmatch(r"[A-Za-z0-9_-]{32,128}", clean_token):
            raise SkillInstallError("媒体下载链接无效")
        with self._media_link_lock:
            record = self._media_links.get(clean_token)
            if not record or float(record.get("expires_at") or 0) <= time.time():
                self._media_links.pop(clean_token, None)
                raise SkillInstallError("媒体下载链接已过期，请重新解析")
            return dict(record)

    def _media_stream_artifacts(self, stdout: str) -> list[dict[str, Any]]:
        try:
            payload = json.loads(stdout)
        except (TypeError, ValueError):
            return []
        results = payload.get("results") if isinstance(payload, dict) else []
        if not isinstance(results, list):
            return []
        artifacts: list[dict[str, Any]] = []
        for result_index, result in enumerate(results):
            if not isinstance(result, dict) or result.get("error"):
                continue
            title = _safe_name(str(result.get("title") or f"media-{result_index + 1}"), f"media-{result_index + 1}")
            videos = result.get("videos") if isinstance(result.get("videos"), list) else []
            images = result.get("images") if isinstance(result.get("images"), list) else []
            for index, media in enumerate(videos):
                if not isinstance(media, dict) or not media.get("direct_url"):
                    continue
                artifacts.append(self.register_media_link(
                    str(media["direct_url"]),
                    headers=media.get("headers") if isinstance(media.get("headers"), dict) else {},
                    name=f"{title}-video-{index + 1}.mp4",
                    media_type="video/mp4",
                ))
            for index, media in enumerate(images):
                if not isinstance(media, dict) or not media.get("direct_url"):
                    continue
                artifacts.append(self.register_media_link(
                    str(media["direct_url"]),
                    headers=media.get("headers") if isinstance(media.get("headers"), dict) else {},
                    name=f"{title}-image-{index + 1}.jpg",
                    media_type="image/jpeg",
                ))
        return artifacts[:MAX_SCRIPT_OUTPUT_FILES]

    def cleanup_expired_runs(self, *, force: bool = False, now: float | None = None) -> int:
        """Remove expired temporary Skill artifacts and idempotency records."""
        current = time.time() if now is None else float(now)
        with self._run_cleanup_lock:
            monotonic_now = time.monotonic()
            if not force and monotonic_now - self._last_run_cleanup_at < 60:
                return 0
            self._last_run_cleanup_at = monotonic_now
            removed = 0
            if self.runs_root.is_dir():
                for marker in self.runs_root.glob("*/*/.expires-at"):
                    try:
                        expires_at = float(marker.read_text(encoding="utf-8").strip())
                    except (OSError, UnicodeError, ValueError):
                        continue
                    if expires_at > current:
                        continue
                    run_dir = marker.parent.resolve()
                    try:
                        run_dir.relative_to(self.runs_root.resolve())
                    except ValueError:
                        continue
                    shutil.rmtree(run_dir, ignore_errors=True)
                    removed += 1
            cache_root = self.runs_root / ".runs"
            if cache_root.is_dir():
                for cache_path in cache_root.glob("*.json"):
                    try:
                        payload = json.loads(cache_path.read_text(encoding="utf-8"))
                        expires_at = float(payload.get("expires_at") or 0) if isinstance(payload, dict) else 0
                    except (OSError, UnicodeError, ValueError, TypeError):
                        continue
                    legacy_run_dir: Path | None = None
                    if isinstance(payload, dict) and payload.get("skill_id") == "media-parser" and not expires_at:
                        artifacts = payload.get("artifacts") if isinstance(payload.get("artifacts"), list) else []
                        first_url = next((str(item.get("url") or "") for item in artifacts if isinstance(item, dict)), "")
                        prefix = "/files/skills/"
                        if first_url.startswith(prefix):
                            relative = PurePosixPath(first_url[len(prefix):])
                            if len(relative.parts) >= 3 and ".." not in relative.parts:
                                candidate = (self.runs_root / Path(*relative.parts[:2])).resolve()
                                try:
                                    candidate.relative_to(self.runs_root.resolve())
                                    if candidate.is_dir():
                                        legacy_run_dir = candidate
                                except ValueError:
                                    legacy_run_dir = None
                        if legacy_run_dir is not None:
                            expires_at = legacy_run_dir.stat().st_mtime + MEDIA_PARSER_ARTIFACT_TTL_SECONDS
                            (legacy_run_dir / ".expires-at").write_text(str(expires_at), encoding="utf-8")
                            payload["expires_at"] = expires_at
                            cache_path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8")
                    if expires_at and expires_at <= current:
                        if legacy_run_dir is not None:
                            shutil.rmtree(legacy_run_dir, ignore_errors=True)
                            removed += 1
                        cache_path.unlink(missing_ok=True)
            return removed

    def _enforce_media_owner_quota(self, owner: str) -> int:
        owner_root = (self.runs_root / owner).resolve()
        try:
            owner_root.relative_to(self.runs_root.resolve())
        except ValueError:
            return 0
        if not owner_root.is_dir():
            return 0
        runs: list[tuple[float, Path, int]] = []
        total = 0
        for marker in owner_root.glob("*/.expires-at"):
            run_dir = marker.parent
            try:
                size = sum(
                    path.stat().st_size
                    for path in run_dir.rglob("*")
                    if path.is_file() and path.name != ".expires-at"
                )
                created = run_dir.stat().st_mtime
            except OSError:
                continue
            total += size
            runs.append((created, run_dir, size))
        removed = 0
        for _created, run_dir, size in sorted(runs):
            if total <= MEDIA_PARSER_OWNER_STORAGE_BYTES:
                break
            shutil.rmtree(run_dir, ignore_errors=True)
            total -= size
            removed += 1
        return removed

    def list(self) -> list[dict[str, Any]]:
        self.root.mkdir(parents=True, exist_ok=True)
        items: list[dict[str, Any]] = []
        for skill_file in sorted(self.root.glob("*/SKILL.md")):
            # Installations are assembled in a hidden staging directory and
            # atomically renamed when complete. Never expose a partial Skill.
            if skill_file.parent.name.startswith("."):
                continue
            skill_dir = skill_file.parent
            files = _scan_skill_files(skill_dir)
            metadata = _frontmatter(skill_file)
            install_metadata = self._install_metadata(skill_dir)
            permissions = self._permissions(skill_dir)
            enabled = self._visibility(skill_dir)
            guide_version = _guide_fingerprint(files)
            presentation = self._presentation_for(
                skill_dir,
                metadata,
                guide_version=guide_version,
                files=files,
                summary_only=True,
            )
            items.append({
                "id": skill_dir.name,
                "name": metadata["name"],
                "description": metadata["description"],
                "source": str(install_metadata.get("source") or "manual"),
                "installed_at": str(install_metadata.get("installed_at") or ""),
                "file_count": len(files),
                "scripts_allowed": bool(permissions.get("allow_scripts")),
                "allowed_scripts": list(permissions.get("allowed_scripts") or []),
                "enabled": enabled,
                "guide_version": guide_version,
                "presentation_summary": presentation,
            })
        return items

    @staticmethod
    def _read_presentation_manifest(skill_dir: Path) -> dict[str, Any]:
        """Return skill-ui.json, or an empty config when it is absent or unusable."""

        path = skill_dir / SKILL_PRESENTATION_FILE
        if not path.is_file():
            return {}
        try:
            value = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeError, ValueError):
            return {}
        return value if isinstance(value, dict) else {}

    @staticmethod
    def _read_presentation_overrides(skill_dir: Path) -> dict[str, Any]:
        """Return the administrator overrides, ignoring unreadable or malformed files."""

        path = skill_dir / SKILL_PRESENTATION_OVERRIDE_FILE
        if not path.is_file():
            return {}
        try:
            value = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, UnicodeError, ValueError):
            return {}
        return value if isinstance(value, dict) else {}

    @staticmethod
    def _merge_presentation_config(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
        result = dict(base)
        for key, value in overrides.items():
            if key == "display" and isinstance(value, dict):
                current = result.get("display") if isinstance(result.get("display"), dict) else {}
                result["display"] = {**current, **value}
            else:
                result[key] = value
        return result

    @staticmethod
    def _safe_presentation_asset(skill_dir: Path, value: object) -> Path | None:
        if not isinstance(value, str) or not value.strip():
            return None
        try:
            relative = _safe_relative_path(value.strip())
        except SkillInstallError:
            return None
        target = (skill_dir / relative).resolve()
        try:
            target.relative_to(skill_dir.resolve())
        except ValueError:
            return None
        if not target.is_file() or target.stat().st_size > MAX_FILE_BYTES:
            return None
        return target

    def _build_presentation(
        self,
        skill_dir: Path,
        metadata: dict[str, str],
        *,
        files: list[_ScannedFile] | None = None,
    ) -> dict[str, Any]:
        scanned = files if files is not None else _scan_skill_files(skill_dir)
        manifest = self._read_presentation_manifest(skill_dir)
        overrides = self._read_presentation_overrides(skill_dir)
        manifest = self._merge_presentation_config(manifest, overrides)
        display = manifest.get("display") if isinstance(manifest.get("display"), dict) else {}
        origin = _skill_origin(skill_dir, metadata, manifest)
        skill_id = skill_dir.name
        name = str(display.get("title") or metadata.get("name") or skill_id).strip()
        description = str(display.get("description") or metadata.get("description") or "").strip()
        category = str(display.get("category") or _presentation_category(skill_id, name, description)).strip()
        tags = _string_list(display.get("tags") or manifest.get("tags"), limit=12)

        content = ""
        try:
            content = (skill_dir / "SKILL.md").read_text(encoding="utf-8")[:50000]
        except (OSError, UnicodeError):
            content = ""
        searchable = f"{name}\n{description}\n{content}".lower()

        capabilities = _string_list(manifest.get("capabilities"), limit=8)
        inferred_capabilities: list[str] = []
        if re.search(r"web[_ -]?search|联网|搜索", searchable):
            inferred_capabilities.append("web_search")
        if re.search(r"image[_ -]?(?:generation|edit)|生图|图生图|图片生成|图片编辑|poster|photo|logo|art", searchable):
            inferred_capabilities.append("image")
        if any((skill_dir / folder).is_dir() for folder in ("scripts",)):
            inferred_capabilities.append("script")
        if re.search(r"html|网页|interactive|交互", searchable):
            inferred_capabilities.append("html")
        if not capabilities:
            capabilities = inferred_capabilities or ["chat"]

        asset_entries: list[dict[str, Any]] = []
        seen_assets: set[str] = set()

        def add_asset(raw: object, *, default_role: str = "gallery") -> None:
            item = raw if isinstance(raw, dict) else {"path": raw}
            path_value = item.get("path") if isinstance(item, dict) else None
            target = self._safe_presentation_asset(skill_dir, path_value)
            if target is None or target.suffix.lower() not in PRESENTATION_ASSET_EXTENSIONS:
                return
            relative = target.relative_to(skill_dir.resolve()).as_posix()
            if relative in seen_assets:
                return
            seen_assets.add(relative)
            mime_type = mimetypes.guess_type(target.name)[0] or "application/octet-stream"
            asset_entries.append({
                "path": relative,
                "name": str(item.get("title") or item.get("name") or target.stem).strip()[:120],
                "description": str(item.get("description") or "").strip()[:500],
                "type": mime_type,
                "kind": _asset_kind(target, mime_type),
                "size": target.stat().st_size,
                "role": str(item.get("role") or default_role).strip()[:40],
            })

        explicit_cover = display.get("cover") or manifest.get("cover")
        if explicit_cover:
            add_asset({"path": explicit_cover, "role": "cover", "title": name}, default_role="cover")
        raw_assets = manifest.get("assets")
        if isinstance(raw_assets, dict):
            raw_assets = raw_assets.get("gallery") or raw_assets.get("items") or []
        if isinstance(raw_assets, list):
            for item in raw_assets[:24]:
                add_asset(item)

        readme_documents = sorted(
            (item for item in scanned if item.is_user_document),
            key=lambda item: _guide_document_score(item.relative),
        )[:4]
        readme_text = ""
        for readme in readme_documents:
            if readme.size > MAX_RESOURCE_BYTES:
                continue
            try:
                text_value = readme.path.read_text(encoding="utf-8")
            except (OSError, UnicodeError):
                continue
            readme_text += f"\n{text_value[:12000]}"
            for match in re.finditer(r"!\[[^\]]*\]\(([^)\s]+)(?:\s+[^)]*)?\)", text_value):
                source = match.group(1).strip("<>\"'")
                if re.match(r"^[a-z]+://", source, re.IGNORECASE):
                    continue
                try:
                    relative_to_skill = (readme.path.parent / source).resolve().relative_to(skill_dir.resolve()).as_posix()
                except (ValueError, OSError):
                    continue
                add_asset(relative_to_skill)

        if len(asset_entries) < 24:
            for item in scanned:
                if not item.is_presentation_asset:
                    continue
                parts = PurePosixPath(item.relative).parts
                if not parts or parts[0].lower() not in PRESENTATION_MEDIA_ROOTS:
                    continue
                add_asset(item.relative)
                if len(asset_entries) >= 24:
                    break

        cover = next((item for item in asset_entries if item["role"] == "cover"), None)
        if cover is None:
            cover = next((item for item in asset_entries if item["kind"] == "image"), None)
            if cover:
                cover = {**cover, "role": "cover"}

        raw_examples = manifest.get("examples") if isinstance(manifest.get("examples"), list) else []
        examples: list[dict[str, Any]] = []
        for index, raw in enumerate(raw_examples[:8]):
            if not isinstance(raw, dict):
                continue
            prompt = str(raw.get("prompt") or "").strip()
            if not prompt:
                continue
            examples.append({
                "id": str(raw.get("id") or f"example-{index + 1}")[:80],
                "title": str(raw.get("title") or f"示例 {index + 1}")[:120],
                "description": str(raw.get("description") or "")[:500],
                "prompt": prompt[:4000],
                "assets": _string_list(raw.get("assets"), limit=8),
                "attachment_hint": str(raw.get("attachmentHint") or raw.get("attachment_hint") or "")[:300],
            })
        if not examples:
            example_source = readme_text or content
            candidates = [
                match.group(1).strip()
                for match in re.finditer(r"```(?:text|prompt|markdown)?\s*\n([\s\S]{12,1200}?)\n```", example_source, re.IGNORECASE)
            ]
            candidates.extend(match.group(0).strip() for match in re.finditer(rf"\${re.escape(skill_id)}\b[^\n]{{4,500}}", example_source, re.IGNORECASE))
            for index, prompt in enumerate(candidates[:4]):
                if not prompt.startswith(f"${skill_id}"):
                    prompt = f"${skill_id} {prompt}"
                examples.append({"id": f"auto-{index + 1}", "title": f"示例 {index + 1}", "description": "", "prompt": prompt, "assets": [], "attachment_hint": ""})
        if not examples:
            examples.append({
                "id": "starter",
                "title": "从你的需求开始",
                "description": description,
                "prompt": f"${skill_id} 请根据我接下来提供的要求完成任务",
                "assets": [],
                "attachment_hint": "如任务依赖图片或文件，请一并上传。",
            })

        def normalize_specs(key: str, defaults: list[dict[str, Any]]) -> list[dict[str, Any]]:
            raw_specs = manifest.get(key)
            if not isinstance(raw_specs, list):
                return defaults
            result: list[dict[str, Any]] = []
            for index, raw in enumerate(raw_specs[:12]):
                if isinstance(raw, str):
                    raw = {"label": raw}
                if not isinstance(raw, dict) or not str(raw.get("label") or raw.get("name") or "").strip():
                    continue
                result.append({
                    "id": str(raw.get("id") or f"{key}-{index + 1}")[:80],
                    "label": str(raw.get("label") or raw.get("name"))[:120],
                    "type": str(raw.get("type") or "text")[:40],
                    "required": bool(raw.get("required", False)),
                    "description": str(raw.get("description") or "")[:500],
                })
            return result or defaults

        default_inputs = [{"id": "request", "label": "任务要求", "type": "text", "required": True, "description": "说明目标、风格、尺寸或交付要求。"}]
        input_searchable = f"{searchable}\n{readme_text.lower()}"
        has_image_input = "image" in capabilities and bool(re.search(
            r"supplied image|uploaded image|user[- ]provided image|source image|reference image|"
            r"参考图|上传图片|上传照片|原图|照片",
            input_searchable,
        ))
        # Auto-discovered image inputs are suggestions, never hard gates. An
        # administrator can explicitly mark an image input as required through
        # skill-ui.json or the presentation overrides editor.
        image_input_required = False
        if has_image_input:
            default_inputs.append({
                "id": "image",
                "label": "参考图片",
                "type": "image",
                "required": image_input_required,
                "description": "上传需要处理或参考的图片。" if image_input_required else "如需基于现有画面处理，可上传参考图片。",
            })
        inputs = normalize_specs("inputs", default_inputs)

        default_outputs: list[dict[str, Any]] = []
        if "image" in capabilities:
            default_outputs.append({"id": "image", "label": "图片结果", "type": "image", "required": False, "description": "可预览、下载并继续作为参考图使用。"})
        if "html" in capabilities:
            default_outputs.append({"id": "html", "label": "互动 HTML", "type": "html", "required": False, "description": "在安全沙箱中预览，可手动启用交互。"})
        if "script" in capabilities:
            default_outputs.append({"id": "file", "label": "文件产物", "type": "file", "required": False, "description": "由受控脚本生成的可下载文件。"})
        if not default_outputs:
            default_outputs.append({"id": "answer", "label": "结构化结果", "type": "text", "required": False, "description": "在对话中直接交付。"})
        outputs = normalize_specs("outputs", default_outputs)

        workflow: list[dict[str, Any]] = []
        workflow_summary = ""
        workflow_path = skill_dir / SKILL_WORKFLOW_FILE
        if workflow_path.is_file():
            try:
                raw_workflow = json.loads(workflow_path.read_text(encoding="utf-8"))
                workflow_summary = str(raw_workflow.get("summary") or "").strip()[:500] if isinstance(raw_workflow, dict) else ""
                tasks = raw_workflow.get("tasks") if isinstance(raw_workflow, dict) else []
                if not isinstance(tasks, list) or not tasks:
                    raise ValueError("tasks missing")
                for index, task in enumerate(tasks[:32]):
                    if not isinstance(task, dict):
                        continue
                    workflow.append({
                        "id": str(task.get("id") or f"step-{index + 1}")[:80],
                        "title": str(task.get("title") or task.get("id") or f"步骤 {index + 1}")[:120],
                        "description": str(task.get("description") or task.get("prompt") or "")[:500],
                        "capability": str(task.get("capability") or ("image" if "image" in capabilities else "task"))[:40],
                    })
                postprocess = raw_workflow.get("postprocess") if isinstance(raw_workflow, dict) else None
                if isinstance(postprocess, dict) and postprocess.get("script"):
                    workflow.append({"id": "postprocess", "title": str(postprocess.get("title") or "整理最终产物")[:120], "description": "将前面步骤的结果组装为最终文件。", "capability": "script"})
            except (OSError, UnicodeError, ValueError, TypeError):
                # A malformed workflow file simply falls back to the plain Skill flow.
                workflow = []

        result = {
            "version": 1,
            "title": name,
            "description": description,
            "category": category,
            "tags": tags,
            "capabilities": capabilities,
            "origin": origin,
            "cover": cover,
            "asset_count": len(asset_entries),
            "example_count": len(examples),
            "input_count": len(inputs),
            "output_count": len(outputs),
            "workflow_count": len(workflow),
        }
        return {
            **result,
            "assets": asset_entries,
            "examples": examples,
            "inputs": inputs,
            "outputs": outputs,
            "workflow": {"summary": workflow_summary, "steps": workflow},
            "manifest": {
                "present": bool(manifest),
                "path": SKILL_PRESENTATION_FILE,
                "overridden": bool(overrides),
                "override_path": SKILL_PRESENTATION_OVERRIDE_FILE,
            },
        }

    def _presentation_for(
        self,
        skill_dir: Path,
        metadata: dict[str, str],
        *,
        guide_version: str | None = None,
        files: list[_ScannedFile] | None = None,
        summary_only: bool = False,
    ) -> dict[str, Any]:
        """Return the cached full presentation, projecting a summary on demand.

        The summary used to live under its own cache key, which meant the list
        request and the detail request could never share a build. Both views now
        read one entry, so opening a Skill normally reuses work already done.
        """
        version = guide_version or (_guide_fingerprint(files) if files is not None else self._guide_version(skill_dir))
        key = (str(skill_dir.resolve()), version)
        cached = self._presentation_cache.get(key)
        if cached is None:
            cached = self._build_presentation(skill_dir, metadata, files=files)
            self._presentation_cache[key] = cached
            if len(self._presentation_cache) > 256:
                self._presentation_cache = dict(list(self._presentation_cache.items())[-128:])
        return _presentation_summary(cached) if summary_only else cached

    def presentation(self, skill_id: str) -> dict[str, Any]:
        skill_dir = self._skill_dir(skill_id)
        metadata = _frontmatter(skill_dir / "SKILL.md")
        return copy.deepcopy(self._presentation_for(skill_dir, metadata))

    def _invalidate_caches(self, skill_dir: Path) -> None:
        prefix = str(skill_dir.resolve())
        self._presentation_cache = {
            key: value for key, value in self._presentation_cache.items() if key[0] != prefix
        }
        self._describe_cache = {
            key: value for key, value in self._describe_cache.items() if key[0] != prefix
        }

    def reindex_presentation(self, skill_id: str) -> dict[str, Any]:
        skill_dir = self._skill_dir(skill_id)
        self._invalidate_caches(skill_dir)
        return self.presentation(skill_id)

    def update_presentation(self, skill_id: str, value: dict[str, Any]) -> dict[str, Any]:
        skill_dir = self._skill_dir(skill_id)
        allowed = {"display", "assets", "examples", "inputs", "outputs", "capabilities", "tags", "author", "version", "source"}
        current = self._read_presentation_overrides(skill_dir)
        incoming = {key: item for key, item in value.items() if key in allowed}
        payload = self._merge_presentation_config(current, incoming)
        display = payload.get("display")
        if display is not None and not isinstance(display, dict):
            raise SkillInstallError("display 必须是对象")
        if isinstance(display, dict):
            payload["display"] = {
                key: item for key, item in display.items()
                if key in {"title", "description", "category", "cover", "tags"}
            }
        for key in ("assets", "examples", "inputs", "outputs", "capabilities", "tags"):
            if key in payload and not isinstance(payload[key], list):
                raise SkillInstallError(f"{key} 必须是数组")
        path = skill_dir / SKILL_PRESENTATION_OVERRIDE_FILE
        temporary = path.with_suffix(f"{path.suffix}.{uuid.uuid4().hex}.tmp")
        temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
        os.replace(temporary, path)
        return self.reindex_presentation(skill_id)

    def upload_presentation_assets(
        self,
        skill_id: str,
        uploads: list[tuple[bytes, str, str]],
    ) -> list[dict[str, Any]]:
        skill_dir = self._skill_dir(skill_id)
        if not uploads:
            raise SkillInstallError("请选择要上传的展示图片")
        if len(uploads) > 12:
            raise SkillInstallError("一次最多上传 12 张展示图片")
        target_dir = skill_dir / SKILL_PRESENTATION_ASSET_DIR
        target_dir.mkdir(parents=True, exist_ok=True)
        results: list[dict[str, Any]] = []
        allowed_extensions = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".svg"}
        for content, original_name, content_type in uploads:
            if not content or len(content) > MAX_FILE_BYTES:
                raise SkillInstallError(f"展示图片 {original_name or '未命名文件'} 为空或超过 8MB")
            suffix = Path(original_name or "image").suffix.lower()
            if suffix not in allowed_extensions or not str(content_type or "").lower().startswith("image/"):
                raise SkillInstallError(f"展示素材只支持 PNG、JPG、WebP、GIF 和 SVG：{original_name}")
            stem = _safe_name(Path(original_name).stem, "showcase")[:48]
            filename = f"{stem}-{uuid.uuid4().hex[:10]}{suffix}"
            target = target_dir / filename
            target.write_bytes(content)
            relative = target.relative_to(skill_dir).as_posix()
            mime_type = mimetypes.guess_type(target.name)[0] or content_type or "application/octet-stream"
            results.append({
                "path": relative,
                "name": Path(original_name).stem[:120] or stem,
                "description": "",
                "type": mime_type,
                "kind": "image",
                "size": len(content),
                "role": "gallery",
            })
        self.reindex_presentation(skill_id)
        return results

    def _read_favorites(self) -> dict[str, list[str]]:
        if not self.favorites_path.is_file():
            return {}
        try:
            value = json.loads(self.favorites_path.read_text(encoding="utf-8"))
        except (OSError, UnicodeError, ValueError):
            return {}
        if not isinstance(value, dict):
            return {}
        result: dict[str, list[str]] = {}
        for owner, items in value.items():
            if isinstance(owner, str) and isinstance(items, list):
                result[owner] = [str(item) for item in items if isinstance(item, str)]
        return result

    def favorite_ids(self, owner_id: str) -> list[str]:
        owner = str(owner_id or "").strip()
        if not owner:
            return []
        with self._favorites_lock:
            known = {item["id"] for item in self.catalog()}
            return [item for item in self._read_favorites().get(owner, []) if item in known]

    def set_favorite(self, owner_id: str, skill_id: str, *, favorite: bool) -> list[str]:
        self.ensure_enabled(skill_id)
        owner = str(owner_id or "").strip()
        if not owner:
            raise SkillInstallError("用户身份无效")
        with self._favorites_lock:
            value = self._read_favorites()
            items = list(dict.fromkeys(value.get(owner, [])))
            if favorite and skill_id not in items:
                items.append(skill_id)
            if not favorite:
                items = [item for item in items if item != skill_id]
            value[owner] = items
            self.favorites_path.parent.mkdir(parents=True, exist_ok=True)
            temporary = self.favorites_path.with_suffix(f"{self.favorites_path.suffix}.{uuid.uuid4().hex}.tmp")
            temporary.write_text(json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8")
            os.replace(temporary, self.favorites_path)
            return items

    def resolve_asset(self, skill_id: str, asset_path: str) -> tuple[Path, str]:
        skill_dir = self._skill_dir(skill_id)
        target = self._safe_presentation_asset(skill_dir, asset_path)
        if target is None or target.suffix.lower() not in PRESENTATION_ASSET_EXTENSIONS:
            raise SkillInstallError("Skill 展示素材不存在、过大或类型不受支持")
        return target, mimetypes.guess_type(target.name)[0] or "application/octet-stream"

    def catalog(self, *, include_content: bool = False) -> list[dict[str, Any]]:
        items = [item for item in self.list() if item.get("enabled", True)]
        if not include_content:
            return items
        for item in items:
            skill_file = self.root / str(item["id"]) / "SKILL.md"
            try:
                item["content"] = skill_file.read_text(encoding="utf-8")[:50000]
            except (OSError, UnicodeError):
                item["content"] = ""
        return items

    def _skill_dir(self, skill_id: str) -> Path:
        safe_id = _safe_name(skill_id)
        if safe_id != skill_id or not safe_id or safe_id.startswith("."):
            raise SkillInstallError("Skill ID 无效")
        target = (self.root / safe_id).resolve()
        root = self.root.resolve()
        try:
            target.relative_to(root)
        except ValueError as exc:
            raise SkillInstallError("Skill 路径无效") from exc
        if not target.is_dir() or not (target / "SKILL.md").is_file():
            raise SkillInstallError("Skill 不存在")
        return target

    @staticmethod
    def _install_metadata(skill_dir: Path) -> dict[str, Any]:
        """Return .install.json — where the Skill came from and when."""

        path = skill_dir / ".install.json"
        if not path.is_file():
            return {}
        try:
            value = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return {}
        return value if isinstance(value, dict) else {}

    def _install_record(
        self,
        *,
        staging: Path,
        target: Path,
        source: str,
        bundle_entry: dict[str, Any] | None,
        package_roots: list[str],
    ) -> dict[str, Any]:
        """Build the .install.json written for a freshly installed Skill.

        The recorded origin is what decides whether an administrator can pull
        updates, so provenance carried by the package must survive the install:
        a migrated export bundle keeps the origin of the deployment it came
        from (that is the whole point of migrating), and re-uploading a plain
        ZIP over an existing Skill keeps that Skill's known origin instead of
        downgrading it to an anonymous local upload. Only archives that declare
        the bundle format may set their own origin — a repository that happens
        to ship a .install.json must not redirect the update address.
        """

        import datetime

        incoming = self._install_metadata(staging) if bundle_entry is not None else {}
        manifest_entry = bundle_entry or {}
        preserved_source = str(incoming.get("source") or manifest_entry.get("source") or "").strip()
        preserved_installed_at = str(incoming.get("installed_at") or manifest_entry.get("installed_at") or "").strip()
        origin = "bundle" if preserved_source else ""
        if not preserved_source and source.startswith("upload:") and target.is_dir():
            previous = self._install_metadata(target)
            previous_source = str(previous.get("source") or "").strip()
            if previous_source:
                preserved_source = previous_source
                preserved_installed_at = str(previous.get("installed_at") or "").strip()
                origin = "previous"

        record: dict[str, Any] = dict(incoming) if origin == "bundle" else {}
        record["source"] = preserved_source or source
        record["installed_at"] = (
            preserved_installed_at or datetime.datetime.now(datetime.timezone.utc).isoformat()
        )
        record["package_roots"] = package_roots
        if origin:
            record["imported_at"] = datetime.datetime.now(datetime.timezone.utc).isoformat()
            record["imported_from"] = source
            record["origin_preserved"] = origin
        return record

    @staticmethod
    def _permissions(skill_dir: Path) -> dict[str, Any]:
        path = skill_dir / ".permissions.json"
        raw: dict[str, Any] = {}
        if path.is_file():
            try:
                value = json.loads(path.read_text(encoding="utf-8"))
                if isinstance(value, dict):
                    raw = value
            except (OSError, ValueError):
                raw = {}
        allowed = [
            str(item).replace("\\", "/")
            for item in raw.get("allowed_scripts") or []
            if isinstance(item, str) and item.strip()
        ]
        try:
            timeout = min(MAX_SCRIPT_TIMEOUT_SECONDS, max(5, int(raw.get("timeout_seconds") or DEFAULT_SCRIPT_TIMEOUT_SECONDS)))
        except (TypeError, ValueError):
            timeout = DEFAULT_SCRIPT_TIMEOUT_SECONDS
        return {
            "allow_scripts": bool(raw.get("allow_scripts")),
            "allowed_scripts": allowed,
            "timeout_seconds": timeout,
        }

    @staticmethod
    def _visibility(skill_dir: Path) -> bool:
        path = skill_dir / ".visibility.json"
        if not path.is_file():
            return True
        try:
            value = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return True
        return bool(value.get("enabled", True)) if isinstance(value, dict) else True

    @staticmethod
    def _guide_version(skill_dir: Path) -> str:
        return _guide_fingerprint(_scan_skill_files(skill_dir))

    def ensure_enabled(self, skill_id: str) -> None:
        skill_dir = self._skill_dir(skill_id)
        if not self._visibility(skill_dir):
            raise SkillInstallError("Skill 已被管理员隐藏")

    def guide_context(self, skill_id: str) -> dict[str, Any]:
        skill_dir = self._skill_dir(skill_id)
        files = _scan_skill_files(skill_dir)
        metadata = _frontmatter(skill_dir / "SKILL.md")
        documents: list[dict[str, str]] = []
        candidates = sorted(
            (item for item in files if item.is_user_document),
            key=lambda item: _guide_document_score(item.relative),
        )[:4]
        for item in candidates:
            if item.size > MAX_RESOURCE_BYTES:
                continue
            try:
                content = item.path.read_text(encoding="utf-8")
            except (OSError, UnicodeError):
                continue
            documents.append({"path": item.relative, "content": content[:10000]})
        return {
            "id": skill_id,
            "name": metadata["name"],
            "description": metadata["description"],
            "version": _guide_fingerprint(files),
            "content": (skill_dir / "SKILL.md").read_text(encoding="utf-8")[:MAX_RESOURCE_BYTES],
            "documents": documents,
        }

    def read_cached_guide(self, skill_id: str) -> dict[str, Any] | None:
        skill_dir = self._skill_dir(skill_id)
        path = skill_dir / GUIDE_CACHE_FILE
        if not path.is_file():
            return None
        try:
            value = json.loads(path.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            return None
        version = self._guide_version(skill_dir)
        guide = str(value.get("guide") or "").strip() if isinstance(value, dict) else ""
        if not guide or value.get("version") != version:
            return None
        return {"guide": guide, "version": version, "cached": True}

    def write_cached_guide(self, skill_id: str, *, version: str, guide: str) -> dict[str, Any]:
        skill_dir = self._skill_dir(skill_id)
        current_version = self._guide_version(skill_dir)
        clean_guide = guide.strip()
        if version != current_version or not clean_guide:
            raise SkillInstallError("Skill 文档已更新，请重新生成使用指南")
        (skill_dir / GUIDE_CACHE_FILE).write_text(
            json.dumps({"version": current_version, "guide": clean_guide}, ensure_ascii=False),
            encoding="utf-8",
        )
        return {"guide": clean_guide, "version": current_version, "cached": False}

    def describe(self, skill_id: str, *, include_content: bool = True) -> dict[str, Any]:
        return self._describe(skill_id, include_content=include_content)[0]

    def describe_with_etag(self, skill_id: str, *, include_content: bool = True) -> tuple[dict[str, Any], str]:
        payload, version = self._describe(skill_id, include_content=include_content)
        return payload, f'"{version}"'

    def _describe(self, skill_id: str, *, include_content: bool) -> tuple[dict[str, Any], str]:
        skill_dir = self._skill_dir(skill_id)
        files = _scan_skill_files(skill_dir)
        content_version = _content_fingerprint(files)
        cache_key = (str(skill_dir.resolve()), content_version, include_content)
        cached = self._describe_cache.get(cache_key)
        if cached is not None:
            return copy.deepcopy(cached), content_version

        metadata = _frontmatter(skill_dir / "SKILL.md")
        permissions = self._permissions(skill_dir)
        resources: list[dict[str, Any]] = []
        scripts: list[dict[str, Any]] = []
        for item in files:
            if item.name in {
                ".install.json",
                ".permissions.json",
                ".visibility.json",
                GUIDE_CACHE_FILE,
                SKILL_PRESENTATION_OVERRIDE_FILE,
            }:
                continue
            kind = _script_kind(item.path)
            entry = {
                "path": item.relative,
                "size": item.size,
                "type": mimetypes.guess_type(item.name)[0] or "application/octet-stream",
                "readable": item.suffix in TEXT_RESOURCE_EXTENSIONS and item.size <= MAX_RESOURCE_BYTES,
            }
            resources.append(entry)
            if item.relative.startswith("scripts/") and kind:
                scripts.append({
                    **entry,
                    "kind": kind,
                    "allowed": bool(permissions["allow_scripts"]) and (
                        not permissions["allowed_scripts"] or item.relative in permissions["allowed_scripts"]
                    ),
                })
        result: dict[str, Any] = {
            "id": skill_id,
            "name": metadata["name"],
            "description": metadata["description"],
            "resources": resources,
            "scripts": scripts,
            "permissions": permissions,
            "enabled": self._visibility(skill_dir),
            "presentation": self._presentation_for(
                skill_dir,
                metadata,
                guide_version=_guide_fingerprint(files),
                files=files,
            ),
        }
        if include_content:
            result["content"] = (skill_dir / "SKILL.md").read_text(encoding="utf-8")[:MAX_RESOURCE_BYTES]
        self._describe_cache[cache_key] = result
        if len(self._describe_cache) > 128:
            self._describe_cache = dict(list(self._describe_cache.items())[-64:])
        return copy.deepcopy(result), content_version

    def read_resource(self, skill_id: str, resource_path: str) -> dict[str, Any]:
        skill_dir = self._skill_dir(skill_id)
        relative = _safe_relative_path(resource_path)
        target = (skill_dir / relative).resolve()
        try:
            target.relative_to(skill_dir)
        except ValueError as exc:
            raise SkillInstallError("资源路径超出 Skill 目录") from exc
        if not target.is_file():
            raise SkillInstallError("Skill 资源不存在")
        size = target.stat().st_size
        if size > MAX_RESOURCE_BYTES:
            raise SkillInstallError(f"Skill 文本资源不能超过 {MAX_RESOURCE_BYTES // 1024}KB")
        if not _is_text_resource(target):
            raise SkillInstallError("该资源不是可读取的文本文件")
        try:
            content = target.read_text(encoding="utf-8")
        except UnicodeError as exc:
            raise SkillInstallError("Skill 资源不是 UTF-8 文本") from exc
        return {
            "skill_id": skill_id,
            "path": relative.as_posix(),
            "content": content,
            "size": size,
        }

    def set_permissions(
        self,
        skill_id: str,
        *,
        allow_scripts: bool,
        allowed_scripts: list[str] | None = None,
        timeout_seconds: int = DEFAULT_SCRIPT_TIMEOUT_SECONDS,
    ) -> dict[str, Any]:
        skill_dir = self._skill_dir(skill_id)
        known_scripts = {
            path.relative_to(skill_dir).as_posix()
            for path in skill_dir.rglob("*")
            if path.is_file() and path.relative_to(skill_dir).as_posix().startswith("scripts/") and _script_kind(path)
        }
        requested: list[str] = []
        for item in allowed_scripts or []:
            relative = _safe_relative_path(item).as_posix()
            if relative not in known_scripts:
                raise SkillInstallError(f"脚本不存在或类型不受支持：{relative}")
            if relative not in requested:
                requested.append(relative)
        payload = {
            "allow_scripts": bool(allow_scripts),
            "allowed_scripts": requested,
            "timeout_seconds": min(MAX_SCRIPT_TIMEOUT_SECONDS, max(5, int(timeout_seconds))),
        }
        (skill_dir / ".permissions.json").write_text(
            json.dumps(payload, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
        return self.describe(skill_id, include_content=False)

    def set_visibility(self, skill_id: str, *, enabled: bool) -> dict[str, Any]:
        skill_dir = self._skill_dir(skill_id)
        (skill_dir / ".visibility.json").write_text(
            json.dumps({"enabled": bool(enabled)}, ensure_ascii=False, indent=2),
            encoding="utf-8",
        )
        return self.describe(skill_id, include_content=False)

    @_serialize_duplicate_runs
    def run_script(
        self,
        skill_id: str,
        identity: dict[str, object],
        *,
        script_path: str,
        args: list[str],
        uploads: list[tuple[bytes, str, str]],
        client_run_id: str = "",
    ) -> dict[str, Any]:
        self.cleanup_expired_runs()
        skill_dir = self._skill_dir(skill_id)
        permissions = self._permissions(skill_dir)
        if not permissions["allow_scripts"]:
            raise PermissionError("管理员尚未允许这个 Skill 执行脚本")
        relative = _safe_relative_path(script_path)
        relative_posix = relative.as_posix()
        if not relative_posix.startswith("scripts/"):
            raise SkillInstallError("只允许执行 Skill 的 scripts/ 目录中的脚本")
        if permissions["allowed_scripts"] and relative_posix not in permissions["allowed_scripts"]:
            raise PermissionError("该脚本不在管理员白名单中")
        script = (skill_dir / relative).resolve()
        try:
            script.relative_to(skill_dir)
        except ValueError as exc:
            raise SkillInstallError("脚本路径超出 Skill 目录") from exc
        kind = _script_kind(script)
        if not script.is_file() or not kind:
            raise SkillInstallError("脚本不存在或类型不受支持")
        if len(args) > MAX_SCRIPT_ARGS:
            raise SkillInstallError(f"脚本参数不能超过 {MAX_SCRIPT_ARGS} 个")

        owner = _safe_name(str(identity.get("id") or "user"), "user")
        clean_client_run_id = str(client_run_id or "").strip()
        if len(clean_client_run_id) > 200 or "\x00" in clean_client_run_id:
            raise SkillInstallError("client_run_id 无效")
        cache_path: Path | None = None
        if clean_client_run_id:
            cache_key = hashlib.sha256(f"{owner}:{skill_id}:{clean_client_run_id}".encode("utf-8")).hexdigest()
            cache_path = self.runs_root / ".runs" / f"{cache_key}.json"
            if cache_path.is_file():
                try:
                    cached = json.loads(cache_path.read_text(encoding="utf-8"))
                    if isinstance(cached, dict):
                        expires_at = float(cached.get("expires_at") or 0)
                        cached_artifacts = cached.get("artifacts") if isinstance(cached.get("artifacts"), list) else []
                        cached_run_id = _safe_name(str(cached.get("run_id") or ""), "")
                        def cached_artifact_exists(item: object) -> bool:
                            if not isinstance(item, dict):
                                return True
                            artifact_url = str(item.get("url") or "")
                            if item.get("streamed") and artifact_url.startswith("/api/skills/media/"):
                                try:
                                    self.resolve_media_link(artifact_url.rsplit("/", 1)[-1])
                                    return True
                                except SkillInstallError:
                                    return False
                            return (
                                self.runs_root
                                / owner
                                / cached_run_id
                                / _safe_output_name(str(item.get("name") or "artifact"))
                            ).is_file()

                        artifacts_exist = all(cached_artifact_exists(item) for item in cached_artifacts)
                        if (expires_at and expires_at <= time.time()) or not artifacts_exist:
                            cache_path.unlink(missing_ok=True)
                        else:
                            return cached
                except (OSError, ValueError):
                    pass
        run_id = f"{int(time.time())}-{uuid.uuid4().hex[:12]}"
        durable_dir = self.runs_root / owner / run_id
        with tempfile.TemporaryDirectory(prefix=f"skill-{skill_id}-") as temp_name:
            workspace = Path(temp_name).resolve()
            input_dir = workspace / "inputs"
            output_dir = workspace / "outputs"
            home_dir = workspace / "home"
            input_dir.mkdir()
            output_dir.mkdir()
            home_dir.mkdir()
            input_paths: list[Path] = []
            total_upload_bytes = 0
            for index, (data, name, _mime_type) in enumerate(uploads):
                total_upload_bytes += len(data)
                if total_upload_bytes > MAX_SCRIPT_OUTPUT_BYTES:
                    raise SkillInstallError("脚本输入文件总大小不能超过 25MB")
                safe_name = _safe_output_name(name or f"input-{index + 1}")
                target = input_dir / f"{index + 1:02d}-{safe_name}"
                target.write_bytes(data)
                input_paths.append(target)

            output_pattern = re.compile(r"\{\{output:([^{}]+)\}\}")
            input_pattern = re.compile(r"\{\{input:(\d+)\}\}")

            def resolve_argument(raw: str) -> str:
                value = str(raw)
                # Arguments are passed as an argv list without a shell.  Keep
                # NUL forbidden, but allow line breaks so parser Skills can
                # process pasted share text containing several URLs.
                if len(value) > MAX_SCRIPT_ARG_CHARS or "\x00" in value:
                    raise SkillInstallError("脚本参数无效或过长")
                used_placeholder = False

                def input_replacement(match: re.Match[str]) -> str:
                    nonlocal used_placeholder
                    used_placeholder = True
                    index = int(match.group(1))
                    if index < 0 or index >= len(input_paths):
                        raise SkillInstallError(f"输入文件索引不存在：{index}")
                    return str(input_paths[index])

                def output_replacement(match: re.Match[str]) -> str:
                    nonlocal used_placeholder
                    used_placeholder = True
                    return str(output_dir / _safe_output_name(match.group(1)))

                value = input_pattern.sub(input_replacement, value)
                value = output_pattern.sub(output_replacement, value)
                for token, replacement in {
                    "{{workspace}}": workspace,
                    "{{skill}}": skill_dir,
                    "{{outputs}}": output_dir,
                }.items():
                    if token in value:
                        used_placeholder = True
                        value = value.replace(token, str(replacement))
                if "{{" in value or "}}" in value:
                    raise SkillInstallError("脚本参数包含未知占位符")
                if not used_placeholder:
                    candidate = Path(value)
                    if candidate.is_absolute() or PurePosixPath(value.replace("\\", "/")).is_absolute() or ".." in PurePosixPath(value.replace("\\", "/")).parts:
                        raise SkillInstallError("脚本参数中的文件路径必须使用受控占位符")
                return value

            resolved_args = [resolve_argument(item) for item in args]
            if kind == "python":
                # Keep Python isolated from user/site configuration while still
                # allowing a Skill entry point to import sibling modules such as
                # scripts/core.py.  The bootstrap is fixed by the host; neither
                # the Skill nor the model can inject code into it.
                bootstrap = (
                    "import runpy,sys; "
                    "sys.path.insert(0,sys.argv[1]); "
                    "sys.argv=sys.argv[2:]; "
                    "runpy.run_path(sys.argv[0],run_name='__main__')"
                )
                command = [
                    sys.executable,
                    "-I",
                    "-c",
                    bootstrap,
                    str(script.parent),
                    str(script),
                    *resolved_args,
                ]
            else:
                node = shutil.which("node")
                if not node:
                    raise SkillInstallError("服务器没有可用的 Node.js 运行时")
                command = [node, str(script), *resolved_args]
            env = {
                "PATH": os.environ.get("PATH", ""),
                "PYTHONIOENCODING": "utf-8",
                "PYTHONUTF8": "1",
                "HOME": str(home_dir),
                "USERPROFILE": str(home_dir),
                "TEMP": str(workspace),
                "TMP": str(workspace),
                "SKILL_DIR": str(skill_dir),
                "SKILL_WORKSPACE": str(workspace),
                "SKILL_OUTPUT_DIR": str(output_dir),
            }
            # Windows needs these variables to initialise Winsock.  Omitting
            # them from the deliberately small child environment makes even
            # ``import asyncio`` fail with WinError 10106 before the Skill's
            # own code starts.  They identify system directories only and do
            # not broaden the script's data or credential access.
            if os.name == "nt":
                for name in ("SYSTEMROOT", "WINDIR", "COMSPEC", "PATHEXT"):
                    value = os.environ.get(name, "")
                    if value:
                        env[name] = value
            started_at = time.monotonic()
            try:
                completed = subprocess.run(
                    command,
                    cwd=workspace,
                    env=env,
                    stdin=subprocess.DEVNULL,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE,
                    text=True,
                    encoding="utf-8",
                    errors="replace",
                    timeout=int(permissions["timeout_seconds"]),
                    check=False,
                    creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
                )
            except subprocess.TimeoutExpired as exc:
                raise SkillInstallError(f"脚本执行超过 {permissions['timeout_seconds']} 秒，已终止") from exc

            # Empty cache markers/lock files are implementation details, not
            # user-facing artifacts.  Do not render them as bogus downloads.
            output_candidates = [
                path for path in output_dir.rglob("*")
                if path.is_file() and path.stat().st_size > 0
            ]
            if len(output_candidates) > MAX_SCRIPT_OUTPUT_FILES:
                raise SkillInstallError(f"脚本输出文件不能超过 {MAX_SCRIPT_OUTPUT_FILES} 个")
            total_output_bytes = sum(path.stat().st_size for path in output_candidates)
            output_limit = MEDIA_PARSER_OUTPUT_BYTES if skill_id.lower() == "media-parser" else MAX_SCRIPT_OUTPUT_BYTES
            if total_output_bytes > output_limit:
                raise SkillInstallError(f"脚本输出文件总大小不能超过 {output_limit // 1024 // 1024}MB")
            artifacts: list[dict[str, Any]] = []
            expires_at: float | None = None
            if output_candidates:
                durable_dir.mkdir(parents=True, exist_ok=False)
                for source in output_candidates:
                    name = _safe_output_name(source.name)
                    target = durable_dir / name
                    suffix = 2
                    while target.exists():
                        target = durable_dir / f"{Path(name).stem}-{suffix}{Path(name).suffix}"
                        suffix += 1
                    shutil.copy2(source, target)
                    try:
                        relative_url = target.relative_to(DATA_DIR / "files").as_posix()
                    except ValueError:
                        # Custom roots are used by tests and embedders. Production
                        # keeps runs under DATA_DIR/files and uses the first path.
                        relative_url = (Path("skills") / owner / run_id / target.name).as_posix()
                    artifacts.append({
                        "name": target.name,
                        "type": mimetypes.guess_type(target.name)[0] or "application/octet-stream",
                        "size": target.stat().st_size,
                        "url": f"/files/{relative_url}",
                    })
                if skill_id.lower() == "media-parser":
                    expires_at = time.time() + MEDIA_PARSER_ARTIFACT_TTL_SECONDS
                    (durable_dir / ".expires-at").write_text(str(expires_at), encoding="utf-8")
                    self._enforce_media_owner_quota(owner)
            elif skill_id.lower() == "media-parser" and completed.returncode == 0:
                artifacts.extend(self._media_stream_artifacts(completed.stdout))
                if artifacts:
                    expires_at = max(float(item.get("expires_at") or 0) for item in artifacts)
            result = {
                "run_id": run_id,
                "skill_id": skill_id,
                "script": relative_posix,
                "status": "success" if completed.returncode == 0 else "error",
                "exit_code": completed.returncode,
                "duration_ms": int((time.monotonic() - started_at) * 1000),
                "stdout": completed.stdout[-MAX_SCRIPT_STDIO_CHARS:],
                "stderr": completed.stderr[-MAX_SCRIPT_STDIO_CHARS:],
                "artifacts": artifacts,
            }
            if expires_at is not None:
                result["expires_at"] = expires_at
            if cache_path is not None:
                cache_path.parent.mkdir(parents=True, exist_ok=True)
                staging = cache_path.with_suffix(f".tmp-{os.getpid()}")
                staging.write_text(json.dumps(result, ensure_ascii=False), encoding="utf-8")
                os.replace(staging, cache_path)
            return result

    def _backup_skill(self, skill_dir: Path) -> Path:
        """Snapshot a Skill before it is overwritten and keep the newest few.

        An update is destructive: the previous files are deleted and replaced.
        Keeping a rolling snapshot means a bad release can be rolled back by
        copying the directory back, and it gives ``_restore_preserved_state`` a
        single source for the administrator state that has to survive.
        """

        import datetime

        backups_root = self.root / SKILL_BACKUP_DIR / skill_dir.name
        backups_root.mkdir(parents=True, exist_ok=True)
        stamp = datetime.datetime.now(datetime.timezone.utc).strftime("%Y%m%d-%H%M%S")
        destination = backups_root / f"{stamp}-{uuid.uuid4().hex[:6]}"
        try:
            shutil.copytree(skill_dir, destination)
        except OSError as exc:
            shutil.rmtree(destination, ignore_errors=True)
            raise SkillInstallError(f"无法备份现有 Skill，已中止覆盖：{exc}") from exc
        self._prune_skill_backups(backups_root)
        return destination

    @staticmethod
    def _prune_skill_backups(backups_root: Path) -> None:
        try:
            snapshots = sorted(
                (path for path in backups_root.iterdir() if path.is_dir()),
                key=lambda path: path.name,
            )
        except OSError:
            return
        for stale in snapshots[:-MAX_SKILL_BACKUPS]:
            shutil.rmtree(stale, ignore_errors=True)

    def install_zip(
        self,
        payload: bytes,
        *,
        source: str,
        overwrite: bool = False,
        subpath: str = "",
        max_bytes: int = MAX_ARCHIVE_BYTES,
        max_files: int = MAX_FILES,
    ) -> list[dict[str, Any]]:
        import datetime

        self.root.mkdir(parents=True, exist_ok=True)
        with tempfile.TemporaryDirectory(prefix="chatgpt2api-skill-") as temp_name:
            extracted = Path(temp_name)
            _extract_archive(payload, extracted, subpath=subpath, max_bytes=max_bytes, max_files=max_files)
            search_root = extracted
            if subpath:
                matches = [path for path in extracted.rglob(subpath.split("/")[-1]) if path.is_dir() and path.as_posix().endswith(subpath)]
                if not matches:
                    raise SkillInstallError(f"仓库中找不到技能目录：{subpath}")
                search_root = matches[0]
            skill_files = sorted(search_root.rglob("SKILL.md"))
            if not skill_files:
                raise SkillInstallError("ZIP 中没有找到 SKILL.md")
            bundle_skills = _export_bundle_sources(extracted)
            installed_ids: list[str] = []
            discovered_ids: set[str] = set()
            for skill_file in skill_files:
                # A parent skill owns its nested resources; do not install a
                # nested SKILL.md twice when a package intentionally includes one.
                if any(parent.name == "SKILL.md" for parent in skill_file.parents):
                    continue
                metadata = _frontmatter(skill_file)
                skill_id = _safe_name(metadata["name"], _safe_name(skill_file.parent.name))
                # Release snapshots and examples sometimes contain older copies
                # of the same Skill. Prefer the first (normally shallowest)
                # definition instead of overwriting a complete runtime with an
                # archived duplicate later in the same package.
                if skill_id in discovered_ids:
                    continue
                discovered_ids.add(skill_id)
                target = self.root / skill_id
                if target.exists() and not overwrite:
                    raise SkillInstallError(f"Skill 已存在：{skill_id}")
                # Overwriting deletes the previous copy outright, so snapshot it
                # first. The snapshot doubles as the source of the administrator
                # state that must survive a reinstall: script permissions,
                # showcase overrides, uploaded showcase media and visibility.
                backup_dir = self._backup_skill(target) if target.is_dir() else None
                staging = self.root / f".{skill_id}.installing-{os.getpid()}"
                if staging.exists():
                    shutil.rmtree(staging)
                try:
                    shutil.copytree(skill_file.parent, staging)
                    package_roots = _referenced_package_roots(skill_file, search_root)
                    for package_root in package_roots:
                        # External implementations become controlled Skill
                        # scripts. Keeping them below scripts/ makes them
                        # discoverable to the runtime and subject to the same
                        # administrator whitelist as native Skill scripts.
                        destination = staging / "scripts" / package_root.name
                        if package_root.is_dir():
                            shutil.copytree(package_root, destination, dirs_exist_ok=True)
                        elif package_root.is_file():
                            destination.parent.mkdir(parents=True, exist_ok=True)
                            shutil.copy2(package_root, destination)
                    # A repository may keep its human-facing README/USAGE next
                    # to (or above) the Skill directory rather than beside
                    # SKILL.md. Preserve those package docs so the chat UI can
                    # explain the Skill to end users after installation.
                    copied_doc_names: set[str] = set()
                    package_docs_dir = staging / "_package-docs"
                    for ancestor in _ancestor_directories(skill_file, search_root):
                        for candidate in sorted(ancestor.iterdir()):
                            if not candidate.is_file() or not PACKAGE_DOC_RE.match(candidate.name):
                                continue
                            normalized_name = candidate.name.lower()
                            if normalized_name in copied_doc_names:
                                continue
                            copied_doc_names.add(normalized_name)
                            package_docs_dir.mkdir(parents=True, exist_ok=True)
                            shutil.copy2(candidate, package_docs_dir / candidate.name)
                    (staging / ".install.json").write_text(json.dumps(
                        self._install_record(
                            staging=staging,
                            target=target,
                            source=source,
                            bundle_entry=bundle_skills.get(skill_id),
                            package_roots=[path.name for path in package_roots],
                        ),
                        ensure_ascii=False,
                        indent=2,
                    ), encoding="utf-8")
                    if target.exists():
                        shutil.rmtree(target)
                    os.replace(staging, target)
                    if backup_dir is not None:
                        _restore_preserved_state(target, backup_dir)
                finally:
                    if staging.exists():
                        shutil.rmtree(staging, ignore_errors=True)
                installed_ids.append(skill_id)
            if not installed_ids:
                raise SkillInstallError("没有可安装的 Skill")
        installed = {item["id"]: item for item in self.list()}
        return [installed[item] for item in installed_ids if item in installed]

    def install_url(self, url: str, *, overwrite: bool = False) -> list[dict[str, Any]]:
        clean_url = url.strip()
        parsed = urlparse(clean_url)
        if parsed.scheme not in {"http", "https"}:
            raise SkillInstallError("仅支持 http/https URL")
        last_error: Exception | None = None
        for download_url, subpath in _github_download_candidates(clean_url):
            try:
                with httpx.Client(follow_redirects=True, timeout=45) as client:
                    response = client.get(download_url, headers={"User-Agent": "chatgpt2api-skill-installer/1.0"})
                    response.raise_for_status()
                    payload = response.content
                return self.install_zip(payload, source=clean_url, overwrite=overwrite, subpath=subpath)
            except SkillInstallError as exc:
                # A valid branch/archive was already found. Do not mask an
                # installation result such as "already exists" by probing a
                # fallback branch that may return 404.
                if "Skill 已存在：" in str(exc):
                    raise
                last_error = exc
            except httpx.HTTPError as exc:
                last_error = exc
        raise SkillInstallError(f"下载或安装失败：{last_error}")

    def update_skill(self, skill_id: str) -> list[dict[str, Any]]:
        """Reinstall a Skill from the source recorded when it was installed.

        Only remotely-sourced Skills can be refreshed. A Skill that arrived as an
        uploaded archive has no address to fetch from, so the administrator has
        to supply a new archive instead.
        """

        skill_dir = self._skill_dir(skill_id)
        source = str(self._install_metadata(skill_dir).get("source") or "").strip()
        if not source or source.startswith("upload:"):
            raise SkillInstallError("这个 Skill 由本地上传安装，没有可用的远端地址，请重新上传新版 ZIP")
        if not self._is_direct_install_url(source):
            raise SkillInstallError(f"该 Skill 的来源不支持自动更新：{source}")
        installed = self.install_url(source, overwrite=True)
        self._invalidate_caches(skill_dir)
        return installed

    def export_skills(self, skill_ids: list[str] | None = None) -> tuple[bytes, str]:
        """Bundle Skills — code plus administrator state — into one archive.

        The result is the unit of migration between deployments: installing it
        elsewhere reproduces the Skills exactly, including script permissions,
        showcase overrides, uploaded showcase media and visibility, because the
        installer copies each Skill directory verbatim.
        """

        import datetime

        if skill_ids:
            directories = [self._skill_dir(item) for item in dict.fromkeys(skill_ids)]
        else:
            directories = [self._skill_dir(str(item["id"])) for item in self.list()]
        if not directories:
            raise SkillInstallError("没有可导出的 Skill")
        manifest_skills: list[dict[str, Any]] = []
        buffer = io.BytesIO()
        with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED, compresslevel=6) as archive:
            for skill_dir in directories:
                metadata = _frontmatter(skill_dir / "SKILL.md")
                install_metadata = self._install_metadata(skill_dir)
                exported = 0
                for item in _scan_skill_files(skill_dir):
                    if _is_export_excluded(item.relative):
                        continue
                    archive.write(item.path, f"{skill_dir.name}/{item.relative}")
                    exported += 1
                manifest_skills.append({
                    "id": skill_dir.name,
                    "name": metadata["name"],
                    "description": metadata["description"],
                    "source": str(install_metadata.get("source") or "manual"),
                    "installed_at": str(install_metadata.get("installed_at") or ""),
                    "file_count": exported,
                })
            archive.writestr(EXPORT_MANIFEST_FILE, json.dumps({
                "format": SKILL_EXPORT_FORMAT,
                "format_version": SKILL_EXPORT_FORMAT_VERSION,
                "exported_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
                "skill_count": len(manifest_skills),
                "skills": manifest_skills,
            }, ensure_ascii=False, indent=2))

        stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
        # Naming the file after a Skill only makes sense when one was picked
        # explicitly; a "export everything" request keeps the bundle name even
        # when the installation happens to hold a single Skill.
        filename = (
            f"skill-{directories[0].name}-{stamp}.zip"
            if skill_ids and len(directories) == 1
            else f"skills-export-{stamp}.zip"
        )
        return buffer.getvalue(), filename

    @staticmethod
    def _instruction_urls(text: str) -> list[str]:
        result: list[str] = []
        for match in re.finditer(r"https?://[^\s<>\]\[()\"']+", text, re.IGNORECASE):
            value = match.group(0).rstrip(".,;:!?，。；：！？")
            if value not in result:
                result.append(value)
        return result[:24]

    @staticmethod
    def _is_direct_install_url(value: str) -> bool:
        parsed = urlparse(value)
        path = parsed.path.lower()
        return (parsed.hostname or "").lower() == "github.com" or path.endswith(".zip")

    def _read_install_instruction(self, url: str) -> str:
        current = _require_public_remote_url(url)
        with httpx.Client(follow_redirects=False, timeout=25) as client:
            for _ in range(6):
                response = client.get(current, headers={"User-Agent": "chatgpt2api-skill-installer/1.0", "Accept": "text/markdown,text/plain;q=0.9,*/*;q=0.2"})
                if response.is_redirect:
                    location = response.headers.get("location")
                    if not location:
                        raise SkillInstallError("安装说明重定向缺少目标地址")
                    current = _require_public_remote_url(str(httpx.URL(current).join(location)))
                    continue
                response.raise_for_status()
                if len(response.content) > MAX_RESOURCE_BYTES:
                    raise SkillInstallError("安装说明文件超过 512KB 限制")
                content_type = response.headers.get("content-type", "").lower()
                if not any(value in content_type for value in ("text/", "markdown", "json", "yaml", "octet-stream")):
                    raise SkillInstallError("安装说明不是可读取的文本文件")
                try:
                    return response.content.decode("utf-8")
                except UnicodeDecodeError as exc:
                    raise SkillInstallError("安装说明必须使用 UTF-8 编码") from exc
        raise SkillInstallError("安装说明重定向次数过多")

    def install_instruction(self, instruction: str, *, overwrite: bool = False) -> dict[str, Any]:
        clean = str(instruction or "").strip()
        if not clean or len(clean) > 8000:
            raise SkillInstallError("安装请求为空或过长")
        supplied_urls = self._instruction_urls(clean)
        if not supplied_urls:
            raise SkillInstallError("没有找到安装说明、GitHub 或 ZIP 地址")

        candidates = [url for url in supplied_urls if self._is_direct_install_url(url)]
        instruction_urls = [url for url in supplied_urls if url not in candidates]
        inspected: list[str] = []
        for instruction_url in instruction_urls[:3]:
            document = self._read_install_instruction(instruction_url)
            inspected.append(instruction_url)
            for candidate in self._instruction_urls(document):
                if self._is_direct_install_url(candidate) and candidate not in candidates:
                    candidates.append(candidate)

        if not candidates:
            raise SkillInstallError("安装说明中没有找到可受控安装的 GitHub 仓库或 ZIP 地址；不会执行网页中的 shell/npx 命令")
        errors: list[str] = []
        for candidate in candidates[:8]:
            try:
                installed = self.install_url(candidate, overwrite=overwrite)
                return {"installed": installed, "resolved_url": candidate, "instruction_urls": inspected}
            except SkillInstallError as exc:
                existing_match = re.search(r"Skill 已存在：([a-z0-9._-]+)", str(exc), re.IGNORECASE)
                if existing_match and not overwrite:
                    existing = next((item for item in self.list() if item["id"] == existing_match.group(1)), None)
                    if existing:
                        return {"installed": [existing], "resolved_url": candidate, "instruction_urls": inspected, "already_installed": True}
                errors.append(str(exc))
        raise SkillInstallError(f"安装说明已读取，但候选 Skill 均安装失败：{'；'.join(errors[-3:])}")

    def delete(self, skill_id: str) -> None:
        target = self._skill_dir(skill_id)
        shutil.rmtree(target)


skill_service = SkillService()