"""Materialize the model-facing, online-only skills filesystem view.

The canonical skills tree is administration state: it intentionally keeps an
offline skill on disk so that it can be brought online again.  It therefore
must not also be the filesystem capability handed to a model sandbox.  This
module builds a separate, read-only-at-the-mount-boundary view containing only
currently enabled skills.

The view lives in the writable DeerFlow runtime directory, not below the
canonical root (which production commonly mounts read-only).  Files are copied,
not linked: LocalSandbox executes as the backend user, so a model modifying the
disposable view must never mutate canonical skill files through a hard link.

When the metadata cache is enabled, sandbox tools do not stat the whole tree
on every call. A command that never names ``/mnt/skills`` reuses the published
view. A command that names a skill reconciles only that skill (and a support
library it declares). Skill-management writes call ``sync_skill_sandbox_skill``
so the copy matches before the next command. ``refresh_skill_sandbox_view``
remains the full rebuild used on a cold start or an availability mismatch.
"""

from __future__ import annotations

import errno
import hashlib
import json
import logging
import os
import shutil
import tempfile
import threading
import time
import uuid
from collections.abc import Callable, Iterator
from contextlib import contextmanager
from contextvars import ContextVar, Token
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from functools import cache
from pathlib import Path
from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from deerflow.skills.storage import SkillStorage

logger = logging.getLogger(__name__)

SKILL_VIEW_DIR_NAME = ".sandbox-online-skills"
SKILL_VIEW_MARKER = ".deerflow-skill-view.json"
SKILL_VIEW_PROTOCOL = "deerflow-online-skills-v1"
SKILL_SUPPORT_MANIFEST = ".deerflow-skill-support.json"

_TOKEN_FILE_NAME = ".sandbox-online-skills.token"
_LOCK_FILE_NAME = ".sandbox-online-skills.lock"
# Reentrant: sandbox setup can refresh the view from inside an in-progress
# refresh on the same thread. A plain Lock self-waits until the timeout and
# the tool call fails with "skill view process lock".
_PROCESS_LOCK = threading.RLock()
_VIEW_LOCK_DEPTH = threading.local()
# Warm metadata checks stay short so a stuck validator cannot pin every tool
# worker. A cold rebuild of the full catalogue (common on a server right after
# deploy, and when many task-analysis seats read skills together) is allowed
# longer; overlapping callers share that one rebuild instead of queueing on
# the process lock and timing out.
_VIEW_WAIT_SECONDS = 30.0
_REBUILD_WAIT_SECONDS = 120.0
_CACHE_FILE_NAME = ".sandbox-online-skills.cache.json"
_CACHE_FORMAT = 1
_VIEW_CACHE: dict[tuple[str, str, str], _ViewCacheEntry] = {}
_EVENT_LOCK = threading.Lock()
_SKILL_VIEW_EVENTS: list[dict] = []
_EVENT_RETENTION = timedelta(days=1)
_EVENT_HARD_CAP = 20000
_EVENT_STORE_NAME = "skill-view-diagnostics.jsonl"
_events_loaded = False
_events_disk_enabled = True
_flush_lock = threading.Lock()
_flush_pending = False
_flush_running = False
_SLOW_CHECK_MS = 3000
_COMMAND_LIMIT = 500
# Armed only around one sandbox tool call. Notes stay private until that call returns.
_TRACE_ARMED: ContextVar[bool] = ContextVar("deerflow_skill_view_trace_armed", default=False)
_PENDING_TRACE: ContextVar[dict | None] = ContextVar("deerflow_skill_view_pending_trace", default=None)
_BUILTIN_SUPPORT_DEPENDENCIES: dict[tuple[str, str], frozenset[str]] = {
    ("public", "task-report-lib"): frozenset(
        {
            "task-report-import",
            "task-report-build",
            "task-report-enemy",
            "task-report-our",
            "task-report-env",
            "task-report-judge",
        }
    ),
}


@dataclass(frozen=True)
class SkillSandboxView:
    """A refreshed online-only filesystem capability."""

    root: Path
    token: str
    enabled_names: frozenset[str]
    enabled_roots: tuple[tuple[str, str, str], ...] = ()
    generation: str = ""

    @property
    def marker_content(self) -> str:
        marker = {"protocol": SKILL_VIEW_PROTOCOL, "token": self.token}
        if self.generation:
            marker["generation"] = self.generation
        return json.dumps(marker, sort_keys=True)


# Metadata is checked on every invocation, not on a TTL. Directory metadata
# detects new/deleted entries; ctime and inode detect same-size, preserved-mtime
# edits and atomic replacements. No skill contents are retained in this cache.
_Metadata = tuple[int, int, int, int, int, int] | None
_TreeSnapshot = tuple[tuple[str, _Metadata], ...]


@dataclass(frozen=True)
class _ViewCacheEntry:
    view: SkillSandboxView
    config_key: str
    source: _TreeSnapshot
    target: _TreeSnapshot


@dataclass
class _WarmValidationFlight:
    """One process-local validation shared only while it is in flight."""

    done: threading.Event = field(default_factory=threading.Event)
    result: bool | None = None
    error: BaseException | None = None
    followers: int = 0


_WARM_VALIDATION_LOCK = threading.Lock()
_WARM_VALIDATION_FLIGHTS: dict[tuple[tuple[str, str, str], str, int], _WarmValidationFlight] = {}


@dataclass
class _RefreshFlight:
    done: threading.Event = field(default_factory=threading.Event)
    result: SkillSandboxView | None = None
    error: BaseException | None = None
    followers: int = 0


_REFRESH_FLIGHTS_LOCK = threading.Lock()
_REFRESH_FLIGHTS: dict[tuple[tuple[str, str, str], str], _RefreshFlight] = {}


def _process_lock_held() -> bool:
    """True when this process's view lock is owned, including pre-3.14 RLock."""
    probe = getattr(_PROCESS_LOCK, "locked", None)
    if probe is not None:
        return bool(probe())
    if _PROCESS_LOCK.acquire(blocking=False):
        _PROCESS_LOCK.release()
        return False
    return True


def _refresh_singleflight(
    key: tuple[str, str, str],
    config_key: str,
    refresh: Callable[[], SkillSandboxView],
    *,
    require_cache: bool = True,
) -> SkillSandboxView:
    """Share one in-flight refresh, including its final metadata validation.

    Cold/invalidated callers previously queued behind the file lock and each
    revalidated the entire published tree while holding it. A large catalogue
    therefore multiplied one validation by the number of waiting tool workers.
    Only overlapping callers share this result; later calls validate afresh.
    """
    flight_key = (key, config_key)
    with _REFRESH_FLIGHTS_LOCK:
        flight = _REFRESH_FLIGHTS.get(flight_key)
        leader = flight is None
        if flight is None:
            flight = _RefreshFlight()
            _REFRESH_FLIGHTS[flight_key] = flight
        else:
            flight.followers += 1

    if not leader:
        if not flight.done.wait(timeout=_REBUILD_WAIT_SECONDS):
            raise TimeoutError("Timed out waiting for skill view refresh; try again after the current refresh finishes")
        if flight.error is not None:
            raise flight.error
        result = flight.result
        if result is None:
            raise RuntimeError("Skill view refresh completed without a result")
        # Another refresh (including one with a different availability hash)
        # may already have replaced or revoked the shared capability. Never
        # hand out the earlier view in that case; the caller can retry.
        if require_cache:
            current = _VIEW_CACHE.get(key)
            if current is None or current.view is not result or _config_key() != current.config_key:
                raise RuntimeError("Skill view changed while waiting for refresh; retry the tool")
        if (result.root / SKILL_VIEW_MARKER).read_text(encoding="utf-8") != result.marker_content:
            raise RuntimeError("Skill view capability changed while waiting for refresh; retry the tool")
        return result

    try:
        flight.result = refresh()
        return flight.result
    except BaseException as exc:
        flight.error = exc
        raise
    finally:
        with _REFRESH_FLIGHTS_LOCK:
            if _REFRESH_FLIGHTS.get(flight_key) is flight:
                _REFRESH_FLIGHTS.pop(flight_key, None)
        flight.done.set()


def _skill_view_cache_enabled() -> bool:
    from deerflow.config.runtime_performance_config import get_runtime_performance_config

    return get_runtime_performance_config().skill_view_cache_enabled


def reset_skill_sandbox_view_cache() -> None:
    """Drop only this process's metadata cache (force=True also rebuilds disk)."""
    _VIEW_CACHE.clear()


def reset_skill_view_events() -> None:
    """Drop timing samples. Disk replay stays off in this process so tests do not write the data dir."""
    global _events_loaded, _events_disk_enabled
    with _EVENT_LOCK:
        _SKILL_VIEW_EVENTS.clear()
        _events_loaded = True
    _events_disk_enabled = False
    _PENDING_TRACE.set(None)


def _event_time(event: dict) -> datetime:
    raw = str(event.get("at") or "")
    try:
        parsed = datetime.fromisoformat(raw)
    except ValueError:
        return datetime.min.replace(tzinfo=UTC)
    if parsed.tzinfo is None:
        return parsed.replace(tzinfo=UTC)
    return parsed.astimezone(UTC)


def _purge_events_locked() -> None:
    cutoff = datetime.now(UTC) - _EVENT_RETENTION
    kept = [event for event in _SKILL_VIEW_EVENTS if _event_time(event) >= cutoff]
    if len(kept) > _EVENT_HARD_CAP:
        kept = kept[-_EVENT_HARD_CAP:]
    if len(kept) != len(_SKILL_VIEW_EVENTS):
        _SKILL_VIEW_EVENTS[:] = kept


def _store_path() -> Path:
    from deerflow.config.paths import get_paths

    return get_paths().base_dir / _EVENT_STORE_NAME


def _read_stored_events() -> list[dict]:
    path = _store_path()
    if not path.is_file():
        return []
    kept: list[dict] = []
    cutoff = datetime.now(UTC) - _EVENT_RETENTION
    for line in path.read_text(encoding="utf-8").splitlines():
        if not line.strip():
            continue
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue
        if isinstance(event, dict) and _event_time(event) >= cutoff:
            kept.append(event)
    return kept


