From 8162719e3622f1fc168ae7d40f9dc186088cd836 Mon Sep 17 00:00:00 2001 From: xuhd Date: Thu, 14 May 2026 15:44:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(skills/notifications):=20=E5=A4=9A?= =?UTF-8?q?=E7=A7=9F=E6=88=B7=E9=98=B6=E6=AE=B5=201=20=E2=80=94=20?= =?UTF-8?q?=E6=8C=81=E4=B9=85=E5=8C=96=E5=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增两套 persistence 模块，仿 persistence/agents/ 模板： 1. persistence/skills/ - SkillRow ORM (PK=name, owner_user_id, published, timestamps) - SkillStore 抽象：list_visible / get_visible / get_any / get_owned / create / ensure_legacy / update / update_admin / delete / delete_admin - SqlAlchemy 实现 SkillRepository - MemorySkillStore 用于测试 - 可见性 = owner OR published OR builtin 2. persistence/notifications/ - NotificationRow ORM (id, user_id, type, title, body, payload JSON, read, created_at) - NotificationStore 抽象：create / list_for_user / unread_count / mark_read / mark_all_read / delete - SqlAlchemy 实现 NotificationRepository - MemoryNotificationStore 用于测试 - sender.notify(user_id, type, ...) 全局接口；harness 不能 import app.*，所以用 set_notification_store / get_notification_store 注入模式 3. persistence/models/__init__.py 注册 SkillRow + NotificationRow 后续阶段会接入路由、cascade、curator per-user 重构、前端。 Co-Authored-By: Claude Opus 4.7 (1M context) --- .../deerflow/persistence/models/__init__.py | 4 + .../persistence/notifications/__init__.py | 19 +++ .../persistence/notifications/base.py | 42 ++++++ .../persistence/notifications/memory.py | 70 +++++++++ .../persistence/notifications/model.py | 24 +++ .../persistence/notifications/sender.py | 72 +++++++++ .../deerflow/persistence/notifications/sql.py | 97 ++++++++++++ .../deerflow/persistence/skills/__init__.py | 7 + .../deerflow/persistence/skills/base.py | 64 ++++++++ .../deerflow/persistence/skills/memory.py | 126 ++++++++++++++++ .../deerflow/persistence/skills/model.py | 28 ++++ .../deerflow/persistence/skills/sql.py | 139 ++++++++++++++++++ 12 files changed, 692 insertions(+) create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/__init__.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/base.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/memory.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/model.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/sender.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/sql.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/__init__.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/base.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/memory.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/model.py create mode 100644 offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/sql.py diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py index 2c428108..9f996f0e 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/models/__init__.py @@ -18,6 +18,7 @@ there is no matching entity directory. from deerflow.persistence.agents.model import AgentRow from deerflow.persistence.feedback.model import FeedbackRow from deerflow.persistence.models.run_event import RunEventRow +from deerflow.persistence.notifications.model import NotificationRow from deerflow.persistence.run.model import RunRow from deerflow.persistence.scheduled_tasks.model import ( ScheduledTaskDeliveryProfileRow, @@ -26,12 +27,14 @@ from deerflow.persistence.scheduled_tasks.model import ( ScheduledTaskSubscriptionRow, SchedulerThreadRow, ) +from deerflow.persistence.skills.model import SkillRow from deerflow.persistence.thread_meta.model import ThreadMetaRow from deerflow.persistence.user.model import UserRow __all__ = [ "FeedbackRow", "AgentRow", + "NotificationRow", "RunEventRow", "RunRow", "ScheduledTaskDeliveryProfileRow", @@ -39,6 +42,7 @@ __all__ = [ "ScheduledTaskRunRow", "ScheduledTaskSubscriptionRow", "SchedulerThreadRow", + "SkillRow", "ThreadMetaRow", "UserRow", ] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/__init__.py new file mode 100644 index 00000000..281ee363 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/__init__.py @@ -0,0 +1,19 @@ +"""Persistence layer for per-user in-app notifications.""" + +from deerflow.persistence.notifications.base import NotificationStore +from deerflow.persistence.notifications.memory import MemoryNotificationStore +from deerflow.persistence.notifications.sender import ( + get_notification_store, + notify, + set_notification_store, +) +from deerflow.persistence.notifications.sql import NotificationRepository + +__all__ = [ + "NotificationStore", + "MemoryNotificationStore", + "NotificationRepository", + "notify", + "set_notification_store", + "get_notification_store", +] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/base.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/base.py new file mode 100644 index 00000000..0e370cd2 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/base.py @@ -0,0 +1,42 @@ +"""Abstract persistence API for in-app notifications.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import Any + + +class NotificationStore(ABC): + @abstractmethod + async def create(self, data: dict[str, Any]) -> dict[str, Any]: + """Insert a notification record. ``data`` must include ``id``, ``user_id``, ``type``.""" + raise NotImplementedError + + @abstractmethod + async def list_for_user( + self, + user_id: str, + *, + unread_only: bool = False, + limit: int = 50, + ) -> list[dict[str, Any]]: + """Return newest-first records for ``user_id``.""" + raise NotImplementedError + + @abstractmethod + async def unread_count(self, user_id: str) -> int: + raise NotImplementedError + + @abstractmethod + async def mark_read(self, notification_id: str, user_id: str) -> bool: + """Mark a single notification as read; returns False if not owned by user.""" + raise NotImplementedError + + @abstractmethod + async def mark_all_read(self, user_id: str) -> int: + """Mark every unread notification of ``user_id`` as read; returns count updated.""" + raise NotImplementedError + + @abstractmethod + async def delete(self, notification_id: str, user_id: str) -> bool: + raise NotImplementedError diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/memory.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/memory.py new file mode 100644 index 00000000..cf3bfdf5 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/memory.py @@ -0,0 +1,70 @@ +"""In-memory ``NotificationStore`` implementation for tests.""" + +from __future__ import annotations + +import copy +from datetime import UTC, datetime +from typing import Any + +from deerflow.persistence.notifications.base import NotificationStore + + +def _now_iso() -> str: + return datetime.now(UTC).isoformat() + + +class MemoryNotificationStore(NotificationStore): + def __init__(self) -> None: + self._items: dict[str, dict[str, Any]] = {} + + async def create(self, data: dict[str, Any]) -> dict[str, Any]: + record = { + "id": data["id"], + "user_id": data["user_id"], + "type": data["type"], + "title": data.get("title") or "", + "body": data.get("body") or "", + "payload": data.get("payload") or {}, + "read": False, + "created_at": _now_iso(), + } + self._items[record["id"]] = record + return copy.deepcopy(record) + + async def list_for_user( + self, + user_id: str, + *, + unread_only: bool = False, + limit: int = 50, + ) -> list[dict[str, Any]]: + filtered = [r for r in self._items.values() if r["user_id"] == user_id] + if unread_only: + filtered = [r for r in filtered if not r["read"]] + filtered.sort(key=lambda r: r["created_at"], reverse=True) + return [copy.deepcopy(r) for r in filtered[:limit]] + + async def unread_count(self, user_id: str) -> int: + return sum(1 for r in self._items.values() if r["user_id"] == user_id and not r["read"]) + + async def mark_read(self, notification_id: str, user_id: str) -> bool: + record = self._items.get(notification_id) + if record is None or record["user_id"] != user_id: + return False + record["read"] = True + return True + + async def mark_all_read(self, user_id: str) -> int: + updated = 0 + for record in self._items.values(): + if record["user_id"] == user_id and not record["read"]: + record["read"] = True + updated += 1 + return updated + + async def delete(self, notification_id: str, user_id: str) -> bool: + record = self._items.get(notification_id) + if record is None or record["user_id"] != user_id: + return False + del self._items[notification_id] + return True diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/model.py new file mode 100644 index 00000000..eb6d9998 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/model.py @@ -0,0 +1,24 @@ +"""ORM model for in-app notifications.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import JSON, Boolean, DateTime, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from deerflow.persistence.base import Base + + +class NotificationRow(Base): + __tablename__ = "notifications" + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + type: Mapped[str] = mapped_column(String(64), nullable=False) + title: Mapped[str] = mapped_column(String(256), nullable=False, default="") + body: Mapped[str] = mapped_column(Text, nullable=False, default="") + payload: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False, default=dict) + read: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, index=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC), index=True) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/sender.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/sender.py new file mode 100644 index 00000000..c343ac2f --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/sender.py @@ -0,0 +1,72 @@ +"""Helper for writing notifications without callers having to thread a store. + +The harness can NOT import ``app.*`` (firewall), so the Gateway app injects +its ``NotificationStore`` instance via ``set_notification_store(...)`` at +startup. Until that happens, every call is a no-op (silently logged) so +non-Gateway entry points (tests, embedded client) don't crash on missing +infrastructure. +""" + +from __future__ import annotations + +import logging +import uuid +from typing import Any + +from deerflow.persistence.notifications.base import NotificationStore + +logger = logging.getLogger(__name__) + +_store: NotificationStore | None = None + + +def set_notification_store(store: NotificationStore | None) -> None: + """Inject the active ``NotificationStore`` instance for the process. + + Pass ``None`` to clear (e.g. on shutdown or in tests). + """ + global _store + _store = store + + +def get_notification_store() -> NotificationStore | None: + return _store + + +async def notify( + user_id: str, + *, + type: str, + title: str = "", + body: str = "", + payload: dict[str, Any] | None = None, +) -> None: + """Write a notification for ``user_id``. Never raises — failures are logged. + + Args: + user_id: Recipient. + type: Stable identifier (e.g. ``agent_skill_removed``, ``skill_taken_down``). + title: One-line title shown in the inbox. + body: Longer description shown when the user expands the entry. + payload: Structured context (skill_name, agent_id, actor_user_id, ...). + """ + try: + if _store is None: + logger.warning( + "notify(user_id=%s, type=%s) dropped — NotificationStore not configured", + user_id, + type, + ) + return + await _store.create( + { + "id": str(uuid.uuid4()), + "user_id": user_id, + "type": type, + "title": title, + "body": body, + "payload": payload or {}, + } + ) + except Exception: + logger.exception("Failed to write notification (user_id=%s, type=%s)", user_id, type) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/sql.py new file mode 100644 index 00000000..c16461cf --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/sql.py @@ -0,0 +1,97 @@ +"""SQLAlchemy-backed notification repository.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from deerflow.persistence.notifications.base import NotificationStore +from deerflow.persistence.notifications.model import NotificationRow + + +def _row_to_dict(row: NotificationRow) -> dict[str, Any]: + data = row.to_dict() + value = data.get("created_at") + if isinstance(value, datetime): + data["created_at"] = value.isoformat() + return data + + +class NotificationRepository(NotificationStore): + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self._sf = session_factory + + async def create(self, data: dict[str, Any]) -> dict[str, Any]: + row = NotificationRow( + id=data["id"], + user_id=data["user_id"], + type=data["type"], + title=data.get("title") or "", + body=data.get("body") or "", + payload=data.get("payload") or {}, + read=False, + created_at=datetime.now(UTC), + ) + async with self._sf() as session: + session.add(row) + await session.commit() + await session.refresh(row) + return _row_to_dict(row) + + async def list_for_user( + self, + user_id: str, + *, + unread_only: bool = False, + limit: int = 50, + ) -> list[dict[str, Any]]: + stmt = select(NotificationRow).where(NotificationRow.user_id == user_id) + if unread_only: + stmt = stmt.where(NotificationRow.read.is_(False)) + stmt = stmt.order_by(NotificationRow.created_at.desc()).limit(limit) + async with self._sf() as session: + result = await session.execute(stmt) + return [_row_to_dict(row) for row in result.scalars()] + + async def unread_count(self, user_id: str) -> int: + stmt = ( + select(func.count()) + .select_from(NotificationRow) + .where(NotificationRow.user_id == user_id, NotificationRow.read.is_(False)) + ) + async with self._sf() as session: + result = await session.execute(stmt) + return int(result.scalar_one() or 0) + + async def mark_read(self, notification_id: str, user_id: str) -> bool: + async with self._sf() as session: + row = await session.get(NotificationRow, notification_id) + if row is None or row.user_id != user_id: + return False + if not row.read: + row.read = True + await session.commit() + return True + + async def mark_all_read(self, user_id: str) -> int: + stmt = ( + update(NotificationRow) + .where(NotificationRow.user_id == user_id, NotificationRow.read.is_(False)) + .values(read=True) + ) + async with self._sf() as session: + result = await session.execute(stmt) + await session.commit() + return int(result.rowcount or 0) + + async def delete(self, notification_id: str, user_id: str) -> bool: + async with self._sf() as session: + row = await session.get(NotificationRow, notification_id) + if row is None or row.user_id != user_id: + return False + await session.delete(row) + await session.commit() + return True diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/__init__.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/__init__.py new file mode 100644 index 00000000..686e0b2b --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/__init__.py @@ -0,0 +1,7 @@ +"""Persistence layer for custom skills (per-user ownership + publish state).""" + +from deerflow.persistence.skills.base import SkillStore +from deerflow.persistence.skills.memory import MemorySkillStore +from deerflow.persistence.skills.sql import SkillRepository + +__all__ = ["SkillStore", "MemorySkillStore", "SkillRepository"] diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/base.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/base.py new file mode 100644 index 00000000..5a52841e --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/base.py @@ -0,0 +1,64 @@ +"""Abstract persistence API for custom skill ownership + publish state.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import Any + + +class SkillStore(ABC): + @abstractmethod + async def list_visible(self, user_id: str) -> list[dict[str, Any]]: + """List all skill records visible to ``user_id``. + + Visible = owner, OR published by anyone, OR built-in (owner_user_id IS NULL). + """ + raise NotImplementedError + + @abstractmethod + async def get_visible(self, name: str, user_id: str) -> dict[str, Any] | None: + raise NotImplementedError + + @abstractmethod + async def get_any(self, name: str) -> dict[str, Any] | None: + """Lookup by name regardless of ownership/visibility (admin / cascade use).""" + raise NotImplementedError + + @abstractmethod + async def get_owned(self, name: str, user_id: str) -> dict[str, Any] | None: + raise NotImplementedError + + @abstractmethod + async def create(self, data: dict[str, Any]) -> dict[str, Any]: + """Create a skill record. Caller is responsible for global-name uniqueness check.""" + raise NotImplementedError + + @abstractmethod + async def ensure_legacy(self, data: dict[str, Any]) -> dict[str, Any]: + """Idempotently insert a legacy on-disk skill into the DB. Updates the row if it + already exists with ``owner_user_id IS NULL`` (so a re-run can migrate ownership). + """ + raise NotImplementedError + + @abstractmethod + async def update(self, name: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None: + """Owner-only update. Returns None if the caller doesn't own the skill.""" + raise NotImplementedError + + @abstractmethod + async def update_admin(self, name: str, data: dict[str, Any]) -> dict[str, Any] | None: + """Admin override — caller is responsible for auth check. + + Unlike ``update_builtin`` for agents, this accepts ANY skill including + user-owned ones, since the spec lets admin take down others' skills. + """ + raise NotImplementedError + + @abstractmethod + async def delete(self, name: str, user_id: str) -> bool: + raise NotImplementedError + + @abstractmethod + async def delete_admin(self, name: str) -> bool: + """Admin-side delete; bypasses ownership check (caller authorizes).""" + raise NotImplementedError diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/memory.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/memory.py new file mode 100644 index 00000000..09cb4266 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/memory.py @@ -0,0 +1,126 @@ +"""In-memory ``SkillStore`` implementation for tests.""" + +from __future__ import annotations + +import copy +from datetime import UTC, datetime +from typing import Any + +from deerflow.persistence.skills.base import SkillStore + + +def _now() -> str: + return datetime.now(UTC).isoformat() + + +def _copy(record: dict[str, Any] | None) -> dict[str, Any] | None: + return copy.deepcopy(record) if record is not None else None + + +def _with_derived(record: dict[str, Any]) -> dict[str, Any]: + out = dict(record) + out["builtin"] = out.get("owner_user_id") is None + return out + + +class MemorySkillStore(SkillStore): + def __init__(self) -> None: + self._skills: dict[str, dict[str, Any]] = {} + + @staticmethod + def _is_visible(record: dict[str, Any], user_id: str) -> bool: + owner = record.get("owner_user_id") + if owner is None: + return True + if owner == user_id: + return True + return bool(record.get("published")) + + async def list_visible(self, user_id: str) -> list[dict[str, Any]]: + return [ + _with_derived(_copy(record)) # type: ignore[arg-type] + for record in self._skills.values() + if self._is_visible(record, user_id) + ] + + async def get_visible(self, name: str, user_id: str) -> dict[str, Any] | None: + record = self._skills.get(name) + if record is None or not self._is_visible(record, user_id): + return None + return _with_derived(_copy(record)) # type: ignore[arg-type] + + async def get_any(self, name: str) -> dict[str, Any] | None: + record = self._skills.get(name) + if record is None: + return None + return _with_derived(_copy(record)) # type: ignore[arg-type] + + async def get_owned(self, name: str, user_id: str) -> dict[str, Any] | None: + record = self._skills.get(name) + if record is None or record.get("owner_user_id") != user_id: + return None + return _with_derived(_copy(record)) # type: ignore[arg-type] + + async def create(self, data: dict[str, Any]) -> dict[str, Any]: + name = data["name"] + if name in self._skills: + raise ValueError(f"Skill '{name}' already exists") + record = { + "name": name, + "owner_user_id": data.get("owner_user_id"), + "published": bool(data.get("published", False)), + "created_at": _now(), + "updated_at": _now(), + } + self._skills[name] = record + return _with_derived(_copy(record)) # type: ignore[arg-type] + + async def ensure_legacy(self, data: dict[str, Any]) -> dict[str, Any]: + name = data["name"] + record = self._skills.get(name) + if record is None: + record = { + "name": name, + "owner_user_id": data.get("owner_user_id"), + "published": bool(data.get("published", True)), + "created_at": _now(), + "updated_at": _now(), + } + self._skills[name] = record + elif record.get("owner_user_id") is None and data.get("owner_user_id") is not None: + record["owner_user_id"] = data["owner_user_id"] + record["updated_at"] = _now() + return _with_derived(_copy(record)) # type: ignore[arg-type] + + async def update(self, name: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None: + record = self._skills.get(name) + if record is None or record.get("owner_user_id") != user_id: + return None + if "published" in data: + record["published"] = bool(data["published"]) + record["updated_at"] = _now() + return _with_derived(_copy(record)) # type: ignore[arg-type] + + async def update_admin(self, name: str, data: dict[str, Any]) -> dict[str, Any] | None: + record = self._skills.get(name) + if record is None: + return None + if "published" in data: + record["published"] = bool(data["published"]) + if "owner_user_id" in data: + record["owner_user_id"] = data["owner_user_id"] + record["updated_at"] = _now() + return _with_derived(_copy(record)) # type: ignore[arg-type] + + async def delete(self, name: str, user_id: str) -> bool: + record = self._skills.get(name) + if record is None or record.get("owner_user_id") != user_id: + return False + del self._skills[name] + return True + + async def delete_admin(self, name: str) -> bool: + if name not in self._skills: + return False + del self._skills[name] + return True diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/model.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/model.py new file mode 100644 index 00000000..95ee32b9 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/model.py @@ -0,0 +1,28 @@ +"""ORM model for custom skill ownership + publish state. + +Note: skill content lives on disk under ``skills/custom/{name}/``; this table +only stores the metadata (owner, published flag, timestamps). The skill +``name`` doubles as the primary key because the filesystem path is already +keyed by name, and we require global name uniqueness across all owners. +""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import Boolean, DateTime, String +from sqlalchemy.orm import Mapped, mapped_column + +from deerflow.persistence.base import Base + + +class SkillRow(Base): + __tablename__ = "skills" + + # Global unique name — must match the on-disk folder name under skills/custom/. + # Hyphen-case or snake_case allowed; see deerflow.skills.validation. + name: Mapped[str] = mapped_column(String(64), primary_key=True) + owner_user_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True) + published: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, index=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC), onupdate=lambda: datetime.now(UTC)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/sql.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/sql.py new file mode 100644 index 00000000..e54fdb16 --- /dev/null +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/sql.py @@ -0,0 +1,139 @@ +"""SQLAlchemy-backed custom skill repository.""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import or_, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from deerflow.persistence.skills.base import SkillStore +from deerflow.persistence.skills.model import SkillRow + + +def _row_to_dict(row: SkillRow) -> dict[str, Any]: + data = row.to_dict() + data["builtin"] = data.get("owner_user_id") is None + for key in ("created_at", "updated_at"): + value = data.get(key) + if isinstance(value, datetime): + data[key] = value.isoformat() + return data + + +class SkillRepository(SkillStore): + def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None: + self._sf = session_factory + + @staticmethod + def _visible_filter(user_id: str): + return or_( + SkillRow.owner_user_id.is_(None), + SkillRow.owner_user_id == user_id, + SkillRow.published.is_(True), + ) + + async def list_visible(self, user_id: str) -> list[dict[str, Any]]: + stmt = select(SkillRow).where(self._visible_filter(user_id)).order_by(SkillRow.created_at.asc()) + async with self._sf() as session: + result = await session.execute(stmt) + return [_row_to_dict(row) for row in result.scalars()] + + async def get_visible(self, name: str, user_id: str) -> dict[str, Any] | None: + stmt = select(SkillRow).where(SkillRow.name == name, self._visible_filter(user_id)) + async with self._sf() as session: + row = (await session.execute(stmt)).scalar_one_or_none() + return _row_to_dict(row) if row is not None else None + + async def get_any(self, name: str) -> dict[str, Any] | None: + async with self._sf() as session: + row = await session.get(SkillRow, name) + return _row_to_dict(row) if row is not None else None + + async def get_owned(self, name: str, user_id: str) -> dict[str, Any] | None: + stmt = select(SkillRow).where(SkillRow.name == name, SkillRow.owner_user_id == user_id) + async with self._sf() as session: + row = (await session.execute(stmt)).scalar_one_or_none() + return _row_to_dict(row) if row is not None else None + + async def create(self, data: dict[str, Any]) -> dict[str, Any]: + now = datetime.now(UTC) + row = SkillRow( + name=data["name"], + owner_user_id=data.get("owner_user_id"), + published=bool(data.get("published", False)), + created_at=now, + updated_at=now, + ) + async with self._sf() as session: + session.add(row) + await session.commit() + await session.refresh(row) + return _row_to_dict(row) + + async def ensure_legacy(self, data: dict[str, Any]) -> dict[str, Any]: + async with self._sf() as session: + row = await session.get(SkillRow, data["name"]) + if row is None: + now = datetime.now(UTC) + row = SkillRow( + name=data["name"], + owner_user_id=data.get("owner_user_id"), + published=bool(data.get("published", True)), + created_at=now, + updated_at=now, + ) + session.add(row) + elif row.owner_user_id is None and data.get("owner_user_id") is not None: + # Only patch ownership if it was previously null — never overwrite + # an existing owner record on a re-migration. + row.owner_user_id = data["owner_user_id"] + row.updated_at = datetime.now(UTC) + await session.commit() + await session.refresh(row) + return _row_to_dict(row) + + async def update(self, name: str, user_id: str, data: dict[str, Any]) -> dict[str, Any] | None: + async with self._sf() as session: + row = await session.get(SkillRow, name) + if row is None or row.owner_user_id != user_id: + return None + if "published" in data: + row.published = bool(data["published"]) + row.updated_at = datetime.now(UTC) + await session.commit() + await session.refresh(row) + return _row_to_dict(row) + + async def update_admin(self, name: str, data: dict[str, Any]) -> dict[str, Any] | None: + async with self._sf() as session: + row = await session.get(SkillRow, name) + if row is None: + return None + if "published" in data: + row.published = bool(data["published"]) + if "owner_user_id" in data: + row.owner_user_id = data["owner_user_id"] + row.updated_at = datetime.now(UTC) + await session.commit() + await session.refresh(row) + return _row_to_dict(row) + + async def delete(self, name: str, user_id: str) -> bool: + async with self._sf() as session: + row = await session.get(SkillRow, name) + if row is None or row.owner_user_id != user_id: + return False + await session.delete(row) + await session.commit() + return True + + async def delete_admin(self, name: str) -> bool: + async with self._sf() as session: + row = await session.get(SkillRow, name) + if row is None: + return False + await session.delete(row) + await session.commit() + return True