diff --git a/offline-backend-20260512/backend/app/gateway/app.py b/offline-backend-20260512/backend/app/gateway/app.py
index f76af9c6..be0d0b3c 100644
--- a/offline-backend-20260512/backend/app/gateway/app.py
+++ b/offline-backend-20260512/backend/app/gateway/app.py
@@ -30,6 +30,7 @@ from app.gateway.routers import (
     mcp,
     memory,
     models,
+    notifications,
     runs,
     scheduled_tasks,
     skills,
@@ -152,6 +153,68 @@ async def _iter_store_items(store, namespace, *, page_size: int = 500):
         offset += page_size
 
 
+async def _sync_legacy_skills(app: FastAPI) -> None:
+    """Idempotently register existing on-disk ``skills/custom/*`` directories
+    in the ``skills`` DB table, owned by the first admin and published=True.
+
+    Without this step, after upgrading to per-user skill ownership the existing
+    custom skill folders would not be visible to anyone (the SkillStore
+    visibility filter relies on a DB row). Running this once on every boot is
+    cheap (folder scan + one ``ensure_legacy`` per name) and idempotent.
+
+    Non-fatal: a failure here MUST NOT block server startup.
+    """
+    from sqlalchemy import select
+
+    from deerflow.persistence.engine import get_session_factory
+    from deerflow.persistence.user.model import UserRow
+    from deerflow.skills.storage import get_or_new_skill_storage
+    from deerflow.skills.types import SkillCategory
+
+    sf = get_session_factory()
+    if sf is None:
+        return
+
+    skill_store = getattr(app.state, "skill_store", None)
+    if skill_store is None:
+        return
+
+    async with sf() as session:
+        stmt = select(UserRow).where(UserRow.system_role == "admin").limit(1)
+        row = (await session.execute(stmt)).scalar_one_or_none()
+    if row is None:
+        # No admin yet — leave ownership null; first admin is responsible for
+        # claiming via the UI once they exist.
+        admin_id: str | None = None
+    else:
+        admin_id = str(row.id)
+
+    config = getattr(app.state, "config", None)
+    storage = get_or_new_skill_storage(app_config=config)
+    try:
+        skills = storage.load_skills(enabled_only=False)
+    except Exception:
+        logger.exception("_sync_legacy_skills: could not list on-disk skills")
+        return
+
+    inserted = 0
+    for skill in skills:
+        if skill.category != SkillCategory.CUSTOM:
+            continue
+        try:
+            existing = await skill_store.get_any(skill.name)
+            if existing is not None:
+                continue
+            await skill_store.ensure_legacy(
+                {"name": skill.name, "owner_user_id": admin_id, "published": True}
+            )
+            inserted += 1
+        except Exception:
+            logger.warning("_sync_legacy_skills: could not register %s", skill.name, exc_info=True)
+    if inserted:
+        logger.info("Skill ownership migration: registered %d legacy custom skill(s)", inserted)
+
+
 async def _migrate_orphaned_threads(store, admin_user_id: str) -> int:
     """Migrate LangGraph store threads with no user_id to the given admin.
 
@@ -228,6 +291,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         # Must run AFTER langgraph_runtime so app.state.store is available for thread migration
         await _ensure_admin_user(app)
 
+        # Migrate "ownerless" on-disk skills into the new SkillStore table —
+        # makes the multitenancy switch transparent for existing deployments.
+        try:
+            await _sync_legacy_skills(app)
+        except Exception:
+            logger.exception("Skill ownership migration failed (non-fatal)")
+
         # Kick off skill curator in background if the run interval has elapsed
         try:
             from deerflow.skills.curator import maybe_run_curator
@@ -510,6 +580,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # Feedback API is mounted at /api/threads/{thread_id}/runs/{run_id}/feedback
     app.include_router(feedback.router)
 
+    # Notifications API is mounted at /api/notifications
+    app.include_router(notifications.router)
+
     # Thread Runs API (LangGraph Platform-compatible runs lifecycle)
     app.include_router(thread_runs.router)
 
diff --git a/offline-backend-20260512/backend/app/gateway/deps.py b/offline-backend-20260512/backend/app/gateway/deps.py
index e163c425..f4a0fb70 100644
--- a/offline-backend-20260512/backend/app/gateway/deps.py
+++ b/offline-backend-20260512/backend/app/gateway/deps.py
@@ -25,7 +25,9 @@ if TYPE_CHECKING:
     from app.gateway.auth.local_provider import LocalAuthProvider
     from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
     from deerflow.persistence.agents import AgentStore
+    from deerflow.persistence.notifications import NotificationStore
     from deerflow.persistence.scheduled_tasks import ScheduledTaskStore
+    from deerflow.persistence.skills import SkillStore
     from deerflow.persistence.thread_meta.base import ThreadMetaStore
 
 
@@ -94,6 +96,20 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
 
         app.state.agent_store = make_agent_store(sf)
 
+        from deerflow.persistence.skills import make_skill_store
+
+        app.state.skill_store = make_skill_store(sf)
+
+        from deerflow.persistence.notifications import (
+            make_notification_store,
+            set_notification_store,
+        )
+
+        app.state.notification_store = make_notification_store(sf)
+        # Make the sender helper (harness layer) able to write without going
+        # through FastAPI request context — used by cascade.py.
+        set_notification_store(app.state.notification_store)
+
         # Run event store (has its own factory with config-driven backend selection)
         run_events_config = getattr(config, "run_events", None)
         app.state.run_event_store = make_run_event_store(run_events_config)
@@ -104,6 +120,9 @@ async def langgraph_runtime(app: FastAPI) -> AsyncGenerator[None, None]:
         try:
             yield
         finally:
+            from deerflow.persistence.notifications import set_notification_store
+
+            set_notification_store(None)
             await close_engine()
 
 
@@ -162,6 +181,22 @@ def get_agent_store(request: Request) -> AgentStore:
     return val
 
 
+def get_skill_store(request: Request) -> SkillStore:
+    """Return the custom-skill store."""
+    val = getattr(request.app.state, "skill_store", None)
+    if val is None:
+        raise HTTPException(status_code=503, detail="Skill store not available")
+    return val
+
+
+def get_notification_store(request: Request) -> NotificationStore:
+    """Return the in-app notification store."""
+    val = getattr(request.app.state, "notification_store", None)
+    if val is None:
+        raise HTTPException(status_code=503, detail="Notification store not available")
+    return val
+
+
 def get_run_context(request: Request) -> RunContext:
     """Build a :class:`RunContext` from ``app.state`` singletons.
 
diff --git a/offline-backend-20260512/backend/app/gateway/routers/notifications.py b/offline-backend-20260512/backend/app/gateway/routers/notifications.py
new file mode 100644
index 00000000..12438c8b
--- /dev/null
+++ b/offline-backend-20260512/backend/app/gateway/routers/notifications.py
@@ -0,0 +1,152 @@
+"""In-app notification inbox API."""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request
+from pydantic import BaseModel, Field
+
+from app.gateway.deps import get_notification_store
+from deerflow.persistence.notifications import NotificationStore
+from deerflow.runtime.user_context import get_effective_user_id
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api", tags=["notifications"])
+
+
+def _current_user_id(request: Request) -> str:
+    user = getattr(request.state, "user", None)
+    if user is not None:
+        return str(user.id)
+    return get_effective_user_id()
+
+
+class NotificationItem(BaseModel):
+    id: str
+    user_id: str
+    type: str
+    title: str
+    body: str
+    payload: dict[str, Any]
+    read: bool
+    created_at: str | None
+
+
+class NotificationListResponse(BaseModel):
+    items: list[NotificationItem]
+
+
+class NotificationUnreadCountResponse(BaseModel):
+    unread: int = Field(..., description="Number of unread notifications for the current user")
+
+
+class MarkReadResponse(BaseModel):
+    success: bool
+
+
+class MarkAllReadResponse(BaseModel):
+    updated: int
+
+
+@router.get(
+    "/notifications",
+    response_model=NotificationListResponse,
+    summary="List Notifications",
+)
+async def list_notifications(
+    request: Request,
+    unread_only: bool = Query(default=False),
+    limit: int = Query(default=50, ge=1, le=200),
+    store: NotificationStore = Depends(get_notification_store),
+) -> NotificationListResponse:
+    try:
+        items = await store.list_for_user(
+            _current_user_id(request),
+            unread_only=unread_only,
+            limit=limit,
+        )
+        return NotificationListResponse(items=[NotificationItem(**item) for item in items])
+    except Exception as e:
+        logger.error("Failed to list notifications: %s", e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to list notifications: {e}")
+
+
+@router.get(
+    "/notifications/unread_count",
+    response_model=NotificationUnreadCountResponse,
+    summary="Unread Notification Count",
+)
+async def unread_count(
+    request: Request,
+    store: NotificationStore = Depends(get_notification_store),
+) -> NotificationUnreadCountResponse:
+    try:
+        count = await store.unread_count(_current_user_id(request))
+        return NotificationUnreadCountResponse(unread=count)
+    except Exception as e:
+        logger.error("Failed to read unread notification count: %s", e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to read unread count: {e}")
+
+
+@router.post(
+    "/notifications/{notification_id}/read",
+    response_model=MarkReadResponse,
+    summary="Mark Notification Read",
+)
+async def mark_read(
+    notification_id: str,
+    request: Request,
+    store: NotificationStore = Depends(get_notification_store),
+) -> MarkReadResponse:
+    try:
+        ok = await store.mark_read(notification_id, _current_user_id(request))
+        if not ok:
+            raise HTTPException(status_code=404, detail="Notification not found")
+        return MarkReadResponse(success=True)
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("Failed to mark notification read: %s", e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to mark read: {e}")
+
+
+@router.post(
+    "/notifications/read_all",
+    response_model=MarkAllReadResponse,
+    summary="Mark All Notifications Read",
+)
+async def mark_all_read(
+    request: Request,
+    store: NotificationStore = Depends(get_notification_store),
+) -> MarkAllReadResponse:
+    try:
+        count = await store.mark_all_read(_current_user_id(request))
+        return MarkAllReadResponse(updated=count)
+    except Exception as e:
+        logger.error("Failed to mark all notifications read: %s", e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to mark all read: {e}")
+
+
+@router.delete(
+    "/notifications/{notification_id}",
+    response_model=MarkReadResponse,
+    summary="Delete Notification",
+)
+async def delete_notification(
+    notification_id: str,
+    request: Request,
+    store: NotificationStore = Depends(get_notification_store),
+) -> MarkReadResponse:
+    try:
+        ok = await store.delete(notification_id, _current_user_id(request))
+        if not ok:
+            raise HTTPException(status_code=404, detail="Notification not found")
+        return MarkReadResponse(success=True)
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("Failed to delete notification: %s", e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to delete: {e}")
diff --git a/offline-backend-20260512/backend/app/gateway/routers/skills.py b/offline-backend-20260512/backend/app/gateway/routers/skills.py
index 2d1b4122..63878141 100644
--- a/offline-backend-20260512/backend/app/gateway/routers/skills.py
+++ b/offline-backend-20260512/backend/app/gateway/routers/skills.py
@@ -3,16 +3,20 @@ import logging
 import tempfile
 from datetime import datetime
 from pathlib import Path
+from typing import Any
 
-from fastapi import APIRouter, Body, Depends, File, HTTPException, UploadFile
+from fastapi import APIRouter, Body, Depends, File, HTTPException, Request, UploadFile
 from pydantic import BaseModel, Field
 
-from app.gateway.deps import get_config
+from app.gateway.deps import get_agent_store, get_config, get_skill_store
 from app.gateway.path_utils import resolve_thread_virtual_path
 from deerflow.agents.lead_agent.prompt import refresh_skills_system_prompt_cache_async
 from deerflow.config.app_config import AppConfig
 from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, get_extensions_config, reload_extensions_config
+from deerflow.persistence.skills import SkillStore
+from deerflow.runtime.user_context import get_effective_user_id
 from deerflow.skills import Skill
+from deerflow.skills.cascade import cascade_skill_unpublished_or_deleted
 from deerflow.skills.installer import SkillAlreadyExistsError
 from deerflow.skills.security_scanner import scan_skill_content
 from deerflow.skills.storage import get_or_new_skill_storage
@@ -23,6 +27,56 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api", tags=["skills"])
 
 
+# ---------------------------------------------------------------------------
+# Per-user authz helpers (mirrors routers/agents.py)
+# ---------------------------------------------------------------------------
+
+
+def _current_user_id(request: Request) -> str:
+    user = getattr(request.state, "user", None)
+    if user is not None:
+        return str(user.id)
+    return get_effective_user_id()
+
+
+def _is_admin_user(request: Request) -> bool:
+    user = getattr(request.state, "user", None)
+    return getattr(user, "system_role", None) == "admin"
+
+
+async def _get_owned_skill_or_raise(store: SkillStore, name: str, user_id: str) -> dict[str, Any]:
+    owned = await store.get_owned(name, user_id)
+    if owned is not None:
+        return owned
+    visible = await store.get_visible(name, user_id)
+    if visible is None:
+        raise HTTPException(status_code=404, detail=f"Skill '{name}' not found")
+    raise HTTPException(status_code=403, detail="You can only modify skills that you own")
+
+
+async def _get_editable_skill_or_raise(
+    store: SkillStore,
+    name: str,
+    user_id: str,
+    *,
+    is_admin: bool,
+) -> tuple[dict[str, Any], str]:
+    """Return (record, edit_mode) where edit_mode is 'owner' or 'admin'.
+
+    Admins can edit/delete any skill (including others' published ones, since
+    the spec lets them force-takedown). Non-admins are limited to their own.
+    """
+    owned = await store.get_owned(name, user_id)
+    if owned is not None:
+        return owned, "owner"
+    record = await store.get_any(name)
+    if record is None:
+        raise HTTPException(status_code=404, detail=f"Skill '{name}' not found")
+    if is_admin:
+        return record, "admin"
+    raise HTTPException(status_code=403, detail="You can only modify skills that you own")
+
+
 class SkillResponse(BaseModel):
     """Response model for skill information."""
 
@@ -38,6 +92,9 @@ class SkillResponse(BaseModel):
     state: str = Field(default="active", description="Lifecycle state: active, stale, or archived")
     pinned: bool = Field(default=False, description="Whether this skill is pinned (excluded from curation)")
     source: str = Field(default="upload", description="Origin: 'agent' for agent-created, 'upload' for user-uploaded")
+    owner_user_id: str | None = Field(default=None, description="Custom skill owner (null = built-in or legacy)")
+    published: bool = Field(default=False, description="Whether the owner has made this skill visible to other users")
+    builtin: bool = Field(default=False, description="True when owner_user_id is null (no human owner)")
 
 
 class SkillsListResponse(BaseModel):
@@ -97,8 +154,14 @@ class SkillRollbackRequest(BaseModel):
     history_index: int = Field(default=-1, description="History entry index to restore from, defaulting to the latest change.")
 
 
-def _skill_to_response(skill: Skill) -> SkillResponse:
-    """Convert a Skill object to a SkillResponse."""
+def _skill_to_response(skill: Skill, ownership: dict[str, Any] | None = None) -> SkillResponse:
+    """Convert a Skill object to a SkillResponse.
+
+    ``ownership`` is the skill's record from ``SkillStore`` (or None for public
+    built-in skills which never have a DB row). When present, contributes the
+    ``owner_user_id`` / ``published`` / ``builtin`` fields used for per-user
+    visibility and the frontend's owner badges.
+    """
     from deerflow.skills.usage import get_entry, latest_activity_at
 
     entry = get_entry(skill.name)
@@ -109,6 +172,15 @@ def _skill_to_response(skill: Skill) -> SkillResponse:
             last_at = datetime.fromisoformat(last_at_str)
         except Exception:
             pass
+
+    owner_user_id: str | None = None
+    published = False
+    builtin = skill.category != SkillCategory.CUSTOM
+    if ownership is not None:
+        owner_user_id = ownership.get("owner_user_id")
+        published = bool(ownership.get("published", False))
+        builtin = bool(ownership.get("builtin", owner_user_id is None))
+
     return SkillResponse(
         name=skill.name,
         description=skill.description,
@@ -122,19 +194,45 @@ def _skill_to_response(skill: Skill) -> SkillResponse:
         state=entry.get("state", "active"),
         pinned=entry.get("pinned", False),
         source=entry.get("source", "upload"),
+        owner_user_id=owner_user_id,
+        published=published,
+        builtin=builtin,
     )
 
 
+async def _build_ownership_index(skill_store: SkillStore, user_id: str) -> dict[str, dict[str, Any]]:
+    """Index visible-to-user skill records by name. Internal call cache for list endpoints."""
+    records = await skill_store.list_visible(user_id)
+    return {record["name"]: record for record in records}
+
+
 @router.get(
     "/skills",
     response_model=SkillsListResponse,
     summary="List All Skills",
     description="Retrieve a list of all available skills from both public and custom directories.",
 )
-async def list_skills(config: AppConfig = Depends(get_config)) -> SkillsListResponse:
+async def list_skills(
+    request: Request,
+    config: AppConfig = Depends(get_config),
+    skill_store: SkillStore = Depends(get_skill_store),
+) -> SkillsListResponse:
     try:
-        skills = get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False)
-        return SkillsListResponse(skills=[_skill_to_response(skill) for skill in skills])
+        user_id = _current_user_id(request)
+        all_skills = get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False)
+        ownership = await _build_ownership_index(skill_store, user_id)
+        responses: list[SkillResponse] = []
+        for skill in all_skills:
+            # Public/built-in skills (no DB row) are visible to everyone.
+            # Custom skills must have an ownership record in the visible set.
+            if skill.category != SkillCategory.CUSTOM:
+                responses.append(_skill_to_response(skill))
+                continue
+            record = ownership.get(skill.name)
+            if record is None:
+                continue
+            responses.append(_skill_to_response(skill, record))
+        return SkillsListResponse(skills=responses)
     except Exception as e:
         logger.error(f"Failed to load skills: {e}", exc_info=True)
         raise HTTPException(status_code=500, detail=f"Failed to load skills: {str(e)}")
@@ -219,19 +317,43 @@ async def validate_skill_upload(
     description="Install a skill directly from an uploaded .zip/.skill archive. Set ?overwrite=true to replace an existing skill with the same name.",
 )
 async def install_skill_upload(
+    request: Request,
     file: UploadFile = File(..., description="The .zip or .skill archive to install"),
     overwrite: bool = False,
     config: AppConfig = Depends(get_config),
+    skill_store: SkillStore = Depends(get_skill_store),
 ) -> SkillInstallResponse:
+    user_id = _current_user_id(request)
     tmp_path = await _persist_uploaded_skill(file)
     try:
         storage = get_or_new_skill_storage(app_config=config)
+        # Pre-parse the archive to learn the skill name without writing anything,
+        # so we can reject the upload before touching the filesystem if another
+        # user already owns this name.
+        try:
+            validation = storage.validate_skill_archive(tmp_path)
+            preview_name = validation.get("skill_name") if isinstance(validation, dict) else None
+        except Exception:
+            preview_name = None
+        if preview_name:
+            existing = await skill_store.get_any(preview_name)
+            if existing and existing.get("owner_user_id") not in (None, user_id):
+                raise HTTPException(
+                    status_code=409,
+                    detail=f"Skill '{preview_name}' already exists and is owned by another user",
+                )
         result = await storage.ainstall_skill_from_archive(tmp_path, overwrite=overwrite, require_skill_suffix=False)
         try:
             from deerflow.skills.usage import mark_uploaded
             mark_uploaded(result["skill_name"])
         except Exception:
             pass
+        # Register ownership in the SkillStore. ``ensure_legacy`` is idempotent
+        # and only patches a NULL owner; it never overwrites an existing owner,
+        # so the pre-parse check above is what really guards cross-user steals.
+        await skill_store.ensure_legacy(
+            {"name": result["skill_name"], "owner_user_id": user_id, "published": False}
+        )
         await refresh_skills_system_prompt_cache_async()
         return SkillInstallResponse(**result)
     except FileNotFoundError as e:
@@ -255,15 +377,24 @@ async def install_skill_upload(
     summary="Install Skill",
     description="Install a skill from a .skill file (ZIP archive) located in the thread's user-data directory.",
 )
-async def install_skill(request: SkillInstallRequest, config: AppConfig = Depends(get_config)) -> SkillInstallResponse:
+async def install_skill(
+    body: SkillInstallRequest,
+    request: Request,
+    config: AppConfig = Depends(get_config),
+    skill_store: SkillStore = Depends(get_skill_store),
+) -> SkillInstallResponse:
     try:
-        skill_file_path = resolve_thread_virtual_path(request.thread_id, request.path)
+        user_id = _current_user_id(request)
+        skill_file_path = resolve_thread_virtual_path(body.thread_id, body.path)
         result = await get_or_new_skill_storage(app_config=config).ainstall_skill_from_archive(skill_file_path)
         try:
             from deerflow.skills.usage import mark_uploaded
             mark_uploaded(result["skill_name"])
         except Exception:
             pass
+        await skill_store.ensure_legacy(
+            {"name": result["skill_name"], "owner_user_id": user_id, "published": False}
+        )
         await refresh_skills_system_prompt_cache_async()
         return SkillInstallResponse(**result)
     except FileNotFoundError as e:
@@ -280,10 +411,21 @@ async def install_skill(request: SkillInstallRequest, config: AppConfig = Depend
 
 
 @router.get("/skills/custom", response_model=SkillsListResponse, summary="List Custom Skills")
-async def list_custom_skills(config: AppConfig = Depends(get_config)) -> SkillsListResponse:
+async def list_custom_skills(
+    request: Request,
+    config: AppConfig = Depends(get_config),
+    skill_store: SkillStore = Depends(get_skill_store),
+) -> SkillsListResponse:
     try:
-        skills = [skill for skill in get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False) if skill.category == SkillCategory.CUSTOM]
-        return SkillsListResponse(skills=[_skill_to_response(skill) for skill in skills])
+        user_id = _current_user_id(request)
+        custom = [s for s in get_or_new_skill_storage(app_config=config).load_skills(enabled_only=False) if s.category == SkillCategory.CUSTOM]
+        ownership = await _build_ownership_index(skill_store, user_id)
+        responses = [
+            _skill_to_response(skill, ownership[skill.name])
+            for skill in custom
+            if skill.name in ownership
+        ]
+        return SkillsListResponse(skills=responses)
     except Exception as e:
         logger.error("Failed to list custom skills: %s", e, exc_info=True)
         raise HTTPException(status_code=500, detail=f"Failed to list custom skills: {str(e)}")
@@ -348,9 +490,20 @@ async def update_custom_skill(skill_name: str, request: CustomSkillUpdateRequest
 
 
 @router.delete("/skills/custom/{skill_name}", summary="Delete Custom Skill")
-async def delete_custom_skill(skill_name: str, config: AppConfig = Depends(get_config)) -> dict[str, bool]:
+async def delete_custom_skill(
+    skill_name: str,
+    request: Request,
+    config: AppConfig = Depends(get_config),
+    skill_store: SkillStore = Depends(get_skill_store),
+    agent_store=Depends(get_agent_store),
+) -> dict[str, bool]:
     try:
         skill_name = skill_name.replace("\r\n", "").replace("\n", "")
+        user_id = _current_user_id(request)
+        is_admin = _is_admin_user(request)
+        record, mode = await _get_editable_skill_or_raise(skill_store, skill_name, user_id, is_admin=is_admin)
+        operation = "admin_takedown" if mode == "admin" and record.get("owner_user_id") not in (None, user_id) else "delete"
+
         storage = get_or_new_skill_storage(app_config=config)
         storage.delete_custom_skill(
             skill_name,
@@ -369,8 +522,27 @@ async def delete_custom_skill(skill_name: str, config: AppConfig = Depends(get_c
             forget(skill_name)
         except Exception:
             pass
+
+        # Persist ownership record removal so the DB stays consistent.
+        if mode == "admin":
+            await skill_store.delete_admin(skill_name)
+        else:
+            await skill_store.delete(skill_name, user_id)
+
+        # Strip references from every agent that pulled this skill in,
+        # and notify their owners (no-op for self).
+        await cascade_skill_unpublished_or_deleted(
+            skill_name,
+            actor_user_id=user_id,
+            operation=operation,
+            skill_owner_user_id=record.get("owner_user_id"),
+            agent_store=agent_store,
+        )
+
         await refresh_skills_system_prompt_cache_async()
         return {"success": True}
+    except HTTPException:
+        raise
     except FileNotFoundError as e:
         raise HTTPException(status_code=404, detail=str(e))
     except ValueError as e:
@@ -380,6 +552,71 @@ async def delete_custom_skill(skill_name: str, config: AppConfig = Depends(get_c
         raise HTTPException(status_code=500, detail=f"Failed to delete custom skill: {str(e)}")
 
 
+class SkillPublishRequest(BaseModel):
+    """Request body for toggling a custom skill's published flag."""
+
+    published: bool = Field(..., description="Whether to publish (True) or unpublish (False)")
+
+
+class SkillPublishResponse(BaseModel):
+    name: str
+    owner_user_id: str | None
+    published: bool
+
+
+@router.put(
+    "/skills/custom/{skill_name}/published",
+    response_model=SkillPublishResponse,
+    summary="Toggle Custom Skill Published State",
+)
+async def update_custom_skill_published(
+    skill_name: str,
+    body: SkillPublishRequest,
+    request: Request,
+    skill_store: SkillStore = Depends(get_skill_store),
+    agent_store=Depends(get_agent_store),
+) -> SkillPublishResponse:
+    try:
+        skill_name = skill_name.replace("\r\n", "").replace("\n", "")
+        user_id = _current_user_id(request)
+        is_admin = _is_admin_user(request)
+        record, mode = await _get_editable_skill_or_raise(skill_store, skill_name, user_id, is_admin=is_admin)
+
+        was_published = bool(record.get("published", False))
+        if mode == "admin":
+            updated = await skill_store.update_admin(skill_name, {"published": body.published})
+        else:
+            updated = await skill_store.update(skill_name, user_id, {"published": body.published})
+        if updated is None:
+            raise HTTPException(status_code=500, detail="Failed to persist published state")
+
+        # If we are *removing* visibility (was published, now not), strip from
+        # other users' agents and notify them. admin takedown is signaled by
+        # the actor not being the owner.
+        if was_published and not body.published:
+            skill_owner = record.get("owner_user_id")
+            operation = "admin_takedown" if mode == "admin" and skill_owner not in (None, user_id) else "unpublish"
+            await cascade_skill_unpublished_or_deleted(
+                skill_name,
+                actor_user_id=user_id,
+                operation=operation,
+                skill_owner_user_id=skill_owner,
+                agent_store=agent_store,
+            )
+            await refresh_skills_system_prompt_cache_async()
+
+        return SkillPublishResponse(
+            name=updated["name"],
+            owner_user_id=updated.get("owner_user_id"),
+            published=bool(updated.get("published", False)),
+        )
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("Failed to update published state for %s: %s", skill_name, e, exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Failed to update published state: {str(e)}")
+
+
 @router.get("/skills/custom/{skill_name}/history", response_model=CustomSkillHistoryResponse, summary="Get Custom Skill History")
 async def get_custom_skill_history(skill_name: str, config: AppConfig = Depends(get_config)) -> CustomSkillHistoryResponse:
     try:
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
index 281ee363..f23d549b 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/__init__.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/notifications/__init__.py
@@ -16,4 +16,11 @@ __all__ = [
     "notify",
     "set_notification_store",
     "get_notification_store",
+    "make_notification_store",
 ]
+
+
+def make_notification_store(session_factory):
+    if session_factory is None:
+        return MemoryNotificationStore()
+    return NotificationRepository(session_factory)
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
index 686e0b2b..b98b9fa8 100644
--- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/__init__.py
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/skills/__init__.py
@@ -4,4 +4,10 @@ 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"]
+__all__ = ["SkillStore", "MemorySkillStore", "SkillRepository", "make_skill_store"]
+
+
+def make_skill_store(session_factory):
+    if session_factory is None:
+        return MemorySkillStore()
+    return SkillRepository(session_factory)
diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/skills/cascade.py b/offline-backend-20260512/backend/packages/harness/deerflow/skills/cascade.py
new file mode 100644
index 00000000..17d3ae0f
--- /dev/null
+++ b/offline-backend-20260512/backend/packages/harness/deerflow/skills/cascade.py
@@ -0,0 +1,151 @@
+"""Cascade handling when a published skill is removed from circulation.
+
+Called from the Gateway skills router whenever a published skill is
+unpublished, deleted, or admin-taken-down. Walks every agent on disk that
+referenced the skill, strips the skill from its ``config.yaml`` ``skills``
+list, and emits an in-app notification to that agent's owner.
+
+The harness/app firewall means we cannot import the FastAPI app side; the
+``AgentStore`` instance must be passed in by the caller (the router gets
+it from ``app.gateway.deps.get_agent_store``).
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Any, Literal
+
+import yaml
+
+from deerflow.config.agents_config import load_agent_config
+from deerflow.config.paths import get_paths
+from deerflow.persistence.notifications import notify
+
+logger = logging.getLogger(__name__)
+
+CascadeOperation = Literal["unpublish", "delete", "admin_takedown"]
+
+
+def _list_agent_dirs() -> list[Path]:
+    agents_dir = get_paths().agents_dir
+    if not agents_dir.exists():
+        return []
+    return [entry for entry in sorted(agents_dir.iterdir()) if entry.is_dir()]
+
+
+def _remove_skill_from_yaml(config_file: Path, skill_name: str) -> bool:
+    """Rewrite ``config.yaml`` removing ``skill_name`` from its ``skills`` list.
+
+    Returns True if the file was modified, False otherwise.
+    """
+    if not config_file.exists():
+        return False
+    try:
+        with config_file.open("r", encoding="utf-8") as f:
+            data: dict[str, Any] = yaml.safe_load(f) or {}
+    except yaml.YAMLError:
+        logger.warning("Cascade: failed to parse %s, skipping", config_file)
+        return False
+
+    skills = data.get("skills")
+    if not isinstance(skills, list) or skill_name not in skills:
+        return False
+
+    data["skills"] = [s for s in skills if s != skill_name]
+    with config_file.open("w", encoding="utf-8") as f:
+        yaml.safe_dump(data, f, default_flow_style=False, allow_unicode=True)
+    return True
+
+
+_OP_LABELS: dict[CascadeOperation, str] = {
+    "unpublish": "取消发布",
+    "delete": "删除",
+    "admin_takedown": "管理员强制下架",
+}
+
+
+async def cascade_skill_unpublished_or_deleted(
+    skill_name: str,
+    *,
+    actor_user_id: str,
+    operation: CascadeOperation,
+    skill_owner_user_id: str | None,
+    agent_store: Any,
+) -> dict[str, Any]:
+    """Detach ``skill_name`` from every referencing agent and notify owners.
+
+    Self-references (agent.user_id == actor_user_id) are still detached but
+    the owner gets no in-app notification — the user just did the action,
+    they know.
+
+    On ``admin_takedown`` we additionally notify ``skill_owner_user_id``.
+
+    Returns counters for testability: ``{checked, modified, notified}``.
+    """
+    counters = {"checked": 0, "modified": 0, "notified": 0}
+    label = _OP_LABELS.get(operation, operation)
+
+    for agent_dir in _list_agent_dirs():
+        config_file = agent_dir / "config.yaml"
+        if not config_file.exists():
+            continue
+        counters["checked"] += 1
+
+        # Quick read to decide whether we need to mutate at all.
+        try:
+            agent_cfg = load_agent_config(agent_dir.name)
+        except Exception:
+            logger.debug("Cascade: failed to load %s, skipping", agent_dir.name, exc_info=True)
+            continue
+        if agent_cfg is None or not agent_cfg.skills or skill_name not in agent_cfg.skills:
+            continue
+
+        modified = _remove_skill_from_yaml(config_file, skill_name)
+        if not modified:
+            continue
+        counters["modified"] += 1
+        logger.info(
+            "Cascade: removed skill '%s' from agent '%s/config.yaml' (op=%s)",
+            skill_name,
+            agent_dir.name,
+            operation,
+        )
+
+        # Look up agent owner from DB to decide where to send the notification.
+        record = await agent_store.get_any(agent_dir.name)
+        agent_owner = record.get("user_id") if record else None
+        if agent_owner and agent_owner != actor_user_id:
+            agent_display = (record.get("name") if record else None) or agent_dir.name
+            await notify(
+                agent_owner,
+                type="agent_skill_removed",
+                title=f"专家 “{agent_display}” 的一个技能已失效",
+                body=(
+                    f"技能 “{skill_name}” 被 {label}，已从该专家的技能列表中自动移除。"
+                ),
+                payload={
+                    "skill_name": skill_name,
+                    "agent_id": agent_dir.name,
+                    "agent_name": agent_display,
+                    "operation": operation,
+                    "actor_user_id": actor_user_id,
+                },
+            )
+            counters["notified"] += 1
+
+    if operation == "admin_takedown" and skill_owner_user_id and skill_owner_user_id != actor_user_id:
+        await notify(
+            skill_owner_user_id,
+            type="skill_taken_down",
+            title=f"你的技能 “{skill_name}” 已被管理员下架",
+            body="管理员对该技能执行了强制下架。如有疑问请联系管理员。",
+            payload={
+                "skill_name": skill_name,
+                "operation": operation,
+                "actor_user_id": actor_user_id,
+            },
+        )
+        counters["notified"] += 1
+
+    return counters