def _ensure_events_loaded() -> None:
    global _events_loaded
    if _events_loaded or not _events_disk_enabled:
        _events_loaded = True
        return
    try:
        stored = _read_stored_events()
    except Exception:
        logger.debug("skill view diagnostic history was not loaded", exc_info=True)
        stored = []
    with _EVENT_LOCK:
        if _events_loaded:
            return
        current = list(_SKILL_VIEW_EVENTS)
        _SKILL_VIEW_EVENTS[:] = [*stored, *current]
        _purge_events_locked()
        _events_loaded = True


def _write_event_snapshot(events: list[dict]) -> None:
    path = _store_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=".skill-view-diagnostics.", suffix=".tmp")
    try:
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            for event in events:
                handle.write(json.dumps(event, ensure_ascii=False) + "\n")
        os.replace(tmp, path)
    except Exception:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise


def _flush_events_loop() -> None:
    global _flush_pending, _flush_running
    try:
        while True:
            with _flush_lock:
                if not _flush_pending:
                    _flush_running = False
                    return
                _flush_pending = False
            with _EVENT_LOCK:
                snapshot = list(_SKILL_VIEW_EVENTS)
            _write_event_snapshot(snapshot)
    except Exception:
        logger.debug("skill view diagnostic history was not saved", exc_info=True)
        with _flush_lock:
            _flush_running = False


def _schedule_event_flush() -> None:
    global _flush_pending, _flush_running
    if not _events_disk_enabled:
        return
    with _flush_lock:
        _flush_pending = True
        if _flush_running:
            return
        _flush_running = True
    threading.Thread(target=_flush_events_loop, name="skill-view-diagnostics", daemon=True).start()


def begin_skill_view_trace() -> Token[bool]:
    """Start holding this tool call's check until the call itself returns."""
    _PENDING_TRACE.set(None)
    return _TRACE_ARMED.set(True)


def end_skill_view_trace(token: Token[bool]) -> None:
    _TRACE_ARMED.reset(token)
    _PENDING_TRACE.set(None)


def note_skill_view_check(**fields) -> None:
    """Remember the check. Publishing waits until the tool call returns."""
    if not _TRACE_ARMED.get():
        return
    current = _PENDING_TRACE.get()
    if current is None:
        current = {}
        _PENDING_TRACE.set(current)
    if "status" not in current:
        current.update(fields)


def remember_invoked_skill(name: str) -> None:
    """Keep a skill name that this call already parsed. No extra directory walk."""
    text = str(name or "").strip()
    if not text or not _TRACE_ARMED.get():
        return
    current = _PENDING_TRACE.get()
    if current is None:
        current = {}
        _PENDING_TRACE.set(current)
    names = current.setdefault("skill_names", [])
    if text not in names:
        names.append(text)


def skill_names_for_paths(paths: list[str], *, container_path: str = "/mnt/skills") -> list[str]:
    """Resolve invoked skill names from an already published view, else the path."""
    from deerflow.sandbox.skill_view_scope import get_active_skill_view_scope

    scope = get_active_skill_view_scope()
    view = scope.view if scope is not None else None
    prefix = container_path.replace("\\", "/").rstrip("/")
    found: list[str] = []
    for path in paths:
        normalized = str(path).replace("\\", "/").rstrip("/")
        matched = False
        if view is not None and normalized.startswith(prefix):
            for name, category, relative in view.enabled_roots:
                if not relative or relative == ".":
                    continue
                root = f"{prefix}/{category}/{relative}".rstrip("/")
                if normalized == root or normalized.startswith(f"{root}/"):
                    if name not in found:
                        found.append(name)
                    matched = True
                    break
        if matched or not normalized.startswith(prefix):
            continue
        parts = [part for part in normalized[len(prefix) :].split("/") if part and part != ".."]
        if len(parts) >= 2 and parts[0] in {"public", "custom"} and parts[1] not in found:
            found.append(parts[1])
    return found


def commit_skill_view_trace(
    *,
    tool: str,
    command: str,
    skill_names: list[str] | None = None,
    tool_ms: float = 0,
    allow_empty: bool = False,
) -> None:
    """Publish one sample after the tool call has finished. Failures stay here."""
    try:
        pending = _PENDING_TRACE.get()
        _PENDING_TRACE.set(None)
        if not pending and not allow_empty:
            return
        pending = dict(pending or {})
        if "status" not in pending:
            if not allow_empty:
                return
            pending["status"] = "skipped"
        names: list[str] = []
        remembered = pending.get("skill_names") or []
        incoming = remembered if remembered else (skill_names or [])
        for name in incoming:
            text = str(name).strip()
            if text and text not in names:
                names.append(text)
        record_skill_view_event(
            status=str(pending.get("status") or "skipped"),
            total_ms=float(pending.get("total_ms") or 0),
            lock_wait_ms=float(pending.get("lock_wait_ms") or 0),
            refresh_ms=float(pending.get("refresh_ms") or 0),
            source_entries=int(pending.get("source_entries") or 0),
            target_entries=int(pending.get("target_entries") or 0),
            enabled_count=int(pending.get("enabled_count") or 0),
            path_count=int(pending.get("path_count") or 0),
            tool=tool,
            command=command,
            skill_names=names,
            tool_ms=tool_ms,
        )
    except Exception:
        logger.debug("skill view diagnostic was not saved", exc_info=True)
        _PENDING_TRACE.set(None)


def record_skill_view_event(
    *,
    status: str,
    total_ms: float = 0,
    lock_wait_ms: float = 0,
    refresh_ms: float = 0,
    source_entries: int = 0,
    target_entries: int = 0,
    enabled_count: int = 0,
    path_count: int = 0,
    tool: str = "",
    command: str = "",
    skill_names: list[str] | None = None,
    tool_ms: float = 0,
) -> None:
    """Keep one day of samples. The disk copy is written off this call."""
    text = " ".join(str(command or "").split())
    if len(text) > _COMMAND_LIMIT:
        text = text[: _COMMAND_LIMIT - 1] + "…"
    names: list[str] = []
    for name in skill_names or []:
        item = str(name).strip()
        if item and item not in names:
            names.append(item)
    event = {
        "at": datetime.now(UTC).isoformat(timespec="seconds"),
        "status": status,
        "total_ms": round(total_ms, 1),
        "lock_wait_ms": round(lock_wait_ms, 1),
        "refresh_ms": round(refresh_ms, 1),
        "source_entries": source_entries,
        "target_entries": target_entries,
        "enabled_count": enabled_count,
        "path_count": path_count,
        "tool": tool,
        "command": text,
        "skill_names": names,
        "tool_ms": round(tool_ms, 1),
    }
    try:
        _ensure_events_loaded()
        with _EVENT_LOCK:
            _SKILL_VIEW_EVENTS.append(event)
            _purge_events_locked()
        _schedule_event_flush()
    except Exception:
        logger.debug("skill view diagnostic was not saved", exc_info=True)


def recent_skill_view_events() -> list[dict]:
    """Newest first. Entries older than one day are already dropped."""
    try:
        _ensure_events_loaded()
    except Exception:
        logger.debug("skill view diagnostic history was not loaded", exc_info=True)
    with _EVENT_LOCK:
        _purge_events_locked()
        return list(reversed(_SKILL_VIEW_EVENTS))


def _event_command(event: dict) -> str:
    return " ".join(str(event.get("command") or "").split())


def _event_skill_names(event: dict) -> list[str]:
    raw = event.get("skill_names") or []
    if isinstance(raw, str):
        raw = [raw]
    names: list[str] = []
    for item in raw:
        text = str(item).strip()
        if text and text not in names:
            names.append(text)
    return names


def explain_skill_view_event(event: dict) -> dict[str, str]:
    """Say where this one sample's time went, and the likely reason."""
    status = str(event.get("status") or "")
    total_ms = float(event.get("total_ms") or 0)
    lock_wait_ms = float(event.get("lock_wait_ms") or 0)
    source_entries = int(event.get("source_entries") or 0)
    path_count = int(event.get("path_count") or 0)
    skill_names = _event_skill_names(event)
    command = _event_command(event)
    slow = total_ms >= _SLOW_CHECK_MS
    if status == "failed":
        return {
            "level": "bad",
            "headline": "技能副本没有建成",
            "look_at": "Gateway 日志里这条 status=failed 前面的异常",
            "cause": "启停配置读失败，或复制过程中断。这次技能调用会报错，而不是在跑脚本。",
        }
    if lock_wait_ms >= _SLOW_CHECK_MS:
        return {
            "level": "warn",
            "headline": f"有 {lock_wait_ms / 1000:.1f} 秒在等别的刷新",
            "look_at": "同一时刻是否还有别的对话正在重建技能副本",
            "cause": "副本锁被占用。等当前这次重建结束后，后面的命令不应再排这么久。",
        }
    if status == "skipped":
        return {
            "level": "ok" if not slow else "warn",
            "headline": "这条命令没有点名技能，检查已跳过",
            "look_at": f"这条命令：{command}" if command else "命令本身：检索、数据转换或外部接口",
            "cause": "界面上如果仍等了很久，时间不在技能副本。看这条命令实际在做什么。",
        }
    if status == "narrow":
        named = "、".join(skill_names) if skill_names else (f"{path_count} 处技能路径" if path_count else "点名的技能")
        if slow:
            return {
                "level": "warn",
                "headline": f"只核对了{named}，但仍花了 {total_ms / 1000:.1f} 秒",
                "look_at": "这次点名的那一个技能目录里有多少文件",
                "cause": "其他技能没有参与。这一个技能里文件很多，或数据卷读写慢时，单技能核对也会变长。",
            }
        return {
            "level": "ok",
            "headline": f"只核对了{named}，检查用了 {total_ms:.0f} 毫秒",
            "look_at": "命令本身：检索、数据转换或外部接口",
            "cause": "技能副本检查已经很快。界面上更长的等待是脚本自己的运行时间。",
        }
    if status in {"refreshed", "hit"} and slow:
        scope = f"，扫了 {source_entries} 个文件记录" if source_entries else ""
        action = "完整重建了技能副本" if status == "refreshed" else "仍在逐文件核对整棵技能树"
        return {
            "level": "bad",
            "headline": f"{action}{scope}，用了 {total_ms / 1000:.1f} 秒",
            "look_at": "是否刚重启、技能开关是否为 false、技能目录里的文件总数、数据卷是否慢",
            "cause": "这是全量检查。冷启动或启停配置对不上时会出现一次；若每条命令都这样，说明热路径没有生效。",
        }
    if status == "joined":
        return {
            "level": "ok",
            "headline": "赶上了别人刚建好的副本",
            "look_at": "如果前面有一次很慢的 refreshed，看那一次的原因",
            "cause": "这次没有自己再扫整棵树。",
        }
    if status in {"refreshed", "hit"}:
        return {
            "level": "ok",
            "headline": f"全量检查用了 {total_ms:.0f} 毫秒",
            "look_at": "若只有启动后的第一条这样，后面应变成跳过或只核对一个技能",
            "cause": "这一次全量检查不慢。持续变慢要看后面的记录是不是又回到 refreshed。",
        }
    return {
        "level": "warn",
        "headline": f"未识别的检查状态 {status or '空'}",
        "look_at": "Gateway 日志中的 SKILL_VIEW_CACHE",
        "cause": "状态不在已知范围内，先看原始日志。",
    }


