mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-08-17 18:25:42 +02:00
129 lines
4.1 KiB
Python
129 lines
4.1 KiB
Python
import json
|
|
import os
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
from fastapi import HTTPException
|
|
from backend.config.Apps import SubApp
|
|
from backend.apps.modes.models import Mode, ModeCreate, ModeUpdate, BUILTIN_MODES
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
from backend.config.paths import MODES_DIR as DATA_DIR
|
|
from backend.config.json_store import read_json_or_none, atomic_write_json
|
|
|
|
|
|
@asynccontextmanager
|
|
async def modes_lifespan():
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
# Migration: Chat merged into Ask; drop a stale built-in chat.json but leave customized copies alone.
|
|
chat_path = os.path.join(DATA_DIR, "chat.json")
|
|
if os.path.exists(chat_path):
|
|
try:
|
|
import json as p_json
|
|
with open(chat_path) as p_f:
|
|
p_data = p_json.load(p_f)
|
|
if p_data.get("is_builtin") is True and p_data.get("id") == "chat":
|
|
os.remove(chat_path)
|
|
logger.info("Removed deprecated built-in chat.json (merged into ask)")
|
|
except Exception:
|
|
logger.exception("Failed to inspect chat.json during migration")
|
|
for builtin in BUILTIN_MODES:
|
|
path = os.path.join(DATA_DIR, f"{builtin.id}.json")
|
|
if not os.path.exists(path):
|
|
save(builtin)
|
|
yield
|
|
|
|
|
|
modes = SubApp("modes", modes_lifespan)
|
|
|
|
|
|
def load_all() -> list[Mode]:
|
|
result = []
|
|
if not os.path.exists(DATA_DIR):
|
|
return result
|
|
for fname in os.listdir(DATA_DIR):
|
|
if fname.endswith(".json"):
|
|
data = read_json_or_none(os.path.join(DATA_DIR, fname))
|
|
if data is None:
|
|
continue
|
|
try:
|
|
result.append(Mode(**data))
|
|
except Exception as e:
|
|
logger.warning("Skipping invalid mode file %s: %s", fname, e)
|
|
return result
|
|
|
|
|
|
def save(mode: Mode):
|
|
atomic_write_json(os.path.join(DATA_DIR, f"{mode.id}.json"), mode.model_dump())
|
|
|
|
|
|
def load(mode_id: str) -> Mode:
|
|
data = read_json_or_none(os.path.join(DATA_DIR, f"{mode_id}.json"))
|
|
if data is None:
|
|
raise HTTPException(status_code=404, detail="Mode not found")
|
|
return Mode(**data)
|
|
|
|
|
|
def load_mode(mode_id: str) -> Mode | None:
|
|
"""Public helper for other modules to resolve a mode by ID."""
|
|
data = read_json_or_none(os.path.join(DATA_DIR, f"{mode_id}.json"))
|
|
return Mode(**data) if data is not None else None
|
|
|
|
|
|
@modes.router.get("/list")
|
|
async def list_modes():
|
|
builtin_defaults = {m.id: m.model_dump() for m in BUILTIN_MODES}
|
|
return {"modes": [m.model_dump() for m in load_all()], "builtin_defaults": builtin_defaults}
|
|
|
|
|
|
@modes.router.get("/{mode_id}")
|
|
async def get_mode(mode_id: str):
|
|
return load(mode_id).model_dump()
|
|
|
|
|
|
@modes.router.post("/create")
|
|
async def create_mode(body: ModeCreate):
|
|
mode = Mode(
|
|
name=body.name,
|
|
description=body.description,
|
|
system_prompt=body.system_prompt,
|
|
tools=body.tools,
|
|
default_next_mode=body.default_next_mode,
|
|
icon=body.icon,
|
|
color=body.color,
|
|
default_folder=body.default_folder,
|
|
is_builtin=False,
|
|
)
|
|
save(mode)
|
|
return {"ok": True, "mode": mode.model_dump()}
|
|
|
|
|
|
@modes.router.put("/{mode_id}")
|
|
async def update_mode(mode_id: str, body: ModeUpdate):
|
|
mode = load(mode_id)
|
|
for k, v in body.model_dump(exclude_unset=True).items():
|
|
setattr(mode, k, v)
|
|
save(mode)
|
|
return {"ok": True, "mode": mode.model_dump()}
|
|
|
|
|
|
@modes.router.post("/{mode_id}/reset")
|
|
async def reset_mode(mode_id: str):
|
|
"""Reset a built-in mode to its hardcoded defaults."""
|
|
builtin = next((m for m in BUILTIN_MODES if m.id == mode_id), None)
|
|
if not builtin:
|
|
raise HTTPException(status_code=400, detail="Only built-in modes can be reset")
|
|
save(builtin)
|
|
return {"ok": True, "mode": builtin.model_dump()}
|
|
|
|
|
|
@modes.router.delete("/{mode_id}")
|
|
async def delete_mode(mode_id: str):
|
|
mode = load(mode_id)
|
|
if mode.is_builtin:
|
|
raise HTTPException(status_code=403, detail="Cannot delete built-in modes")
|
|
path = os.path.join(DATA_DIR, f"{mode_id}.json")
|
|
if os.path.exists(path):
|
|
os.remove(path)
|
|
return {"ok": True}
|