feat(skills/notifications): 多租户阶段 1 — 持久化层

新增两套 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) <noreply@anthropic.com>

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

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

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

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

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)

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)

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

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

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

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

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))

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