def diagnose_skill_view(*, cache_enabled: bool, reuse_enabled: bool, events: list[dict] | None = None) -> dict:
    """Summarize recent samples for the diagnostic page.

    The two skill-view switches stay on unless config.yaml sets them false.
    The page does not show that card; a closed switch shows up as repeated
    full scans in the samples instead.
    """
    del cache_enabled, reuse_enabled
    samples = list(recent_skill_view_events() if events is None else events)
    checks: list[dict[str, str]] = []

    if not samples:
        checks.append(
            {
                "id": "samples",
                "title": "最近的技能检查",
                "result": "还没有记录",
                "level": "empty",
                "look_at": "先去问答里执行一条会慢的命令，再回到本页",
                "cause": "记录只留在当前后台进程里。重启后要重新跑一条命令才会出现。",
            }
        )
        conclusion = "还没有可对照的耗时。先跑一条命令，再看本页最新一条落在检查还是脚本上。"
        level = "empty"
    else:
        latest = samples[0]
        explained = explain_skill_view_event(latest)
        checks.append(
            {
                "id": "latest",
                "title": "最近一次检查",
                "result": explained["headline"],
                "level": explained["level"],
                "look_at": explained["look_at"],
                "cause": explained["cause"],
                "command": _event_command(latest),
                "skill_names": _event_skill_names(latest),
            }
        )
        full_slow = [item for item in samples[:12] if item.get("status") in {"refreshed", "hit"} and float(item.get("total_ms") or 0) >= _SLOW_CHECK_MS]
        fast_path = [item for item in samples[:12] if item.get("status") in {"skipped", "narrow"}]
        if len(full_slow) >= 2:
            checks.append(
                {
                    "id": "repeat",
                    "title": "是不是每条命令都在全量检查",
                    "result": f"最近 12 条里有 {len(full_slow)} 条慢的全量检查",
                    "level": "bad",
                    "look_at": "运行的是不是这次更新后的后台",
                    "cause": "偶发一次是冷启动。连续出现说明热路径没生效，或启停配置一直对不上。",
                }
            )
        elif fast_path and not full_slow:
            checks.append(
                {
                    "id": "repeat",
                    "title": "是不是每条命令都在全量检查",
                    "result": "最近没有慢的全量检查",
                    "level": "ok",
                    "look_at": "慢的那条命令自己的脚本和外部接口",
                    "cause": "副本检查已经让开。剩下的等待去看命令在检索、转换还是在等外部系统。",
                }
            )
        conclusion = explained["headline"]
        level = explained["level"]

    explained_events = []
    for event in samples:
        item = dict(event)
        item["explanation"] = explain_skill_view_event(event)
        explained_events.append(item)
    return {"level": level, "conclusion": conclusion, "checks": checks, "events": explained_events}


def query_skill_view_events(
    *,
    page: int = 1,
    page_size: int = 20,
    sort: str = "time",
    order: str = "desc",
    cache_enabled: bool = True,
    reuse_enabled: bool = True,
) -> dict:
    """Return one page of the last day, plus the summary built from newest-first samples."""
    try:
        _ensure_events_loaded()
    except Exception:
        logger.debug("skill view diagnostic history was not loaded", exc_info=True)
    with _EVENT_LOCK:
        _purge_events_locked()
        items = list(_SKILL_VIEW_EVENTS)
    newest = sorted(items, key=lambda item: str(item.get("at") or ""), reverse=True)
    report = diagnose_skill_view(cache_enabled=cache_enabled, reuse_enabled=reuse_enabled, events=newest)
    chosen = sort if sort in {"time", "duration", "check"} else "time"
    direction = "asc" if order == "asc" else "desc"

    def sort_key(item: dict) -> tuple | str:
        if chosen == "duration":
            return (float(item.get("tool_ms") or 0), str(item.get("at") or ""))
        if chosen == "check":
            return (float(item.get("total_ms") or 0), str(item.get("at") or ""))
        return str(item.get("at") or "")

    ordered = sorted(items, key=sort_key, reverse=direction == "desc")
    size = min(100, max(1, int(page_size)))
    total = len(ordered)
    pages = max(1, (total + size - 1) // size)
    current = min(max(1, int(page)), pages)
    start = (current - 1) * size
    sliced = []
    for event in ordered[start : start + size]:
        item = dict(event)
        item["explanation"] = explain_skill_view_event(event)
        sliced.append(item)
    report["events"] = sliced
    report["total"] = total
    report["page"] = current
    report["page_size"] = size
    report["sort"] = chosen
    report["order"] = direction
    return report


@cache
def _windows_file_api():
    """Initialize read-only Win32 metadata functions once per process."""
    import ctypes
    from ctypes import wintypes

    class FileBasicInfo(ctypes.Structure):
        _fields_ = [
            ("creation", ctypes.c_longlong),
            ("access", ctypes.c_longlong),
            ("write", ctypes.c_longlong),
            ("change", ctypes.c_longlong),
            ("attributes", wintypes.DWORD),
        ]

    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
    create = kernel.CreateFileW
    create.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, wintypes.LPVOID, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE]
    create.restype = wintypes.HANDLE
    get_info = kernel.GetFileInformationByHandleEx
    get_info.argtypes = [wintypes.HANDLE, ctypes.c_int, wintypes.LPVOID, wintypes.DWORD]
    get_info.restype = wintypes.BOOL
    close = kernel.CloseHandle
    close.argtypes = [wintypes.HANDLE]
    close.restype = wintypes.BOOL
    return create, get_info, close, FileBasicInfo


def _windows_change_time(path: Path) -> int:
    """Windows st_ctime is birth time; use the real NTFS metadata-change time."""
    import ctypes

    create, get_info, close, basic_info = _windows_file_api()
    # FILE_READ_ATTRIBUTES, share read/write/delete, OPEN_EXISTING, and open
    # reparse points/directories without following them (matching lstat).
    handle = create(str(path), 0x80, 7, None, 3, 0x02200000, None)
    if handle == ctypes.c_void_p(-1).value:
        raise ctypes.WinError(ctypes.get_last_error())
    try:
        info = basic_info()
        if not get_info(handle, 0, ctypes.byref(info), ctypes.sizeof(info)):
            raise ctypes.WinError(ctypes.get_last_error())
        return info.change * 100
    finally:
        close(handle)


def _metadata(path: Path, *, root: bool = False) -> _Metadata:
    try:
        value = path.lstat()
    except FileNotFoundError:
        return None
    return (
        value.st_mode,
        value.st_dev,
        value.st_ino,
        0 if root else value.st_size,
        0 if root else value.st_mtime_ns,
        0 if root else (_windows_change_time(path) if os.name == "nt" else value.st_ctime_ns),
    )


def _scan_tree(root: Path, *, source: bool) -> _TreeSnapshot:
    records: dict[str, _Metadata] = {".": _metadata(root, root=source)}
    starts = [root / "public", root / "custom"] if source else [root]
    for start in starts:
        relative = start.relative_to(root).as_posix()
        records[relative] = _metadata(start, root=source and start == root)
        if records[relative] is None or not start.is_dir():
            continue
        visited: set[Path] = set()

        def fail_walk(error: OSError) -> None:
            raise error

        for current, directories, files in os.walk(start, followlinks=source, onerror=fail_walk):
            current_path = Path(current)
            resolved = current_path.resolve()
            if not resolved.is_relative_to(root) or resolved in visited:
                directories[:] = []
                continue
            visited.add(resolved)
            if source and resolved != current_path:
                # Catalogue discovery permits aliases to another directory
                # inside the canonical root. Track the referent's directory
                # metadata as well as the alias inode and its child files.
                records[resolved.relative_to(root).as_posix()] = _metadata(resolved)
            for name in [*directories, *files]:
                path = current_path / name
                records[path.relative_to(root).as_posix()] = _metadata(path)
            if not source:
                directories[:] = [name for name in directories if not _is_view_link(current_path / name)]
    return tuple(sorted(records.items()))


def _snapshot_matches(root: Path, snapshot: _TreeSnapshot, *, source: bool) -> bool:
    return all(_metadata(root / relative, root=source and relative == ".") == expected for relative, expected in snapshot)


def _config_key() -> str:
    """Hash the fresh small availability document, including its resolved path."""
    from deerflow.config.extensions_config import ExtensionsConfig

    path = ExtensionsConfig.resolve_config_path()
    if path is None:
        return hashlib.sha256(b"no-extensions-config").hexdigest()
    payload = Path(path).read_bytes()
    return hashlib.sha256(os.fsencode(str(Path(path).resolve())) + b"\0" + payload).hexdigest()


def _cache_key(skills_root: Path, runtime_root: Path, storage: SkillStorage) -> tuple[str, str, str]:
    storage_type = type(storage)
    return (str(skills_root), str(runtime_root), f"{storage_type.__module__}.{storage_type.__qualname__}")


