441 lines
16 KiB
Python
441 lines
16 KiB
Python
"""MySQL-backed projection for the Gateway skill list endpoint."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from sqlalchemy import or_, select, update
|
|
from sqlalchemy.exc import IntegrityError
|
|
from sqlalchemy.schema import CreateColumn
|
|
|
|
from deerflow.config.app_config import AppConfig
|
|
from deerflow.persistence.engine import get_engine, get_session_factory
|
|
from deerflow.persistence.skills.model import SkillCatalogSnapshotRow, SkillRow
|
|
from deerflow.skills.storage import get_or_new_skill_storage
|
|
|
|
logger = logging.getLogger(__name__)
|
|
_CATALOG_CACHE_KEY = "global"
|
|
_catalog_warm_lock = asyncio.Lock()
|
|
_memory_catalog_snapshot: list[dict[str, Any]] | None = None
|
|
_memory_catalog_is_fallback = False
|
|
|
|
|
|
_SKILL_LIST_COLUMNS = (
|
|
SkillRow.name,
|
|
SkillRow.category,
|
|
SkillRow.description,
|
|
SkillRow.license,
|
|
SkillRow.enabled,
|
|
SkillRow.always_on,
|
|
SkillRow.use_count,
|
|
SkillRow.view_count,
|
|
SkillRow.patch_count,
|
|
SkillRow.last_activity_at,
|
|
SkillRow.created_at,
|
|
SkillRow.state,
|
|
SkillRow.pinned,
|
|
SkillRow.featured,
|
|
SkillRow.source,
|
|
SkillRow.owner_user_id,
|
|
SkillRow.published,
|
|
SkillRow.name_zh,
|
|
SkillRow.square_id,
|
|
SkillRow.detail_preview,
|
|
)
|
|
|
|
_SKILL_CATALOG_COLUMNS = {
|
|
"category",
|
|
"description",
|
|
"license",
|
|
"enabled",
|
|
"available",
|
|
"use_count",
|
|
"view_count",
|
|
"patch_count",
|
|
"last_activity_at",
|
|
"state",
|
|
"pinned",
|
|
"featured",
|
|
"source",
|
|
"source_mtime_ns",
|
|
"detail_preview",
|
|
}
|
|
|
|
|
|
async def ensure_skill_catalog_schema() -> None:
|
|
"""Add compact catalog columns to an existing MySQL/SQLite skills table.
|
|
|
|
``create_all`` creates new tables but intentionally does not alter existing
|
|
ones. These additive columns all have safe defaults (or are nullable), so
|
|
this migration preserves existing ownership/detail data and is repeatable.
|
|
"""
|
|
|
|
engine = get_engine()
|
|
if engine is None:
|
|
return
|
|
|
|
def _ensure(sync_conn) -> None:
|
|
from sqlalchemy import func, inspect, update
|
|
|
|
table = SkillRow.__table__
|
|
inspector = inspect(sync_conn)
|
|
if table.name not in inspector.get_table_names():
|
|
table.create(sync_conn, checkfirst=True)
|
|
return
|
|
|
|
existing = {column["name"] for column in inspector.get_columns(table.name)}
|
|
preparer = sync_conn.dialect.identifier_preparer
|
|
quoted_table = preparer.quote(table.name)
|
|
for name in sorted(_SKILL_CATALOG_COLUMNS - existing):
|
|
column = table.columns[name]
|
|
definition = str(CreateColumn(column).compile(dialect=sync_conn.dialect))
|
|
sync_conn.exec_driver_sql(f"ALTER TABLE {quoted_table} ADD COLUMN {definition}")
|
|
existing.add(name)
|
|
|
|
# Preserve the short description previously exposed from ``detail``
|
|
# without selecting that potentially large field in normal list reads.
|
|
sync_conn.execute(
|
|
update(table)
|
|
.where(table.c.detail_preview.is_(None), table.c.detail.is_not(None))
|
|
.values(detail_preview=func.substr(table.c.detail, 1, 1024))
|
|
)
|
|
|
|
existing_indexes = {index["name"] for index in inspector.get_indexes(table.name)}
|
|
for index in table.indexes:
|
|
if index.name not in existing_indexes:
|
|
index.create(sync_conn, checkfirst=True)
|
|
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(_ensure)
|
|
|
|
|
|
async def load_skill_catalog_projection(
|
|
*,
|
|
user_id: str,
|
|
admin: bool = False,
|
|
) -> list[dict[str, Any]] | None:
|
|
"""Read lightweight skill catalog columns from MySQL, never SKILL.md or LONGTEXT.
|
|
|
|
``None`` means persistence is unavailable; an empty list is a valid empty
|
|
catalog. Access filtering happens in SQL so private skill metadata never
|
|
crosses into the request process for an unrelated user.
|
|
"""
|
|
|
|
session_factory = get_session_factory()
|
|
if session_factory is None:
|
|
return None
|
|
stmt = select(*_SKILL_LIST_COLUMNS).where(SkillRow.available.is_(True))
|
|
if not admin:
|
|
stmt = stmt.where(
|
|
or_(
|
|
SkillRow.owner_user_id.is_(None),
|
|
SkillRow.owner_user_id == user_id,
|
|
SkillRow.published.is_(True),
|
|
)
|
|
)
|
|
stmt = stmt.order_by(SkillRow.created_at.asc())
|
|
try:
|
|
async with session_factory() as session:
|
|
rows = (await session.execute(stmt)).mappings().all()
|
|
except Exception:
|
|
logger.warning("Skill MySQL catalog projection unavailable; retaining legacy fallback", exc_info=True)
|
|
return None
|
|
result: list[dict[str, Any]] = []
|
|
for row in rows:
|
|
item = dict(row)
|
|
item["builtin"] = item.get("owner_user_id") is None
|
|
result.append(item)
|
|
return result
|
|
|
|
|
|
def _parse_catalog_datetime(value: Any) -> datetime | None:
|
|
if isinstance(value, datetime):
|
|
return value
|
|
if not isinstance(value, str) or not value.strip():
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
async def refresh_skill_catalog_projection(
|
|
app_config: AppConfig | None = None,
|
|
*,
|
|
skills: list[Any] | None = None,
|
|
) -> int:
|
|
"""Synchronize filesystem metadata into one compact MySQL row per skill.
|
|
|
|
This runs at startup and after explicit skill changes. Request handlers
|
|
only read the projection and never walk the skill tree or usage sidecars.
|
|
Ownership, publication, and long-form detail columns are preserved.
|
|
"""
|
|
|
|
session_factory = get_session_factory()
|
|
if session_factory is None:
|
|
return 0
|
|
await ensure_skill_catalog_schema()
|
|
storage = get_or_new_skill_storage(app_config=app_config)
|
|
if skills is None:
|
|
skills = await asyncio.to_thread(storage.load_skills, enabled_only=False)
|
|
usage_entries = await asyncio.to_thread(_load_usage_entries)
|
|
catalog = await asyncio.to_thread(_build_catalog_payload, skills, usage_entries)
|
|
metadata_by_name: dict[str, tuple[str, int]] = {}
|
|
for skill in skills:
|
|
try:
|
|
stat = skill.skill_file.stat()
|
|
metadata_by_name[skill.name] = (
|
|
datetime.fromtimestamp(stat.st_mtime, tz=UTC).isoformat(),
|
|
int(stat.st_mtime_ns),
|
|
)
|
|
except OSError:
|
|
metadata_by_name[skill.name] = ("", 0)
|
|
|
|
async with session_factory() as session:
|
|
existing_rows = (
|
|
await session.execute(select(SkillRow.name, SkillRow.owner_user_id, SkillRow.always_on))
|
|
).all()
|
|
existing = {str(name): (owner_id, bool(always_on)) for name, owner_id, always_on in existing_rows}
|
|
catalog_names = {str(item.get("name") or "").strip() for item in catalog}
|
|
removed_names = sorted(name for name in existing if name not in catalog_names)
|
|
if removed_names:
|
|
await session.execute(
|
|
update(SkillRow)
|
|
.where(SkillRow.name.in_(removed_names))
|
|
.values(available=False, updated_at=datetime.now(UTC))
|
|
)
|
|
for item in catalog:
|
|
name = str(item.get("name") or "").strip()
|
|
if not name:
|
|
continue
|
|
category = str(item.get("category") or "custom")[:16]
|
|
created_at_text, source_mtime_ns = metadata_by_name.get(name, ("", 0))
|
|
created_at = _parse_catalog_datetime(item.get("created_at") or created_at_text) or datetime.now(UTC)
|
|
fields: dict[str, Any] = {
|
|
"category": category,
|
|
"description": str(item.get("description") or "")[:1024],
|
|
"license": str(item.get("license") or "")[:255] or None,
|
|
"enabled": bool(item.get("enabled", True)),
|
|
"available": True,
|
|
"use_count": max(0, int(item.get("use_count") or 0)),
|
|
"view_count": max(0, int(item.get("view_count") or 0)),
|
|
"patch_count": max(0, int(item.get("patch_count") or 0)),
|
|
"last_activity_at": _parse_catalog_datetime(item.get("last_activity_at")),
|
|
"state": str(item.get("state") or "active")[:16],
|
|
"pinned": bool(item.get("pinned")),
|
|
"featured": bool(item.get("featured")),
|
|
"source": str(item.get("source") or "upload")[:16],
|
|
"source_mtime_ns": source_mtime_ns,
|
|
"updated_at": datetime.now(UTC),
|
|
}
|
|
if name in existing:
|
|
# The per-skill always_on preference is admin-editable in MySQL
|
|
# and must survive filesystem metadata refreshes.
|
|
await session.execute(update(SkillRow).where(SkillRow.name == name).values(**fields))
|
|
else:
|
|
owner_id = None
|
|
session.add(
|
|
SkillRow(
|
|
name=name,
|
|
owner_user_id=owner_id,
|
|
published=category != "custom",
|
|
always_on=bool(item.get("always_on", False)),
|
|
created_at=created_at,
|
|
**fields,
|
|
)
|
|
)
|
|
await session.commit()
|
|
return len(catalog)
|
|
|
|
|
|
def _build_catalog_payload(skills: list[Any], usage_entries: dict[str, Any]) -> list[dict[str, Any]]:
|
|
payload: list[dict[str, Any]] = []
|
|
defaults = {
|
|
"use_count": 0,
|
|
"view_count": 0,
|
|
"patch_count": 0,
|
|
"last_used_at": None,
|
|
"last_viewed_at": None,
|
|
"last_patched_at": None,
|
|
"created_at": None,
|
|
"state": "active",
|
|
"pinned": False,
|
|
"featured": False,
|
|
"source": "upload",
|
|
}
|
|
for skill in skills:
|
|
usage = {**defaults, **(usage_entries.get(skill.name) or {})}
|
|
created_at = usage.get("created_at")
|
|
if not created_at:
|
|
try:
|
|
created_at = datetime.fromtimestamp(skill.skill_file.stat().st_mtime, tz=UTC).isoformat()
|
|
except OSError:
|
|
created_at = None
|
|
activity = [
|
|
value
|
|
for value in (
|
|
usage.get("last_used_at"),
|
|
usage.get("last_viewed_at"),
|
|
usage.get("last_patched_at"),
|
|
)
|
|
if value
|
|
]
|
|
last_activity_at = max(activity) if activity else created_at
|
|
payload.append(
|
|
{
|
|
"name": skill.name,
|
|
"description": skill.description or "",
|
|
"license": skill.license,
|
|
"category": str(getattr(skill.category, "value", skill.category)),
|
|
"enabled": bool(skill.enabled),
|
|
"always_on": bool(getattr(skill, "always_on", False)),
|
|
"use_count": int(usage.get("use_count") or 0),
|
|
"view_count": int(usage.get("view_count") or 0),
|
|
"patch_count": int(usage.get("patch_count") or 0),
|
|
"last_activity_at": last_activity_at,
|
|
"created_at": created_at,
|
|
"state": str(usage.get("state") or "active"),
|
|
"pinned": bool(usage.get("pinned")),
|
|
"featured": bool(usage.get("featured")),
|
|
"source": str(usage.get("source") or "upload"),
|
|
}
|
|
)
|
|
return payload
|
|
|
|
|
|
async def refresh_skill_catalog_snapshot(
|
|
app_config: AppConfig | None = None,
|
|
*,
|
|
skills: list[Any] | None = None,
|
|
) -> int:
|
|
"""Read skill files off-request and persist a single JSON snapshot in MySQL."""
|
|
|
|
session_factory = get_session_factory()
|
|
if session_factory is None:
|
|
return 0
|
|
|
|
storage = get_or_new_skill_storage(app_config=app_config)
|
|
if skills is None:
|
|
skills = await asyncio.to_thread(storage.load_skills, enabled_only=False)
|
|
usage_entries = await asyncio.to_thread(_load_usage_entries)
|
|
catalog = await asyncio.to_thread(_build_catalog_payload, skills, usage_entries)
|
|
encoded = json.dumps(catalog, ensure_ascii=False, separators=(",", ":"))
|
|
|
|
await _persist_catalog_snapshot(encoded)
|
|
global _memory_catalog_snapshot, _memory_catalog_is_fallback
|
|
_memory_catalog_snapshot = catalog
|
|
_memory_catalog_is_fallback = False
|
|
return len(catalog)
|
|
|
|
|
|
async def _persist_catalog_snapshot(encoded: str) -> None:
|
|
session_factory = get_session_factory()
|
|
if session_factory is None:
|
|
return
|
|
row = SkillCatalogSnapshotRow(
|
|
cache_key=_CATALOG_CACHE_KEY,
|
|
catalog_json=encoded,
|
|
updated_at=datetime.now(UTC),
|
|
)
|
|
for attempt in range(2):
|
|
try:
|
|
async with session_factory() as session:
|
|
await session.merge(row)
|
|
await session.commit()
|
|
return
|
|
except IntegrityError:
|
|
if attempt:
|
|
raise
|
|
# Two workers can warm an empty table at the same time. The loser
|
|
# retries as an update after the winner's transaction is visible.
|
|
|
|
|
|
def _load_usage_entries() -> dict[str, Any]:
|
|
from deerflow.skills.usage import all_entries
|
|
|
|
return all_entries()
|
|
|
|
|
|
async def load_skill_catalog_snapshot() -> list[dict[str, Any]] | None:
|
|
"""Load the prepared catalogue using MySQL only; None means not warmed."""
|
|
|
|
session_factory = get_session_factory()
|
|
if session_factory is None:
|
|
return None
|
|
async with session_factory() as session:
|
|
row = await session.get(SkillCatalogSnapshotRow, _CATALOG_CACHE_KEY)
|
|
if row is None:
|
|
return None
|
|
try:
|
|
payload = json.loads(row.catalog_json)
|
|
except (TypeError, ValueError):
|
|
logger.error("Invalid skill catalogue snapshot JSON; refresh is required")
|
|
return None
|
|
if not isinstance(payload, list):
|
|
logger.error("Invalid skill catalogue snapshot payload type; refresh is required")
|
|
return None
|
|
catalog = [item for item in payload if isinstance(item, dict) and item.get("name")]
|
|
global _memory_catalog_snapshot, _memory_catalog_is_fallback
|
|
_memory_catalog_snapshot = catalog
|
|
_memory_catalog_is_fallback = False
|
|
return catalog
|
|
|
|
|
|
async def load_or_warm_skill_catalog_snapshot(
|
|
app_config: AppConfig | None = None,
|
|
) -> list[dict[str, Any]] | None:
|
|
"""Load MySQL snapshot, warming it once on a cold/missing snapshot.
|
|
|
|
Startup warm-up is best-effort. If it missed because of transient startup
|
|
timing, the first list request builds the catalogue off the event loop and
|
|
stores it before returning. If persistence is temporarily unavailable, a
|
|
process-local copy keeps the endpoint usable without rescanning on every
|
|
request.
|
|
"""
|
|
|
|
global _memory_catalog_snapshot, _memory_catalog_is_fallback
|
|
if get_session_factory() is None:
|
|
return None
|
|
if _memory_catalog_is_fallback and _memory_catalog_snapshot is not None:
|
|
return _memory_catalog_snapshot
|
|
|
|
try:
|
|
catalog = await load_skill_catalog_snapshot()
|
|
except Exception:
|
|
logger.warning("Skill catalogue snapshot read failed; attempting a one-time warm-up", exc_info=True)
|
|
catalog = None
|
|
if catalog is not None:
|
|
return catalog
|
|
|
|
async with _catalog_warm_lock:
|
|
if _memory_catalog_is_fallback and _memory_catalog_snapshot is not None:
|
|
return _memory_catalog_snapshot
|
|
try:
|
|
catalog = await load_skill_catalog_snapshot()
|
|
except Exception:
|
|
logger.warning("Skill catalogue snapshot is unavailable during warm-up", exc_info=True)
|
|
catalog = None
|
|
if catalog is not None:
|
|
return catalog
|
|
|
|
storage = get_or_new_skill_storage(app_config=app_config)
|
|
skills = await asyncio.to_thread(storage.load_skills, enabled_only=False)
|
|
usage_entries = await asyncio.to_thread(_load_usage_entries)
|
|
catalog = await asyncio.to_thread(_build_catalog_payload, skills, usage_entries)
|
|
encoded = json.dumps(catalog, ensure_ascii=False, separators=(",", ":"))
|
|
try:
|
|
await _persist_catalog_snapshot(encoded)
|
|
except Exception:
|
|
logger.exception("Skill catalogue snapshot persistence failed; serving the warmed process-local copy")
|
|
_memory_catalog_snapshot = catalog
|
|
_memory_catalog_is_fallback = True
|
|
return catalog
|
|
_memory_catalog_snapshot = catalog
|
|
_memory_catalog_is_fallback = False
|
|
logger.info("Skill catalogue snapshot warmed on first request: skills=%d", len(catalog))
|
|
return catalog
|