import os
import shutil
import json
import logging
import yaml
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.orm import Session
from typing import List, Optional, Any
from app.db import models
from app.api import schemas
from app.api.dependencies import ServiceContainer, get_current_user, get_db
from app.config import settings
logger = logging.getLogger(__name__)
def create_skills_router(services: ServiceContainer) -> APIRouter:
os.makedirs(settings.SKILLS_DIR, exist_ok=True)
router = APIRouter(prefix="/skills", tags=["Skills"])
# We load the FS loader from our new backend system
from app.core.skills.fs_loader import fs_loader
@router.get("/", response_model=List[Any])
def list_skills(
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user),
feature: Optional[str] = Query(None, description="Filter skills by feature (e.g., 'swarm_control', 'voice_chat')")
):
"""List all skills accessible to the user (via File System)."""
all_skills = fs_loader.get_all_skills()
filtered = []
is_admin = current_user.role == 'admin'
for s in all_skills:
# Enforce RBAC Metadata visibility
s_owner = s.get("owner_id")
s_sys = s.get("is_system", False)
# Admins see everything. Users see their own and system skills.
if not is_admin:
if s_owner != current_user.id and not s_sys:
continue
# If group policies apply, we could add them here in the future
# Filter by Feature Folder
if feature and feature not in s.get("features", []):
continue
filtered.append(s)
return filtered
@router.post("/", response_model=Any)
def create_skill(
skill: schemas.SkillCreate,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user)
):
"""Create a new skill folder on the physical filesystem."""
feature = skill.features[0] if skill.features else "swarm_control"
# We sanitize the name for the folder to prevent path traversal
folder_name = "".join([c for c in skill.name if c.isalnum() or c in (" ", "-", "_")]).strip().replace(" ", "_").lower()
if not folder_name:
raise HTTPException(status_code=400, detail="Invalid skill name.")
skill_path = os.path.join(settings.SKILLS_DIR, feature, folder_name)
if os.path.exists(skill_path):
raise HTTPException(status_code=400, detail="A skill with this ID/folder already exists in this feature sector.")
try:
os.makedirs(skill_path, exist_ok=True)
# Generate the hidden metadata file
meta = {
"owner_id": current_user.id,
"is_system": skill.is_system if current_user.role == 'admin' else False,
"extra_metadata": skill.extra_metadata or {"emoji": "🛠️"}
}
with open(os.path.join(skill_path, ".metadata.json"), "w") as f:
json.dump(meta, f, indent=4)
# Create a boilerplate SKILL.md
content = f"---\nname: {skill.name}\ndescription: {skill.description}\n---\n\n### Execution Logic (Bash)\n```bash\necho \"Executing {skill.name}...\"\n```\n"
with open(os.path.join(skill_path, "SKILL.md"), "w") as f:
f.write(content)
except Exception as e:
logger.error(f"Failed to create FS skill: {e}")
raise HTTPException(status_code=500, detail=str(e))
return {
"id": f"fs-{folder_name}",
"name": skill.name,
"description": skill.description,
"skill_type": "local",
"is_enabled": True,
"features": [feature],
"owner_id": current_user.id,
"is_system": meta["is_system"],
"extra_metadata": meta["extra_metadata"],
"created_at": datetime.utcnow().isoformat()
}
@router.put("/{skill_id}", response_model=Any)
def update_skill(
skill_id: str,
skill_update: schemas.SkillUpdate,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user)
):
"""Update top level metadata of an FS skill (emoji and system flag)."""
# skill_id usually follows 'fs-folder_name' from the frontend
folder_name = skill_id.replace("fs-", "", 1) if skill_id.startswith("fs-") else skill_id
# Scan to find which feature it belongs to, since ID alone doesn't contain feature path
matched_path = None
for feature in os.listdir(settings.SKILLS_DIR):
f_path = os.path.join(settings.SKILLS_DIR, feature, folder_name)
if os.path.exists(f_path):
matched_path = f_path
break
if not matched_path:
raise HTTPException(status_code=404, detail="Skill folder not found.")
meta_path = os.path.join(matched_path, ".metadata.json")
try:
with open(meta_path, "r") as f:
meta = json.load(f)
except:
meta = fs_loader._ensure_metadata(matched_path, folder_name)
if meta.get("is_system") and current_user.role != 'admin':
raise HTTPException(status_code=403, detail="System skills cannot be modified.")
if meta.get("owner_id") != current_user.id and current_user.role != 'admin':
raise HTTPException(status_code=403, detail="Not authorized.")
# Update allowed metadata
if skill_update.extra_metadata is not None:
meta["extra_metadata"] = skill_update.extra_metadata
if skill_update.is_system is not None and current_user.role == 'admin':
meta["is_system"] = skill_update.is_system
# We don't support moving features or renaming folders via API easily right now
# The user can just recreate it to keep it simple, or edit SKILL.md.
with open(meta_path, "w") as f:
json.dump(meta, f, indent=4)
# Read the skill model back via fs_loader logic (slow but perfect sync)
skills = fs_loader.get_all_skills()
for s in skills:
if getattr(s, "id", "") == skill_id or s.get("id") == skill_id:
return s
return {"id": skill_id, "updated": True}
@router.delete("/{skill_id}")
def delete_skill(
skill_id: str,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user)
):
"""Delete a skill permanently via shutil."""
folder_name = skill_id.replace("fs-", "", 1) if skill_id.startswith("fs-") else skill_id
matched_path = None
for feature in os.listdir(settings.SKILLS_DIR):
f_path = os.path.join(settings.SKILLS_DIR, feature, folder_name)
if os.path.exists(f_path):
matched_path = f_path
break
if not matched_path:
raise HTTPException(status_code=404, detail="Skill folder not found.")
meta_path = os.path.join(matched_path, ".metadata.json")
try:
with open(meta_path, "r") as f:
meta = json.load(f)
except:
meta = {}
if meta.get("is_system") and current_user.role != 'admin':
raise HTTPException(status_code=403, detail="System skills cannot be deleted.")
if meta.get("owner_id") != current_user.id and current_user.role != 'admin':
raise HTTPException(status_code=403, detail="Not authorized.")
try:
shutil.rmtree(matched_path)
# Send reload signal
return {"message": "Skill directory successfully deleted."}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Filesystem deletion error: {e}")
# --- Skill File System (VFS) Endpoints ---
@router.get("/{skill_id}/files", response_model=List[Any])
def list_skill_files(
skill_id: str,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user)
):
"""Get file tree for a skill."""
folder_name = skill_id.replace("fs-", "", 1) if skill_id.startswith("fs-") else skill_id
matched_path = None
for feature in os.listdir(settings.SKILLS_DIR):
f_path = os.path.join(settings.SKILLS_DIR, feature, folder_name)
if os.path.exists(f_path):
matched_path = f_path
break
if not matched_path:
raise HTTPException(status_code=404, detail="Skill not found")
files = []
for root, _, filenames in os.walk(matched_path):
for file in filenames:
if file == ".metadata.json" or file.startswith('.'): continue
file_abs = os.path.join(root, file)
file_rel = os.path.relpath(file_abs, matched_path).replace('\\', '/')
files.append({"path": file_rel})
return files
@router.get("/{skill_id}/files/{path:path}", response_model=Any)
def read_skill_file(
skill_id: str,
path: str,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user)
):
"""Read a specific skill file."""
folder_name = skill_id.replace("fs-", "", 1) if skill_id.startswith("fs-") else skill_id
matched_path = None
for feature in os.listdir(settings.SKILLS_DIR):
f_path = os.path.join(settings.SKILLS_DIR, feature, folder_name)
if os.path.exists(f_path):
matched_path = f_path
break
if not matched_path:
raise HTTPException(status_code=404, detail="Skill not found")
file_abs = os.path.join(matched_path, path)
if not os.path.exists(file_abs):
raise HTTPException(status_code=404, detail="File not found on disk")
try:
with open(file_abs, "r", encoding="utf-8") as f:
content = f.read()
return {"file_path": path, "content": content}
except Exception as e:
raise HTTPException(status_code=500, detail="Cannot read physical file.")
@router.post("/{skill_id}/files/{path:path}", response_model=Any)
def create_or_update_skill_file(
skill_id: str,
path: str,
file_update: schemas.SkillFileUpdate,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user)
):
"""Create or update a skill file."""
folder_name = skill_id.replace("fs-", "", 1) if skill_id.startswith("fs-") else skill_id
matched_path = None
for feature in os.listdir(settings.SKILLS_DIR):
f_path = os.path.join(settings.SKILLS_DIR, feature, folder_name)
if os.path.exists(f_path):
matched_path = f_path
break
if not matched_path:
raise HTTPException(status_code=404, detail="Skill not found")
meta_path = os.path.join(matched_path, ".metadata.json")
try:
with open(meta_path, "r") as f:
meta = json.load(f)
except:
meta = {}
# Security Guard
if meta.get("is_system") and current_user.role != 'admin':
raise HTTPException(status_code=403, detail="System skills cannot be modified.")
if meta.get("owner_id") != current_user.id and current_user.role != 'admin':
raise HTTPException(status_code=403, detail="Not authorized to edit skill files")
# Path Traversal Check
file_abs = os.path.abspath(os.path.join(matched_path, path))
if not file_abs.startswith(os.path.abspath(matched_path)):
raise HTTPException(status_code=400, detail="Path traversal not allowed.")
os.makedirs(os.path.dirname(file_abs), exist_ok=True)
try:
with open(file_abs, "w", encoding="utf-8") as f:
f.write(file_update.content)
return {"file_path": path, "content": file_update.content}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Failed to write physical file: {e}")
@router.delete("/{skill_id}/files/{path:path}")
def delete_skill_file(
skill_id: str,
path: str,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_user)
):
"""Delete a skill file literally from disk."""
folder_name = skill_id.replace("fs-", "", 1) if skill_id.startswith("fs-") else skill_id
matched_path = None
for feature in os.listdir(settings.SKILLS_DIR):
f_path = os.path.join(settings.SKILLS_DIR, feature, folder_name)
if os.path.exists(f_path):
matched_path = f_path
break
if not matched_path:
raise HTTPException(status_code=404, detail="Skill not found")
meta_path = os.path.join(matched_path, ".metadata.json")
try:
with open(meta_path, "r") as f:
meta = json.load(f)
except:
meta = {}
if meta.get("is_system") and current_user.role != 'admin':
raise HTTPException(status_code=403, detail="System skills cannot be modified.")
if meta.get("owner_id") != current_user.id and current_user.role != 'admin':
raise HTTPException(status_code=403, detail="Not authorized to edit skill files")
file_abs = os.path.abspath(os.path.join(matched_path, path))
if not file_abs.startswith(os.path.abspath(matched_path)):
raise HTTPException(status_code=400, detail="Path traversal not allowed.")
if not os.path.exists(file_abs):
raise HTTPException(status_code=404, detail="File not found")
try:
os.remove(file_abs)
return {"message": "File removed physically"}
except Exception as e:
raise HTTPException(status_code=500, detail=f"Could not delete: {e}")
return router