def _entry_matches(entry: _ViewCacheEntry, skills_root: Path, config_key: str) -> bool:
    if entry.config_key != config_key:
        return False
    marker_path = entry.view.root / SKILL_VIEW_MARKER
    try:
        if not _snapshot_matches(skills_root, entry.source, source=True):
            return False
        if not _snapshot_matches(entry.view.root, entry.target, source=False):
            return False
        return marker_path.read_text(encoding="utf-8") == entry.view.marker_content and _config_key() == config_key
    except OSError:
        return False


def _entry_matches_singleflight(
    key: tuple[str, str, str],
    entry: _ViewCacheEntry,
    skills_root: Path,
    config_key: str,
) -> bool:
    """Coalesce only concurrently overlapping validation of one cache entry.

    There is deliberately no completed-result cache here. The leader removes
    the flight before waking followers, so a later caller always starts a new
    metadata validation. The fresh extensions-config hash is part of the key;
    callers that observed different availability bytes can never join.
    """

    flight_key = (key, config_key, id(entry))
    with _WARM_VALIDATION_LOCK:
        flight = _WARM_VALIDATION_FLIGHTS.get(flight_key)
        leader = flight is None
        if flight is None:
            flight = _WarmValidationFlight()
            _WARM_VALIDATION_FLIGHTS[flight_key] = flight
        else:
            flight.followers += 1

    if not leader:
        if not flight.done.wait(timeout=_VIEW_WAIT_SECONDS):
            raise TimeoutError("Timed out waiting for skill view validation; try again after the current refresh finishes")
        if flight.error is not None:
            raise flight.error
        if flight.result is None:
            raise RuntimeError("Warm skill-view validation completed without a result")
        return flight.result

    try:
        flight.result = _entry_matches(entry, skills_root, config_key)
        return flight.result
    except BaseException as exc:
        flight.error = exc
        raise
    finally:
        # Drop the flight before publishing completion: calls which arrive
        # after this validation finished form a new batch and must revalidate.
        with _WARM_VALIDATION_LOCK:
            if _WARM_VALIDATION_FLIGHTS.get(flight_key) is flight:
                _WARM_VALIDATION_FLIGHTS.pop(flight_key, None)
        flight.done.set()


def _atomic_text(path: Path, content: str) -> None:
    temp_path: Path | None = None
    try:
        with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as output:
            temp_path = Path(output.name)
            output.write(content)
        os.replace(temp_path, path)
        temp_path = None
    finally:
        if temp_path is not None:
            temp_path.unlink(missing_ok=True)


def _save_cache(runtime_root: Path, key: tuple[str, str, str], entry: _ViewCacheEntry) -> None:
    _atomic_text(
        runtime_root / _CACHE_FILE_NAME,
        json.dumps(
            {
                "format": _CACHE_FORMAT,
                "key": key,
                "config_key": entry.config_key,
                "token": entry.view.token,
                "generation": entry.view.generation,
                "enabled_roots": entry.view.enabled_roots,
                "source": entry.source,
                "target": entry.target,
            },
            separators=(",", ":"),
        ),
    )


def _load_cache(runtime_root: Path, key: tuple[str, str, str]) -> _ViewCacheEntry | None:
    """Reuse the last published generation in other gateway processes."""
    try:
        payload = json.loads((runtime_root / _CACHE_FILE_NAME).read_text(encoding="utf-8"))
        if payload["format"] != _CACHE_FORMAT or tuple(payload["key"]) != key:
            return None

        def snapshot(value) -> _TreeSnapshot:
            result = []
            for relative, metadata in value:
                path = Path(relative)
                if path.is_absolute() or ".." in path.parts or not isinstance(relative, str):
                    raise ValueError("Invalid cached relative path")
                if metadata is not None and (len(metadata) != 6 or not all(type(item) is int for item in metadata)):
                    raise ValueError("Invalid cached metadata")
                result.append((relative, tuple(metadata) if metadata is not None else None))
            if not result or result[0][0] != ".":
                raise ValueError("Missing cached root")
            return tuple(result)

        roots = tuple(tuple(item) for item in payload["enabled_roots"])
        if any(len(item) != 3 or not all(isinstance(part, str) for part in item) for item in roots):
            return None
        token = str(uuid.UUID(payload["token"]))
        generation = payload["generation"]
        if not isinstance(generation, str) or len(generation) != 64:
            return None
        view = SkillSandboxView(runtime_root / SKILL_VIEW_DIR_NAME, token, frozenset(item[0] for item in roots), roots, generation)
        return _ViewCacheEntry(view, payload["config_key"], snapshot(payload["source"]), snapshot(payload["target"]))
    except (OSError, ValueError, KeyError, TypeError):
        return None


def _log_cache(status: str, started: float, *, lock_wait_ms: float = 0, refresh_ms: float = 0, entry: _ViewCacheEntry | None = None) -> None:
    total_ms = (time.perf_counter() - started) * 1000
    note_skill_view_check(
        status=status,
        total_ms=total_ms,
        lock_wait_ms=lock_wait_ms,
        refresh_ms=refresh_ms,
        source_entries=len(entry.source) if entry else 0,
        target_entries=len(entry.target) if entry else 0,
        enabled_count=len(entry.view.enabled_names) if entry else 0,
    )
    log = logger.debug if status == "hit" and total_ms < 500 else logger.info
    log(
        "SKILL_VIEW_CACHE status=%s lock_wait_ms=%.1f refresh_ms=%.1f total_ms=%.1f source_entries=%d target_entries=%d enabled_count=%d",
        status,
        lock_wait_ms,
        refresh_ms,
        total_ms,
        len(entry.source) if entry else 0,
        len(entry.target) if entry else 0,
        len(entry.view.enabled_names) if entry else 0,
    )


def _lock_file(lock_file, *, deadline: float) -> None:
    if os.name == "nt":
        import msvcrt

        lock_file.seek(0, os.SEEK_END)
        if lock_file.tell() == 0:
            lock_file.write(b"\0")
            lock_file.flush()
        lock_file.seek(0)

        def try_lock() -> None:
            lock_file.seek(0)
            msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1)

    else:
        import fcntl

        def try_lock() -> None:
            fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)

    while True:
        try:
            try_lock()
            return
        except OSError as exc:
            if exc.errno not in {errno.EACCES, errno.EAGAIN, errno.EDEADLK}:
                raise
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise TimeoutError("Timed out waiting for skill view file lock") from exc
            time.sleep(min(0.05, remaining))


def _unlock_file(lock_file) -> None:
    if os.name == "nt":
        import msvcrt

        lock_file.seek(0)
        msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
        return

    import fcntl

    fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)


@contextmanager
def _view_lock(runtime_root: Path) -> Iterator[None]:
    """Serialize view refreshes in this process and across gateway workers.

    The same thread may re-enter (provider setup calls refresh while a refresh
    already holds the lock). Nested entry does not take either lock again, so
    it cannot wait out the timeout against itself.
    """
    depth = getattr(_VIEW_LOCK_DEPTH, "depth", 0)
    if depth:
        _VIEW_LOCK_DEPTH.depth = depth + 1
        try:
            yield
        finally:
            _VIEW_LOCK_DEPTH.depth = depth
        return

    lock_path = runtime_root / _LOCK_FILE_NAME
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    if not _PROCESS_LOCK.acquire(timeout=_REBUILD_WAIT_SECONDS):
        raise TimeoutError("Timed out waiting for skill view process lock")
    _VIEW_LOCK_DEPTH.depth = 1
    try:
        with lock_path.open("a+b") as lock_file:
            # Budget the file lock on its own. Sharing the process-lock budget
            # made a caller that just acquired the process lock fail immediately
            # on the file lock.
            _lock_file(lock_file, deadline=time.monotonic() + _REBUILD_WAIT_SECONDS)
            try:
                yield
            finally:
                _unlock_file(lock_file)
    finally:
        _VIEW_LOCK_DEPTH.depth = 0
        _PROCESS_LOCK.release()


def _load_or_create_token(runtime_root: Path) -> str:
    token_path = runtime_root / _TOKEN_FILE_NAME
    try:
        token = token_path.read_text(encoding="ascii").strip()
        uuid.UUID(token)
        return token
    except (FileNotFoundError, OSError, UnicodeError, ValueError):
        token = str(uuid.uuid4())
        token_path.write_text(token + "\n", encoding="ascii")
        return token


def _support_roots(skills_root: Path, enabled_names: frozenset[str]) -> list[Path]:
    """Return explicitly declared internal libraries needed by online skills.

    Cross-skill code is exceptional: ordinary skills are self-contained.  A
    directory without ``SKILL.md`` is included only when it carries a manifest
    whose ``required_by`` list intersects the enabled skill set.  This prevents
    an arbitrary unregistered directory from becoming model-readable merely by
    being placed next to a skill.
    """
    roots: list[Path] = []
    for category in ("public", "custom"):
        category_root = skills_root / category
        if not category_root.is_dir():
            continue
        for candidate in category_root.iterdir():
            if not candidate.is_dir() or candidate.is_symlink():
                continue
            manifest_path = candidate / SKILL_SUPPORT_MANIFEST
            dependencies = set(_BUILTIN_SUPPORT_DEPENDENCIES.get((category, candidate.name), ()))
            if manifest_path.is_file():
                try:
                    payload = json.loads(manifest_path.read_text(encoding="utf-8"))
                    required_by = payload.get("required_by", []) if isinstance(payload, dict) else []
                    if not isinstance(required_by, list):
                        raise ValueError("required_by must be a list")
                    dependencies.update(str(name).strip() for name in required_by if str(name).strip())
                except (OSError, UnicodeError, ValueError) as exc:
                    logger.warning("Ignoring invalid skill support manifest %s: %s", manifest_path, exc)
            if dependencies & enabled_names:
                roots.append(candidate)
    return roots


