Files

2184 lines
88 KiB
Python

"""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",
]