2184 lines
88 KiB
Python
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",
|
|
]
|