def _add_tree(
    source_root: Path,
    skills_root: Path,
    desired_files: dict[Path, Path],
    desired_dirs: set[Path],
) -> None:
    """Add regular files below ``source_root`` to the desired view snapshot."""
    resolved_skills_root = skills_root.resolve()
    resolved_source = source_root.resolve()
    try:
        source_relative = resolved_source.relative_to(resolved_skills_root)
    except ValueError as exc:
        raise RuntimeError(f"Skill source escapes configured root: {source_root}") from exc

    desired_dirs.add(source_relative)
    desired_dirs.update(parent for parent in source_relative.parents if parent != Path("."))
    for current_root, dir_names, file_names in os.walk(resolved_source, followlinks=False):
        current = Path(current_root)
        relative_dir = current.relative_to(resolved_skills_root)
        desired_dirs.add(relative_dir)

        # Symlinks are not copied or followed.  Following one could turn a
        # harmless online skill into a path to an offline/out-of-tree file.
        retained_dirs: list[str] = []
        for name in sorted(dir_names):
            child = current / name
            if child.is_symlink():
                logger.warning("Skipping symlink in model-facing skill view: %s", child)
                continue
            retained_dirs.append(name)
            desired_dirs.add(relative_dir / name)
        dir_names[:] = retained_dirs

        for name in sorted(file_names):
            source = current / name
            if source.is_symlink() or not source.is_file():
                logger.warning("Skipping non-regular file in model-facing skill view: %s", source)
                continue
            desired_files[relative_dir / name] = source


def _same_file_snapshot(source: Path, destination: Path) -> bool:
    try:
        source_stat = source.stat()
        destination_stat = destination.stat()
    except OSError:
        return False
    return source_stat.st_size == destination_stat.st_size and source_stat.st_mtime_ns == destination_stat.st_mtime_ns


def _copy_file_atomic(source: Path, destination: Path) -> None:
    destination.parent.mkdir(parents=True, exist_ok=True)
    temp_path: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(delete=False, dir=str(destination.parent)) as temp_file:
            temp_path = Path(temp_file.name)
        shutil.copy2(source, temp_path)
        os.replace(temp_path, destination)
        temp_path = None
    finally:
        if temp_path is not None:
            temp_path.unlink(missing_ok=True)


def _remove_stale_entries(
    view_root: Path,
    desired_files: set[Path],
    desired_dirs: set[Path],
) -> None:
    marker = Path(SKILL_VIEW_MARKER)
    entries = sorted(
        view_root.rglob("*"),
        key=lambda path: len(path.relative_to(view_root).parts),
        reverse=True,
    )
    for entry in entries:
        relative = entry.relative_to(view_root)
        if relative == marker:
            continue
        if entry.is_symlink():
            entry.unlink(missing_ok=True)
            continue
        if entry.is_dir():
            if relative not in desired_dirs:
                shutil.rmtree(entry)
            continue
        if relative not in desired_files:
            entry.unlink(missing_ok=True)


def _clear_view(view_root: Path) -> None:
    if not view_root.exists():
        return
    for entry in list(view_root.iterdir()):
        if entry.is_dir() and not entry.is_symlink():
            shutil.rmtree(entry)
        else:
            entry.unlink(missing_ok=True)


def _refresh_uncached_skill_sandbox_view(*, storage: SkillStorage | None = None) -> SkillSandboxView:
    """Refresh and return the online-only skills view.

    Availability is loaded fresh through ``SkillStorage.load_skills`` on every
    call.  Any failure clears the capability marker and fails closed; callers
    must not fall back to mounting the canonical skills directory.
    """
    if storage is None:
        from deerflow.skills.storage import get_or_new_skill_storage

        storage = get_or_new_skill_storage()

    from deerflow.config.paths import get_paths

    skills_root = Path(storage.get_skills_root_path()).resolve()
    runtime_root = get_paths().base_dir
    runtime_root.mkdir(parents=True, exist_ok=True)
    view_root = runtime_root / SKILL_VIEW_DIR_NAME

    with _view_lock(runtime_root):
        view_root.mkdir(parents=True, exist_ok=True)
        marker_path = view_root / SKILL_VIEW_MARKER
        try:
            # Ordinary catalogue discovery deliberately degrades open when the
            # extensions document is unreadable. A model capability must fail
            # closed, so validate the switch document explicitly and derive
            # the enabled subset ourselves.
            from deerflow.config.extensions_config import ExtensionsConfig

            extensions = ExtensionsConfig.from_file()
            discovered_skills = storage.load_skills(enabled_only=False)
            enabled_skills = [skill for skill in discovered_skills if extensions.is_skill_enabled(skill.name, skill.category)]
            enabled_names = frozenset(skill.name for skill in enabled_skills)
            token = _load_or_create_token(runtime_root)

            desired_files: dict[Path, Path] = {}
            desired_dirs: set[Path] = {Path("public"), Path("custom")}
            for skill in enabled_skills:
                _add_tree(Path(skill.skill_dir), skills_root, desired_files, desired_dirs)
            for support_root in _support_roots(skills_root, enabled_names):
                _add_tree(support_root, skills_root, desired_files, desired_dirs)

            for relative_dir in sorted(desired_dirs, key=lambda path: len(path.parts)):
                (view_root / relative_dir).mkdir(parents=True, exist_ok=True)
            for relative, source in desired_files.items():
                destination = view_root / relative
                if not _same_file_snapshot(source, destination):
                    _copy_file_atomic(source, destination)

            _remove_stale_entries(view_root, set(desired_files), desired_dirs)
            enabled_roots = tuple((skill.name, str(skill.category), skill.relative_path.as_posix()) for skill in enabled_skills)
            view = SkillSandboxView(root=view_root, token=token, enabled_names=enabled_names, enabled_roots=enabled_roots)
            marker_path.write_text(view.marker_content, encoding="utf-8")
            return view
        except Exception:
            # Without a marker an AIO sandbox cannot attest that it mounted the
            # filtered capability.  Clear stale content as an additional guard
            # for LocalSandbox and then propagate the failure.
            _clear_view(view_root)
            raise


def _clear_cached_view(view_root: Path) -> None:
    # A local tool can replace a view directory with a symlink. Never follow it
    # while clearing the disposable capability.
    if _is_view_link(view_root):
        _unlink_view_link(view_root)
    else:
        _remove_cached_stale_entries(view_root, set(), set())
        (view_root / SKILL_VIEW_MARKER).unlink(missing_ok=True)


def _is_view_link(path: Path) -> bool:
    return path.is_symlink() or (os.name == "nt" and path.is_junction())


def _unlink_view_link(path: Path) -> None:
    if os.name == "nt" and path.is_junction():
        path.rmdir()
    else:
        path.unlink(missing_ok=True)


def _remove_cached_stale_entries(view_root: Path, desired_files: set[Path], desired_dirs: set[Path]) -> None:
    """Do not traverse Windows junctions while reconciling a disposable view."""
    if not view_root.exists():
        return
    entries: list[Path] = []
    for current, directories, files in os.walk(view_root, followlinks=False):
        current_path = Path(current)
        entries.extend(current_path / name for name in [*directories, *files])
        directories[:] = [name for name in directories if not _is_view_link(current_path / name)]
    for entry in sorted(entries, key=lambda path: len(path.parts), reverse=True):
        relative = entry.relative_to(view_root)
        if relative == Path(SKILL_VIEW_MARKER):
            continue
        if _is_view_link(entry):
            _unlink_view_link(entry)
        elif entry.is_dir():
            if relative not in desired_dirs:
                entry.rmdir()
        elif relative not in desired_files:
            entry.unlink(missing_ok=True)


def _refresh_cached_skill_sandbox_view(*, storage: SkillStorage | None, force: bool) -> SkillSandboxView:
    from deerflow.config.paths import get_paths
    from deerflow.skills.storage.local_skill_storage import LocalSkillStorage

    if storage is None:
        from deerflow.skills.storage import get_or_new_skill_storage

        storage = get_or_new_skill_storage()
    if type(storage) is not LocalSkillStorage:
        # Custom reflection backends can have instance-specific discovery
        # semantics that no local filesystem fingerprint can represent.
        return _refresh_uncached_skill_sandbox_view(storage=storage)
    started = time.perf_counter()
    skills_root = Path(storage.get_skills_root_path()).resolve()
    runtime_root = get_paths().base_dir.resolve()
    runtime_root.mkdir(parents=True, exist_ok=True)
    key = _cache_key(skills_root, runtime_root, storage)
    candidate = _VIEW_CACHE.get(key)
    config_key = None
    try:
        config_key = _config_key()
        if not force and candidate is not None and _entry_matches_singleflight(key, candidate, skills_root, config_key):
            _log_cache("hit", started, entry=candidate)
            return candidate.view
    except OSError:
        # The locked path below validates again and clears a stale capability
        # if the availability document cannot be read.
        config_key = None

    def refresh() -> SkillSandboxView:
        return _refresh_cached_view_locked(storage, skills_root, runtime_root, key, candidate, force, started)

    # A missing extensions file used to skip singleflight, so every parallel
    # tool call queued on the process lock and the later ones timed out while
    # the first one copied the catalogue. Share that rebuild too.
    flight_config = config_key if config_key is not None else "extensions-unavailable"
    if force:
        flight_config = f"force:{flight_config}"
    return _refresh_singleflight(key, flight_config, refresh)


def _refresh_cached_view_locked(
    storage: SkillStorage,
    skills_root: Path,
    runtime_root: Path,
    key: tuple[str, str, str],
    candidate: _ViewCacheEntry | None,
    force: bool,
    started: float,
) -> SkillSandboxView:
    from deerflow.config.extensions_config import ExtensionsConfig

    view_root = runtime_root / SKILL_VIEW_DIR_NAME
    marker_path = view_root / SKILL_VIEW_MARKER

    wait_started = time.perf_counter()
    with _view_lock(runtime_root):
        lock_wait_ms = (time.perf_counter() - wait_started) * 1000
        refresh_started = time.perf_counter()
        try:
            config_key = _config_key()
            if not force:
                # Another thread/process may have published the same update
                # while this caller waited. Never rebuild it a second time.
                candidates = [_VIEW_CACHE.get(key), _load_cache(runtime_root, key)]
                for current in candidates:
                    if current is not None and _entry_matches(current, skills_root, config_key):
                        _VIEW_CACHE[key] = current
                        _log_cache("joined", started, lock_wait_ms=lock_wait_ms, entry=current)
                        return current.view
                candidate = next((entry for entry in candidates if entry is not None), candidate)

            if _is_view_link(view_root):
                _unlink_view_link(view_root)
            elif view_root.exists() and not view_root.is_dir():
                view_root.unlink()
            view_root.mkdir(parents=True, exist_ok=True)
            marker_path.unlink(missing_ok=True)
            for _attempt in range(3):
                config_key = _config_key()
                source_snapshot = _scan_tree(skills_root, source=True)
                extensions = ExtensionsConfig.from_file()
                discovered = storage.load_skills(enabled_only=False)
                enabled = [skill for skill in discovered if extensions.is_skill_enabled(skill.name, skill.category)]
                enabled_names = frozenset(skill.name for skill in enabled)
                enabled_roots = tuple((skill.name, str(skill.category), skill.relative_path.as_posix()) for skill in enabled)
                desired_files: dict[Path, Path] = {}
                desired_dirs: set[Path] = {Path("public"), Path("custom")}
                for skill in enabled:
                    _add_tree(Path(skill.skill_dir), skills_root, desired_files, desired_dirs)
                for support_root in _support_roots(skills_root, enabled_names):
                    _add_tree(support_root, skills_root, desired_files, desired_dirs)

                # Remove links before creating directories or copying through
                # their parents, then repair only changed source/target files.
                _remove_cached_stale_entries(view_root, set(desired_files), desired_dirs)
                for relative in sorted(desired_dirs, key=lambda path: len(path.parts)):
                    destination = view_root / relative
                    if _is_view_link(destination):
                        _unlink_view_link(destination)
                    elif destination.exists() and not destination.is_dir():
                        destination.unlink()
                    destination.mkdir(parents=True, exist_ok=True)
                previous_source = dict(candidate.source) if candidate is not None and not force else {}
                previous_target = dict(candidate.target) if candidate is not None and not force else {}
                for relative, source in desired_files.items():
                    destination = view_root / relative
                    source_relative = source.relative_to(skills_root).as_posix()
                    target_relative = relative.as_posix()
                    unchanged = source_relative in previous_source and target_relative in previous_target and _metadata(source) == previous_source[source_relative] and _metadata(destination) == previous_target[target_relative]
                    if not unchanged:
                        _copy_file_atomic(source, destination)
                if _config_key() != config_key or not _snapshot_matches(skills_root, source_snapshot, source=True):
                    candidate = None
                    continue

                generation = hashlib.sha256(json.dumps([key, config_key, source_snapshot], separators=(",", ":")).encode()).hexdigest()
                view = SkillSandboxView(view_root, _load_or_create_token(runtime_root), enabled_names, enabled_roots, generation)
                _atomic_text(marker_path, view.marker_content)
                entry = _ViewCacheEntry(view, config_key, source_snapshot, _scan_tree(view_root, source=False))
                if not _entry_matches(entry, skills_root, config_key):
                    marker_path.unlink(missing_ok=True)
                    candidate = None
                    continue
                _save_cache(runtime_root, key, entry)
                # One runtime has one mounted view. Bound process cache growth
                # when tests or deployments switch canonical roots.
                for stale_key in list(_VIEW_CACHE):
                    if stale_key[1] == str(runtime_root):
                        _VIEW_CACHE.pop(stale_key, None)
                _VIEW_CACHE[key] = entry
                _log_cache("refreshed", started, lock_wait_ms=lock_wait_ms, refresh_ms=(time.perf_counter() - refresh_started) * 1000, entry=entry)
                return view
            raise RuntimeError("Skill source changed repeatedly during view refresh")
        except Exception:
            _VIEW_CACHE.pop(key, None)
            (runtime_root / _CACHE_FILE_NAME).unlink(missing_ok=True)
            _clear_cached_view(view_root)
            _log_cache("failed", started, lock_wait_ms=lock_wait_ms, refresh_ms=(time.perf_counter() - refresh_started) * 1000)
            raise


def _refresh_uncached_coalesced(*, storage: SkillStorage | None) -> SkillSandboxView:
    """One in-process catalogue copy, even when the metadata cache is off.

    Task-analysis seats call ``read_file`` / ``bash`` together. Without this,
    each call takes the process lock and copies every enabled skill. The lock
    holder keeps it for the whole copy, and a later seat fails the tool with
    ``Timed out waiting for skill view process lock``.
    """
    if storage is None:
        from deerflow.skills.storage import get_or_new_skill_storage

        storage = get_or_new_skill_storage()

    from deerflow.config.paths import get_paths

    skills_root = Path(storage.get_skills_root_path()).resolve()
    runtime_root = get_paths().base_dir
    runtime_root.mkdir(parents=True, exist_ok=True)
    try:
        config_key = _config_key()
    except OSError:
        config_key = "extensions-unavailable"
    # Instance id keeps differently filtered storage objects from sharing a
    # result. The process singleton is one object, so parallel tool calls join.
    flight_config = f"uncached:{id(storage)}:{config_key}"
    return _refresh_singleflight(
        _cache_key(skills_root, runtime_root.resolve(), storage),
        flight_config,
        lambda: _refresh_uncached_skill_sandbox_view(storage=storage),
        require_cache=False,
    )


def _local_view_context(storage: SkillStorage | None):
    """Return ``(storage, skills_root, runtime_root, cache_key)`` for a local tree."""
    from deerflow.config.paths import get_paths
    from deerflow.skills.storage.local_skill_storage import LocalSkillStorage

    if storage is None:
        from deerflow.skills.storage import get_or_new_skill_storage

        storage = get_or_new_skill_storage()
    if type(storage) is not LocalSkillStorage:
        return None
    skills_root = Path(storage.get_skills_root_path()).resolve()
    runtime_root = get_paths().base_dir.resolve()
    return storage, skills_root, runtime_root, _cache_key(skills_root, runtime_root, storage)


def _marker_matches(entry: _ViewCacheEntry) -> bool:
    try:
        return (entry.view.root / SKILL_VIEW_MARKER).read_text(encoding="utf-8") == entry.view.marker_content
    except OSError:
        return False


def _peek_entry(storage: SkillStorage | None = None, *, allow_stale_config: bool = False) -> _ViewCacheEntry | None:
    """Return a published view without stating every skill file.

    The default check still reads the extensions document. A hash mismatch
    means availability changed outside this process, and the caller rebuilds.
    ``allow_stale_config`` is only for the management sync that just wrote
    that document and is about to publish the new hash itself.
    """
    if not _skill_view_cache_enabled():
        return None
    context = _local_view_context(storage)
    if context is None:
        return None
    _storage, _skills_root, runtime_root, key = context
    try:
        config_key = _config_key()
    except OSError:
        return None
    entry = _VIEW_CACHE.get(key)
    if entry is None:
        loaded = _load_cache(runtime_root, key)
        if loaded is not None and (allow_stale_config or loaded.config_key == config_key) and _marker_matches(loaded):
            _VIEW_CACHE[key] = loaded
            entry = loaded
    if entry is None or not _marker_matches(entry):
        return None
    if not allow_stale_config and entry.config_key != config_key:
        return None
    return entry


def peek_skill_sandbox_view(*, storage: SkillStorage | None = None) -> SkillSandboxView | None:
    """Return the warm online view, or None when a full rebuild is required."""
    entry = _peek_entry(storage)
    return None if entry is None else entry.view


def current_skill_view_root(*, storage: SkillStorage | None = None) -> Path:
    """Host path of the online view, without a full-tree stat when the cache is warm."""
    from deerflow.sandbox.skill_view_scope import get_active_skill_view_scope

    scope = get_active_skill_view_scope()
    if scope is not None:
        if scope.error is not None:
            raise scope.error
        if scope.view is not None:
            return scope.view.root
    if _skill_view_cache_enabled():
        peeked = peek_skill_sandbox_view(storage=storage)
        if peeked is not None:
            return peeked.root
    return refresh_skill_sandbox_view(storage=storage).root


def _capture_prefix(root: Path, prefix: str, *, source: bool) -> dict[str, _Metadata]:
    """Metadata for one skill directory, using the same rules as a full scan."""
    base = root / prefix
    info = _metadata(base)
    if info is None or not base.is_dir() or _is_view_link(base):
        return {}
    records: dict[str, _Metadata] = {prefix: info}
    visited: set[Path] = set()
    resolved_root = root.resolve()
    for current, directories, files in os.walk(base, followlinks=source):
        current_path = Path(current)
        resolved = current_path.resolve()
        if not resolved.is_relative_to(resolved_root) or resolved in visited:
            directories[:] = []
            continue
        visited.add(resolved)
        if source and resolved != current_path:
            records[resolved.relative_to(resolved_root).as_posix()] = _metadata(resolved)
        for name in [*directories, *files]:
            path = current_path / name
            records[path.relative_to(root).as_posix()] = _metadata(path)
        if not source:
            directories[:] = [name for name in directories if not _is_view_link(current_path / name)]
    return records


def _refresh_ancestor_metadata(records: dict[str, _Metadata], root: Path, prefix: str, *, source: bool) -> None:
    for parent in Path(prefix).parents:
        key = "." if parent == Path(".") else parent.as_posix()
        records[key] = _metadata(root if key == "." else root / parent, root=source and key == ".")


def _parse_skill_dir(skills_root: Path, skill_dir: Path):
    from deerflow.skills.parser import parse_skill_file
    from deerflow.skills.types import SkillCategory

    try:
        relative = skill_dir.resolve().relative_to(skills_root.resolve())
    except ValueError:
        return None
    if len(relative.parts) < 2 or relative.parts[0] not in {"public", "custom"}:
        return None
    category = SkillCategory(relative.parts[0])
    return parse_skill_file(skill_dir / "SKILL.md", category, Path(*relative.parts[1:]))


def _find_skill_by_name(skills_root: Path, name: str):
    """Return the skill whose frontmatter name is ``name``.

    Writes land in ``<category>/<name>/``. Those two directories are checked
    before any catalogue walk, in the same public-then-custom order as the
    walk, and the first frontmatter match is returned. A package whose
    directory name differs from its frontmatter name still falls through to
    the walk.
    """
    for category in ("public", "custom"):
        skill_dir = skills_root / category / name
        if not (skill_dir / "SKILL.md").is_file():
            continue
        skill = _parse_skill_dir(skills_root, skill_dir)
        if skill is not None and skill.name == name:
            return skill
    for category in ("public", "custom"):
        base = skills_root / category
        if not base.is_dir():
            continue
        for dirpath, dirnames, filenames in os.walk(base, followlinks=False):
            current = Path(dirpath)
            dirnames[:] = [item for item in dirnames if not (current / item).is_symlink()]
            if "SKILL.md" not in filenames:
                continue
            skill = _parse_skill_dir(skills_root, current)
            if skill is not None and skill.name == name:
                return skill
    return None


def _source_for_virtual_path(skills_root: Path, container_path: str, virtual_path: str) -> Path | None:
    from pathlib import PurePosixPath

    prefix = container_path.replace("\\", "/").rstrip("/")
    normalized = virtual_path.replace("\\", "/").rstrip("/")
    if ".." in PurePosixPath(normalized).parts:
        return None
    if normalized != prefix and not normalized.startswith(prefix + "/"):
        return None
    remainder = normalized[len(prefix) :].lstrip("/")
    parts = PurePosixPath(remainder).parts
    if not parts or parts[0] not in {"public", "custom"} or ".." in parts:
        return None
    for end in range(len(parts), 1, -1):
        candidate = skills_root.joinpath(*parts[:end])
        if (candidate / "SKILL.md").is_file() and not candidate.is_symlink():
            return candidate
    if len(parts) >= 2:
        candidate = skills_root / parts[0] / parts[1]
        if candidate.is_dir() and not candidate.is_symlink() and not (candidate / "SKILL.md").is_file():
            return candidate
    return None


def _remove_view_tree(view_root: Path, relative: Path) -> None:
    target = view_root / relative
    if not target.exists() and not target.is_symlink():
        return
    if _is_view_link(target):
        _unlink_view_link(target)
    elif target.is_dir():
        shutil.rmtree(target)
    else:
        target.unlink(missing_ok=True)


def _remove_stale_under(view_root: Path, view_base: Path, desired_files: dict[Path, Path], desired_dirs: set[Path]) -> None:
    if not view_base.exists():
        return
    entries: list[Path] = []
    for current, directories, files in os.walk(view_base, followlinks=False):
        current_path = Path(current)
        entries.extend(current_path / name for name in [*directories, *files])
        directories[:] = [name for name in directories if not _is_view_link(current_path / name)]
    for entry in sorted(entries, key=lambda path: len(path.parts), reverse=True):
        relative = entry.relative_to(view_root)
        if _is_view_link(entry):
            _unlink_view_link(entry)
        elif entry.is_dir():
            if relative not in desired_dirs:
                entry.rmdir()
        elif relative not in desired_files:
            entry.unlink(missing_ok=True)


def _reconcile_tree(skills_root: Path, view_root: Path, source_root: Path, entry: _ViewCacheEntry) -> str | None:
    """Copy one source directory into the view. Return its prefix, or None if it is gone."""
    prefix = source_root.resolve().relative_to(skills_root.resolve()).as_posix()
    if not source_root.is_dir():
        _remove_view_tree(view_root, Path(prefix))
        return None
    desired_files: dict[Path, Path] = {}
    desired_dirs: set[Path] = set()
    _add_tree(source_root, skills_root, desired_files, desired_dirs)
    view_base = view_root / prefix
    if _is_view_link(view_base):
        _unlink_view_link(view_base)
    elif view_base.exists():
        _remove_stale_under(view_root, view_base, desired_files, desired_dirs)
    cached_source = dict(entry.source)
    cached_target = dict(entry.target)
    for relative_dir in sorted(desired_dirs, key=lambda path: len(path.parts)):
        destination = view_root / relative_dir
        if _is_view_link(destination):
            _unlink_view_link(destination)
        elif destination.exists() and not destination.is_dir():
            destination.unlink()
        destination.mkdir(parents=True, exist_ok=True)
    for relative, source in desired_files.items():
        key = relative.as_posix()
        destination = view_root / relative
        source_same = cached_source.get(key) == _metadata(source)
        destination_metadata = _metadata(destination)
        # A missing file and an uncached file are both represented by ``None``.
        # After an offline skill has been removed from the published view, the
        # source metadata can still be current while the target key is absent.
        # Do not mistake that cache miss for an unchanged destination or the
        # skill will be marked online without restoring any of its files.
        target_same = destination_metadata is not None and cached_target.get(key) == destination_metadata
        if not (source_same and target_same):
            _copy_file_atomic(source, destination)
    return prefix


def _apply_prefix(source_map: dict[str, _Metadata], target_map: dict[str, _Metadata], skills_root: Path, view_root: Path, prefix: str) -> None:
    for records in (source_map, target_map):
        stale = [key for key in records if key == prefix or key.startswith(prefix + "/")]
        for key in stale:
            records.pop(key, None)
    source_map.update(_capture_prefix(skills_root, prefix, source=True))
    target_map.update(_capture_prefix(view_root, prefix, source=False))
    _refresh_ancestor_metadata(source_map, skills_root, prefix, source=True)
    _refresh_ancestor_metadata(target_map, view_root, prefix, source=False)


def _publish_entry(
    runtime_root: Path,
    key: tuple[str, str, str],
    entry: _ViewCacheEntry,
    *,
    names: frozenset[str],
    roots: tuple[tuple[str, str, str], ...],
    source_map: dict[str, _Metadata],
    target_map: dict[str, _Metadata],
    config_key: str,
) -> SkillSandboxView:
    view = SkillSandboxView(entry.view.root, entry.view.token, names, roots, entry.view.generation)
    source = tuple(sorted(source_map.items()))
    target = tuple(sorted(target_map.items()))
    if view.enabled_names == entry.view.enabled_names and view.enabled_roots == entry.view.enabled_roots and source == entry.source and target == entry.target and config_key == entry.config_key:
        return entry.view
    published = _ViewCacheEntry(view, config_key, source, target)
    _VIEW_CACHE[key] = published
    _save_cache(runtime_root, key, published)
    return view


def _skill_enabled(name: str, category: str) -> bool:
    from deerflow.config.extensions_config import ExtensionsConfig

    return ExtensionsConfig.from_file().is_skill_enabled(name, category)


def _skill_prefix(skill) -> str:
    relative = skill.relative_path.as_posix()
    if not relative or relative == ".":
        return str(skill.category)
    return f"{skill.category}/{relative}"


def _ensure_support(skills_root: Path, view_root: Path, entry: _ViewCacheEntry, names: frozenset[str], source_map: dict[str, _Metadata], target_map: dict[str, _Metadata]) -> None:
    for support_root in _support_roots(skills_root, names):
        prefix = _reconcile_tree(skills_root, view_root, support_root, entry)
        if prefix is not None:
            _apply_prefix(source_map, target_map, skills_root, view_root, prefix)


def _is_support_candidate(skills_root: Path, category: str, name: str) -> bool:
    if (category, name) in _BUILTIN_SUPPORT_DEPENDENCIES:
        return True
    return (skills_root / category / name / SKILL_SUPPORT_MANIFEST).is_file()


def _prune_support(
    skills_root: Path,
    view_root: Path,
    names: frozenset[str],
    source_map: dict[str, _Metadata],
    target_map: dict[str, _Metadata],
) -> None:
    needed = {source.resolve().relative_to(skills_root.resolve()).as_posix() for source in _support_roots(skills_root, names)}
    for category in ("public", "custom"):
        view_category = view_root / category
        if not view_category.is_dir():
            continue
        for child in list(view_category.iterdir()):
            prefix = f"{category}/{child.name}"
            if prefix in needed or not _is_support_candidate(skills_root, category, child.name):
                continue
            _remove_view_tree(view_root, Path(prefix))
            _apply_prefix(source_map, target_map, skills_root, view_root, prefix)


def _mutate_skill(
    skills_root: Path,
    view_root: Path,
    entry: _ViewCacheEntry,
    skill,
    *,
    enabled: bool,
    source_map: dict[str, _Metadata],
    target_map: dict[str, _Metadata],
    names: set[str],
    roots: list[tuple[str, str, str]],
) -> None:
    prefix_path = Path(_skill_prefix(skill))
    if not enabled:
        _remove_view_tree(view_root, prefix_path)
        _apply_prefix(source_map, target_map, skills_root, view_root, prefix_path.as_posix())
        names.discard(skill.name)
        roots[:] = [item for item in roots if item[0] != skill.name]
        _prune_support(skills_root, view_root, frozenset(names), source_map, target_map)
        return
    prefix = _reconcile_tree(skills_root, view_root, Path(skill.skill_dir), entry)
    if prefix is None:
        names.discard(skill.name)
        roots[:] = [item for item in roots if item[0] != skill.name]
        return
    _apply_prefix(source_map, target_map, skills_root, view_root, prefix)
    names.add(skill.name)
    root = (skill.name, str(skill.category), skill.relative_path.as_posix())
    roots[:] = [item for item in roots if item[0] != skill.name]
    roots.append(root)
    _ensure_support(skills_root, view_root, entry, frozenset({skill.name}), source_map, target_map)


def _sync_entry_skill(entry: _ViewCacheEntry, skills_root: Path, runtime_root: Path, key: tuple[str, str, str], name: str) -> SkillSandboxView:
    skill = _find_skill_by_name(skills_root, name)
    config_key = _config_key()
    with _view_lock(runtime_root):
        source_map = dict(entry.source)
        target_map = dict(entry.target)
        names = set(entry.view.enabled_names)
        roots = list(entry.view.enabled_roots)
        if skill is None:
            removed = [item for item in roots if item[0] == name]
            for _skill_name, category, relative in removed:
                prefix = category if not relative or relative == "." else f"{category}/{relative}"
                _remove_view_tree(entry.view.root, Path(prefix))
                _apply_prefix(source_map, target_map, skills_root, entry.view.root, prefix)
            names.discard(name)
            roots = [item for item in roots if item[0] != name]
            _prune_support(skills_root, entry.view.root, frozenset(names), source_map, target_map)
        else:
            _mutate_skill(
                skills_root,
                entry.view.root,
                entry,
                skill,
                enabled=_skill_enabled(skill.name, str(skill.category)),
                source_map=source_map,
                target_map=target_map,
                names=names,
                roots=roots,
            )
        return _publish_entry(
            runtime_root,
            key,
            entry,
            names=frozenset(names),
            roots=tuple(roots),
            source_map=source_map,
            target_map=target_map,
            config_key=config_key,
        )


def sync_skill_sandbox_skill(name: str, *, storage: SkillStorage | None = None) -> SkillSandboxView | None:
    """Copy or remove one skill in the online view after a management write.

    A cold view is built in full so the first publication still contains every
    online skill. Later writes touch only the named skill and its support
    library. Failures propagate; callers that must not fail a user save should
    catch them. The next explicit skill command reconciles that skill anyway.
    """
    if not name or not _skill_view_cache_enabled():
        return None
    context = _local_view_context(storage)
    if context is None:
        return refresh_skill_sandbox_view(storage=storage)
    _storage, skills_root, runtime_root, key = context
    entry = _peek_entry(storage, allow_stale_config=True)
    if entry is None:
        return refresh_skill_sandbox_view(storage=storage)
    return _sync_entry_skill(entry, skills_root, runtime_root, key, name)


def _skill_dirs_for_relative_paths(skills_root: Path, relative_paths: list[str]) -> list[Path]:
    found: list[Path] = []
    seen: set[Path] = set()
    for relative in relative_paths:
        path = skills_root / relative
        if ".." in Path(relative).parts:
            continue
        current = path if path.is_dir() else path.parent
        skill_dir: Path | None = None
        try:
            current.resolve().relative_to(skills_root.resolve())
        except ValueError:
            continue
        while True:
            try:
                current.resolve().relative_to(skills_root.resolve())
            except ValueError:
                break
            if (current / "SKILL.md").is_file():
                skill_dir = current
                break
            if current == skills_root:
                break
            current = current.parent
        if skill_dir is None and len(Path(relative).parts) >= 2:
            candidate = skills_root / Path(relative).parts[0] / Path(relative).parts[1]
            if candidate.is_dir() and not (candidate / "SKILL.md").is_file():
                skill_dir = candidate
        if skill_dir is not None:
            resolved = skill_dir.resolve()
            if resolved not in seen:
                seen.add(resolved)
                found.append(skill_dir)
    return found


def sync_skill_sandbox_relative_paths(relative_paths: list[str], *, storage: SkillStorage | None = None) -> None:
    """Refresh the skills that contain files just rewritten by address management."""
    if not relative_paths or not _skill_view_cache_enabled():
        return
    context = _local_view_context(storage)
    if context is None:
        refresh_skill_sandbox_view(storage=storage)
        return
    _storage, skills_root, runtime_root, key = context
    entry = _peek_entry(storage, allow_stale_config=True)
    if entry is None:
        refresh_skill_sandbox_view(storage=storage)
        return
    for skill_dir in _skill_dirs_for_relative_paths(skills_root, relative_paths):
        skill = _parse_skill_dir(skills_root, skill_dir)
        if skill is not None:
            entry = _peek_entry(storage, allow_stale_config=True) or entry
            _sync_entry_skill(entry, skills_root, runtime_root, key, skill.name)
            continue
        prefix = skill_dir.resolve().relative_to(skills_root.resolve()).as_posix()
        if prefix not in {item.resolve().relative_to(skills_root.resolve()).as_posix() for item in _support_roots(skills_root, entry.view.enabled_names)}:
            continue
        with _view_lock(runtime_root):
            current = _VIEW_CACHE.get(key) or entry
            source_map = dict(current.source)
            target_map = dict(current.target)
            copied = _reconcile_tree(skills_root, current.view.root, skill_dir, current)
            if copied is not None:
                _apply_prefix(source_map, target_map, skills_root, current.view.root, copied)
            _publish_entry(
                runtime_root,
                key,
                current,
                names=current.view.enabled_names,
                roots=current.view.enabled_roots,
                source_map=source_map,
                target_map=target_map,
                config_key=_config_key(),
            )


def _reconcile_virtual_paths(
    entry: _ViewCacheEntry,
    skills_root: Path,
    runtime_root: Path,
    key: tuple[str, str, str],
    virtual_paths: list[str],
    container_path: str,
) -> SkillSandboxView:
    config_key = _config_key()
    with _view_lock(runtime_root):
        current = _VIEW_CACHE.get(key) or entry
        source_map = dict(current.source)
        target_map = dict(current.target)
        names = set(current.view.enabled_names)
        roots = list(current.view.enabled_roots)
        for virtual_path in virtual_paths:
            source = _source_for_virtual_path(skills_root, container_path, virtual_path)
            if source is None:
                continue
            skill = _parse_skill_dir(skills_root, source)
            if skill is not None:
                remember_invoked_skill(skill.name)
            if skill is None:
                allowed = {item.resolve() for item in _support_roots(skills_root, frozenset(names))}
                if source.resolve() in allowed:
                    prefix = _reconcile_tree(skills_root, current.view.root, source, current)
                    if prefix is not None:
                        _apply_prefix(source_map, target_map, skills_root, current.view.root, prefix)
                continue
            _mutate_skill(
                skills_root,
                current.view.root,
                current,
                skill,
                enabled=_skill_enabled(skill.name, str(skill.category)),
                source_map=source_map,
                target_map=target_map,
                names=names,
                roots=roots,
            )
        return _publish_entry(
            runtime_root,
            key,
            current,
            names=frozenset(names),
            roots=tuple(roots),
            source_map=source_map,
            target_map=target_map,
            config_key=config_key,
        )


def prepare_skill_tool_view(
    virtual_paths: list[str],
    *,
    container_path: str = "/mnt/skills",
    storage: SkillStorage | None = None,
) -> SkillSandboxView | None:
    """Make the online view ready for one sandbox tool call.

    An empty path list means the command does not name a skill: leave the
    published view untouched. Named skills are reconciled individually. A
    missing or availability-mismatched view falls back to a full rebuild so a
    cold start still exposes every online skill.
    """
    from deerflow.sandbox.skill_view_scope import get_active_skill_view_scope

    if not _skill_view_cache_enabled():
        return refresh_skill_sandbox_view(storage=storage)
    scope = get_active_skill_view_scope()
    if scope is not None and (scope.view is not None or scope.error is not None):
        return scope.view
    if not virtual_paths:
        started = time.perf_counter()
        viewed = peek_skill_sandbox_view(storage=storage)
        note_skill_view_check(
            status="skipped",
            total_ms=(time.perf_counter() - started) * 1000,
            enabled_count=len(viewed.enabled_names) if viewed is not None else 0,
        )
        logger.debug("SKILL_VIEW_CACHE status=skipped")
        return viewed
    context = _local_view_context(storage)
    entry = None if context is None else _peek_entry(storage)
    if context is None or entry is None:
        return refresh_skill_sandbox_view(storage=storage)
    _storage, skills_root, runtime_root, key = context
    started = time.perf_counter()
    try:
        view = _reconcile_virtual_paths(entry, skills_root, runtime_root, key, virtual_paths, container_path)
    except Exception as exc:
        note_skill_view_check(status="failed", total_ms=(time.perf_counter() - started) * 1000, path_count=len(virtual_paths))
        if scope is not None:
            scope.error = exc
        raise
    if scope is not None:
        scope.view = view
    note_skill_view_check(
        status="narrow",
        total_ms=(time.perf_counter() - started) * 1000,
        enabled_count=len(view.enabled_names),
        path_count=len(virtual_paths),
    )
    logger.debug("SKILL_VIEW_CACHE status=narrow paths=%d enabled_count=%d", len(virtual_paths), len(view.enabled_names))
    return view


def refresh_skill_sandbox_view(*, storage: SkillStorage | None = None, force: bool = False) -> SkillSandboxView:
    """Return a verified online-only view; caching/reuse are opt-in switches.

    Cached mode revalidates availability bytes and all known source/target
    metadata on every call. It never delays an offline change until a TTL.
    """
    from deerflow.sandbox.skill_view_scope import get_active_skill_view_scope

    scope = get_active_skill_view_scope() if storage is None and not force else None
    if scope is not None:
        if scope.error is not None:
            raise scope.error
        if scope.view is not None:
            return scope.view
    started = time.perf_counter()
    try:
        if _skill_view_cache_enabled():
            view = _refresh_cached_skill_sandbox_view(storage=storage, force=force)
        else:
            view = _refresh_uncached_coalesced(storage=storage)
            try:
                note_skill_view_check(
                    status="refreshed",
                    total_ms=(time.perf_counter() - started) * 1000,
                    enabled_count=len(getattr(view, "enabled_names", ()) or ()),
                )
            except Exception:
                logger.debug("skill view diagnostic was not saved", exc_info=True)
        if scope is not None:
            scope.view = view
        return view
    except Exception as exc:
        try:
            note_skill_view_check(status="failed", total_ms=(time.perf_counter() - started) * 1000)
        except Exception:
            logger.debug("skill view diagnostic was not saved", exc_info=True)
        if scope is not None:
            scope.error = exc
        raise


def expected_skill_view_marker(*, storage: SkillStorage | None = None) -> tuple[Path, str]:
    """Refresh the view and return its marker path plus exact expected text."""
    view = refresh_skill_sandbox_view(storage=storage)
    return view.root / SKILL_VIEW_MARKER, view.marker_content


__all__ = [
    "SKILL_SUPPORT_MANIFEST",
    "SKILL_VIEW_DIR_NAME",
    "SKILL_VIEW_MARKER",
    "SKILL_VIEW_PROTOCOL",
    "SkillSandboxView",
    "current_skill_view_root",
    "expected_skill_view_marker",
    "peek_skill_sandbox_view",
    "prepare_skill_tool_view",
    "refresh_skill_sandbox_view",
    "reset_skill_sandbox_view_cache",
    "sync_skill_sandbox_relative_paths",
    "sync_skill_sandbox_skill",
]