enderofwings/NexusOS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(synapse): backend updates, add icons module, relocate playbooks to data/playbooks

Browse Source 
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
jon
2026-07-11 07:25:08 -05:00
parent 3ec57f9140
commit f4f77c5196
24 changed files with 1820 additions and 288 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/main.py
Regular → Executable
+415 -115
View File
@@ -4,46 +4,169 @@ from __future__ import annotations
import asyncio as _asyncio
import json as _json
import uuid as _uuid
from collections import Counter as _Counter
from typing import Any, AsyncGenerator, Dict, List, Optional, Tuple
from uuid import UUID
import httpx
import os as _os
from pathlib import Path
from fastapi import FastAPI, HTTPException, Body
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from fastapi.responses import StreamingResponse, FileResponse
from .nexus_config import settings
from .chat import generate_chat_response, stream_chat_response
from .nexus_config import settings, VERSION
from .chat import generate_chat_response, stream_chat_response, _synapse_trace
from .ollama_manager import initialize_ollama, initialize_ollama_async, get_ollama_manager
from .playbook_manager import PlaybookManager
def _render_memory_block(facts) -> str:
"""Render memory items as grouped ## Section / - bullet markdown."""
"""Render memory items as grouped ## Section / - bullet markdown.
Leading-space indent on a fact's text is preserved so nested bullets stay nested.
Behavioural "Instructions" entries are skipped — they belong in the playbook,
not the "what you know about Jon" facts block. Injecting imperative directives
("provide full rewrites", "include file paths") as facts pushes a small model
to reformat/organise the user's input instead of conversing."""
from collections import defaultdict
sections: dict = defaultdict(list)
for item in facts:
if (item.section or "").strip().lower() == "instructions":
continue
sections[item.section or "General"].append(item.text)
parts = []
for section, lines in sections.items():
parts.append(f"## {section}\n" + "\n".join(f" - {line}" for line in lines))
rendered = []
for line in lines:
stripped = line.lstrip(" ")
indent = len(line) - len(stripped)
rendered.append(" " * indent + f" - {stripped}")
parts.append(f"## {section}\n" + "\n".join(rendered))
return "\n\n".join(parts)
async def _auto_select_model() -> str:
# Preamble wrapped around the injected memory facts. It stays anti-recite for
# everyday chat, but explicitly permits surfacing facts when Jon asks about
# himself / to be quizzed — the old absolute "do NOT acknowledge these facts"
# made small models play dumb on exactly that request.
_MEMORY_PREAMBLE = (
"\n\n---\nBackground on Jon — use this to personalize your replies. Don't dump "
"or recite these facts unprompted, but when Jon asks about himself or asks you "
"to recall or quiz what you know, use them directly and specifically:\n\n"
)
_CODING_KEYWORDS = frozenset({
"code", "coding", "function", "class", "method", "variable", "bug", "error",
"debug", "fix", "refactor", "script", "program", "syntax", "compile", "import",
"module", "library", "algorithm", "loop", "array", "string", "integer", "boolean",
"return", "def", "const", "let", "var", "test", "api", "endpoint", "database",
"query", "sql", "bash", "terminal", "command", "package", "dependency",
"python", "javascript", "typescript", "rust", "golang", "java", "html", "css",
".py", ".js", ".ts", ".jsx", ".tsx", ".sh", ".json", ".yaml", ".sql", ".css",
})
def _detect_intent(message: str) -> str:
lower = message.lower()
return "code" if any(kw in lower for kw in _CODING_KEYWORDS) else "chat"
import re as _re
def _route_playbooks(message: str, candidates: list) -> list:
"""Return the reference playbook(s) whose tags appear directly in the user's
message. Returns [] when nothing matches, so casual chat doesn't drag in a
specialist playbook.
Dropped an old memory-fallback tier that, when the message matched no tags,
scored playbook tags against the user's WHOLE memory corpus. Because memory
permanently mentions e.g. the home network, that injected the Home Network
playbook (~1.7k chars) into unrelated chats. Direct references like "my truck"
already match here ('truck' is a Ford tag), so the fallback was mostly noise.
"""
if not candidates or not message:
return candidates
msg_tokens = set(_re.findall(r'\b\w+\b', message.lower()))
scores = [(sum(1 for tag in pb.tags if tag.lower() in msg_tokens), pb) for pb in candidates]
best = max(s for s, _ in scores)
if best > 0:
return [pb for s, pb in scores if s == best]
return []
async def _auto_select_model(message: str = "") -> str:
"""A pinned settings.model wins; otherwise pick the preferred installed model
for the detected intent. The preference lists live in one place now —
ollama_manager._MODEL_PREFERENCE, via select_best_model(intent)."""
try:
s = store.get_settings()
if s.get("model"):
return s["model"]
return await get_ollama_manager().select_best_model()
intent = _detect_intent(message) if message else "chat"
return await get_ollama_manager().select_best_model(intent)
except Exception:
return getattr(settings, "default_model", None) or "mistral"
_TITLE_SYSTEM_PROMPT = (
"You generate a short, descriptive title for a chat conversation based on the "
"user's first message. Reply with ONLY the title: 3 to 6 words, no quotes, no "
"trailing punctuation, no preamble. Use plain text in title case. The title "
"names the TOPIC — never echo the user's question or phrase it as a question.\n\n"
"Examples:\n"
"Message: can you help me fix a bug in my python script?\n"
"Title: Python Script Bug Fix\n"
"Message: what's a good recipe for sourdough bread?\n"
"Title: Sourdough Bread Recipe\n"
"Message: i want to try out your memory, ask me questions about myself\n"
"Title: Testing Memory Recall"
)
async def _generate_conversation_title(first_message: str, model: str) -> Optional[str]:
"""Ask the LLM for a concise title. Titling is a background 'curation' task,
so it runs on the CURATOR model (mistral) rather than the chat model: the 7B
follows the terse title format better than the 3B chat model, and it's already
warm in RAM. Crucially we pass the curator's own num_gpu (CPU) so we hit that
warm CPU-resident instance — same memory pool, no reload, and the GPU chat
model is never disturbed. `model` is only a fallback if no curator is set.
Best-effort: returns None on any failure so titling never breaks the chat."""
snippet = first_message.strip()[:1000]
if not snippet:
return None
try:
s = store.get_settings()
title_model = s.get("memory_model") or model
num_gpu = await get_ollama_manager().resolve_num_gpu(s.get("memory_gpu_offload", 0), title_model)
result = await generate_chat_response(
user_message=snippet,
metadata={"system": _TITLE_SYSTEM_PROMPT, "model": title_model,
"temperature": 0.2, "num_gpu": num_gpu},
timeout=30,
)
title = (result.get("response") or "").strip()
# Strip stray quotes/wrapping the model sometimes adds, collapse whitespace.
title = title.strip().strip('"').strip("'").splitlines()[0].strip()
# Drop a "Title:" prefix the model may echo from the examples, and strip
# trailing punctuation the prompt forbids but small models still add.
if title.lower().startswith("title:"):
title = title[len("title:"):].strip()
title = " ".join(title.split()).rstrip("?.!,;:")
if not title:
return None
return title[:120]
except Exception:
return None
from .memory.store import store, MemoryItem
from .playbooks.store import playbook_store, PlaybookItem
from .search import needs_web_search, web_search
MEMORY_SERVICE = "http://localhost:8001"
app = FastAPI(title="Synapse Backend", version="1.0")
app = FastAPI(title="Synapse Backend", version=VERSION)
# Alias for startup scripts
sio_app = app
@@ -73,6 +196,10 @@ async def startup_event():
global ollama
try:
ollama = await initialize_ollama_async()
# Keep the model resident per the persisted setting, then preload the
# model the first chat would pick so that message doesn't pay a cold load.
ollama.keep_alive = store.get_settings().get("keep_alive") or ollama.keep_alive
_asyncio.create_task(ollama.warm(await _auto_select_model()))
print("[Synapse] Ollama service is running.")
except Exception as e:
# Start in degraded mode — chat endpoints will return errors until Ollama
@@ -92,44 +219,7 @@ async def root():
status = ollama.get_status() if (ollama is not None and hasattr(ollama, "get_status")) else None
except Exception:
status = None
return {"status": "online", "ollama": status}
# -------------------------
# Chat (non-streaming)
# -------------------------
@app.post("/chat")
async def chat_endpoint(payload: Dict[str, Any]):
try:
message = payload.get("message", "")
app_settings = store.get_settings()
model = payload.get("model") or await _auto_select_model()
context = payload.get("context", {})
history = payload.get("history", [])
temperature = payload.get("temperature", app_settings.get("temperature"))
if not message:
raise HTTPException(status_code=400, detail="Missing 'message'")
rendered_message = playbooks.render_prompt(message)
system_prompt = playbooks.get_system_prompt() or app_settings.get("system_prompt", "")
memory_facts = store.all()
if memory_facts:
facts_block = _render_memory_block(memory_facts)
system_prompt = (system_prompt + "\n\n---\nWhat you know about Jon:\n\n" + facts_block) if system_prompt else facts_block
metadata: Dict[str, Any] = {"model": model, "context": context, "system": system_prompt, "temperature": temperature}
result = await generate_chat_response(
user_message=rendered_message, metadata=metadata, history=history
)
return {"response": result.get("response", ""), "model": model}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
return {"status": "online", "version": VERSION, "ollama": status}
# -------------------------
@@ -140,20 +230,25 @@ async def chat_stream_endpoint(payload: Dict[str, Any]):
try:
message = payload.get("message", "")
app_settings = store.get_settings()
model = payload.get("model") or await _auto_select_model()
model = payload.get("model") or await _auto_select_model(message)
context = payload.get("context", {})
conversation_id = payload.get("conversation_id") or str(_uuid.uuid4())
history = payload.get("history", [])
temperature = payload.get("temperature", app_settings.get("temperature"))
gpu_offload = payload.get("gpu_offload", app_settings.get("gpu_offload", -1))
num_gpu = await get_ollama_manager().resolve_num_gpu(gpu_offload, model)
if not message:
raise HTTPException(status_code=400, detail="Missing 'message'")
rendered_message = playbooks.render_prompt(message)
rendered_message = message # chat has no template vars; render_prompt is for the playbook path
system_prompt = playbooks.get_system_prompt() or app_settings.get("system_prompt", "")
# Append reference playbooks to the system prompt
context_pbs = playbooks.get_context_playbooks()
# Fetch memory facts once — used for both playbook routing and system prompt injection
memory_facts = store.all()
# Append the best-matching reference playbook(s) to the system prompt
context_pbs = _route_playbooks(rendered_message, playbooks.get_context_playbooks())
if context_pbs:
refs = "\n\n".join(
f"### {pb.title}\nGoal: {pb.goal}\n\n{pb.instructions}"
@@ -161,16 +256,17 @@ async def chat_stream_endpoint(payload: Dict[str, Any]):
)
separator = "\n\n---\nReference playbooks (read these as additional context):\n\n"
system_prompt = (system_prompt + separator + refs) if system_prompt else refs
# Inject persistent memory facts about Jon
memory_facts = store.all()
if memory_facts:
facts_block = _render_memory_block(memory_facts)
system_prompt = (system_prompt + "\n\n---\nWhat you know about Jon:\n\n" + facts_block) if system_prompt else facts_block
system_prompt = (system_prompt + _MEMORY_PREAMBLE + facts_block) if system_prompt else facts_block
# Search past conversations for relevant context and inject the top matches.
# This gives the model memory of prior exchanges without requiring tool-calling support.
past_context = store.search_conversations(message, limit=2)
# Semantic recall (embeddings) finds relevant exchanges even without shared
# keywords; it falls back to lexical substring match if embeddings are down.
past_context = await store.semantic_search_conversations(
message, get_ollama_manager().embed, limit=2
)
if past_context:
snippets = []
for conv in past_context:
@@ -183,7 +279,65 @@ async def chat_stream_endpoint(payload: Dict[str, Any]):
separator = "\n\n---\nRelevant past exchanges (use as background context only):\n\n"
system_prompt = (system_prompt + separator + memory_block) if system_prompt else memory_block
metadata: Dict[str, Any] = {"model": model, "context": context, "system": system_prompt, "temperature": temperature}
# Fetch web search results for time-sensitive queries
search_results = ""
if needs_web_search(message):
search_results = await _asyncio.to_thread(web_search, message)
if search_results:
separator = "\n\n---\nWeb search results (treat as current information):\n\n"
system_prompt = (system_prompt + separator + search_results) if system_prompt else search_results
# ── MindTrace pre-flight ──────────────────────────────────────────
_trace_intent = _detect_intent(message) if message else "chat"
if payload.get("model"):
_trace_src = "user-override"
elif store.get_settings().get("model"):
_trace_src = "settings"
else:
_trace_src = f"auto/{_trace_intent}"
_synapse_trace(f"\n{'═' * 55}\n")
_synapse_trace(f"▶ MODEL : {model} [{_trace_src}]\n")
if _trace_intent == "code":
_kws = [kw for kw in _CODING_KEYWORDS if kw in message.lower()][:5]
_synapse_trace(f" INTENT: code → {', '.join(_kws)}\n")
else:
_synapse_trace(f" INTENT: chat\n")
_main_pb = playbooks.get_main_playbook()
if _main_pb:
_synapse_trace(f" PLAYBOOK: {_main_pb.title}\n")
if _main_pb.goal:
_synapse_trace(f" goal: {_main_pb.goal[:100]}\n")
else:
_synapse_trace(f" PLAYBOOK: none\n")
if context_pbs:
_synapse_trace(f" ROUTED : {', '.join(pb.title for pb in context_pbs)}\n")
else:
_synapse_trace(f" ROUTED : none (no tag match)\n")
if search_results:
_synapse_trace(f" SEARCH : {len(search_results)} chars injected\n")
elif needs_web_search(message):
_synapse_trace(f" SEARCH : triggered but returned no results\n")
if memory_facts:
_secs = _Counter(f.section or "General" for f in memory_facts)
_sec_str = " ".join(f"{s}({n})" for s, n in _secs.items())
_synapse_trace(f" MEMORY : {len(memory_facts)} facts [{_sec_str}]\n")
else:
_synapse_trace(f" MEMORY : none\n")
if past_context:
_synapse_trace(f" CONTEXT : {len(past_context)} past conversation match(es) injected\n")
_synapse_trace(f" SYS LEN : {len(system_prompt)} chars\n")
_synapse_trace(f"{'─' * 55}\n")
# ── end MindTrace pre-flight ──────────────────────────────────────
metadata: Dict[str, Any] = {"model": model, "context": context, "system": system_prompt, "temperature": temperature, "num_gpu": num_gpu}
# Persist conversation and user message before streaming
store.create_conversation(conversation_id)
@@ -192,6 +346,9 @@ async def chat_stream_endpoint(payload: Dict[str, Any]):
async def event_stream() -> AsyncGenerator[str, None]:
response_chunks: list[str] = []
meta: dict = {}
final_model = model
# ── Phase 1: stream primary model response ────────────────────
try:
async for chunk in stream_chat_response(
user_message=rendered_message,
@@ -206,22 +363,51 @@ async def chat_stream_endpoint(payload: Dict[str, Any]):
yield f"event: meta\ndata: {chunk[8:]}\n\n"
continue
response_chunks.append(chunk)
yield f"data: {chunk}\n\n"
# Persist the completed assistant response with model and token stats
if response_chunks:
store.add_message(
conversation_id, "assistant", "".join(response_chunks),
model=meta.get("model") or model,
tokens=meta.get("tokens"),
)
except Exception as e:
yield f"event: error\ndata: {_json.dumps({'detail': str(e)})}\n\n"
yield f"data: {_json.dumps(chunk)}\n\n"
except _asyncio.TimeoutError:
_tval = store.get_settings().get("timeout", 120)
yield f"event: error\ndata: {_json.dumps({'detail': f'Model timed out after {_tval}s — try a smaller/faster model'})}\n\n"
return
except Exception as e:
detail = str(e) or type(e).__name__
yield f"event: error\ndata: {_json.dumps({'detail': detail})}\n\n"
return
# ── Persist completed response ───────────────────────────────
if response_chunks:
store.add_message(
conversation_id, "assistant", "".join(response_chunks),
model=meta.get("model") or final_model,
tokens=meta.get("tokens"),
)
# Response is complete — let the client re-enable its input now,
# so the slow title/memory work below doesn't freeze the UI.
yield "event: done\ndata: {}\n\n"
# Generate an AI title from the opening message. Retried on any turn
# while still untitled, so an interrupted first stream can recover.
if response_chunks:
try:
conv = store.get_conversation(conversation_id)
if conv and not conv.title:
first_user = next(
(m.content for m in conv.messages if m.role == "user"),
rendered_message,
)
title = await _generate_conversation_title(
first_user, meta.get("model") or final_model
)
if title:
store.set_conversation_title(conversation_id, title)
yield f"event: title\ndata: {_json.dumps({'title': title})}\n\n"
except Exception:
pass
# Ask the memory service curator to evaluate this exchange
if response_chunks:
try:
async with httpx.AsyncClient(timeout=25.0) as _mc:
async with httpx.AsyncClient(timeout=310.0) as _mc:
r = await _mc.post(
f"{MEMORY_SERVICE}/memories/extract",
json={
@@ -231,11 +417,11 @@ async def chat_stream_endpoint(payload: Dict[str, Any]):
)
if r.status_code == 200:
data = r.json()
if data.get("saved"):
mem_result = {"section": data["section"], "text": data["text"]}
for it in data.get("items", []):
mem_result = {"section": it["section"], "text": it["text"]}
yield f"event: memory\ndata: {_json.dumps(mem_result)}\n\n"
except Exception:
pass
except Exception as e:
_synapse_trace(f"\n⚠ memory extraction call failed: {e}\n")
return StreamingResponse(event_stream(), media_type="text/event-stream")
@@ -244,24 +430,6 @@ async def chat_stream_endpoint(payload: Dict[str, Any]):
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# -------------------------
# Playbook Execution
# -------------------------
@app.post("/playbook/run")
async def run_playbook(payload: Dict[str, Any]):
name = payload.get("name")
variables = payload.get("variables", {})
if not name:
raise HTTPException(status_code=400, detail="Missing playbook name")
try:
result = playbooks.render_prompt(name, variables=variables)
return {"result": result}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# -------------------------
# Settings
# -------------------------
@@ -273,7 +441,10 @@ async def get_settings_endpoint():
@app.put("/settings")
async def put_settings_endpoint(payload: Dict[str, Any] = Body(...)):
store.update_settings(payload)
return store.get_settings()
merged = store.get_settings()
if "keep_alive" in payload:
get_ollama_manager().keep_alive = merged.get("keep_alive") or None
return merged
# -------------------------
@@ -286,8 +457,8 @@ async def get_memory():
@app.post("/memory")
async def add_memory(payload: Dict[str, Any] = Body(...)):
text = (payload.get("text") or "").strip()
if not text:
text = (payload.get("text") or "").rstrip()
if not text.strip():
raise HTTPException(status_code=400, detail="Missing 'text'")
from .memory.store import MemoryItem
import uuid as _mem_uuid
@@ -310,7 +481,7 @@ async def update_memory(item_id: str, payload: Dict[str, Any] = Body(...)):
updated = MemoryItem(
id=item_id,
section=(payload.get("section") or existing.section or "General").strip(),
text=(payload.get("text") or existing.text).strip(),
text=(payload.get("text") or existing.text).rstrip(),
tags=payload.get("tags", existing.tags),
)
store.update(updated)
@@ -325,6 +496,16 @@ async def delete_memory(item_id: str):
return {"status": "deleted"}
@app.post("/memory/reorder")
async def reorder_memory(payload: Dict[str, Any] = Body(...)):
section = (payload.get("section") or "General").strip() or "General"
ids = payload.get("ids")
if not isinstance(ids, list) or not all(isinstance(x, str) for x in ids):
raise HTTPException(status_code=400, detail="ids must be a list of strings")
ok = store.reorder_section(section, ids)
return {"ok": ok}
# -------------------------
# Models
# -------------------------
@@ -332,8 +513,11 @@ async def delete_memory(item_id: str):
async def get_models():
try:
mgr = get_ollama_manager()
models = await mgr.list_models()
selected = await mgr.select_best_model()
# Embedding models (e.g. nomic-embed-text) can't chat — hide from picker.
models = [m for m in await mgr.list_models() if "embed" not in m.lower()]
# Report the SAME model the chat path would auto-pick (honors a pin),
# so the picker's "Auto (…)" label matches what actually answers.
selected = await _auto_select_model()
return {"models": models, "selected": selected}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@@ -641,26 +825,6 @@ async def delete_playbook_endpoint(id: UUID):
# -------------------------
# Unified Search
# -------------------------
@app.get("/search")
async def search_endpoint(q: Optional[str] = None):
"""Search conversations and playbooks by keyword."""
if not q or not q.strip():
raise HTTPException(status_code=400, detail="Missing query parameter 'q'")
try:
conversations = store.search_conversations(q, limit=5)
playbook_hits = playbook_store.search_playbooks(q)
return {
"query": q,
"conversations": conversations,
"playbooks": [
{"id": p.id, "title": p.title, "goal": p.goal, "tags": p.tags}
for p in playbook_hits
],
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# -------------------------
# Conversations
# -------------------------
@@ -682,6 +846,7 @@ async def get_conversations(q: Optional[str] = None):
"timestamp": c.created_at,
"updated_at": c.updated_at,
"preview": c.preview,
"title": c.title,
}
for c in conversations
]
@@ -690,6 +855,47 @@ async def get_conversations(q: Optional[str] = None):
raise HTTPException(status_code=500, detail=str(e))
@app.get("/conversations/export")
async def export_conversations(min_turns: int = 1):
"""Export conversations as ShareGPT JSONL for fine-tuning.
Each line is one conversation:
{"conversations": [{"from": "human", "value": "..."}, {"from": "gpt", "value": "..."}]}
Query params:
min_turns — minimum user/assistant exchanges to include (default 1)
"""
from fastapi.responses import Response
import datetime
conversations = store.all_conversations()
lines = []
for conv in conversations:
msgs = [m for m in conv.messages if m.role in ("user", "assistant")]
if len(msgs) < 2:
continue
if sum(1 for m in msgs if m.role == "user") < min_turns:
continue
sharegpt_msgs = [
{"from": "human" if m.role == "user" else "gpt", "value": m.content}
for m in msgs
]
lines.append(_json.dumps({"conversations": sharegpt_msgs}))
date_str = datetime.date.today().isoformat()
filename = f"nexus-conversations-{date_str}.jsonl"
return Response(
content="\n".join(lines),
media_type="application/x-ndjson",
headers={
"Content-Disposition": f"attachment; filename={filename}",
"X-Exported-Count": str(len(lines)),
},
)
@app.get("/conversations/{conversation_id}")
async def get_conversation(conversation_id: str):
try:
@@ -699,6 +905,7 @@ async def get_conversation(conversation_id: str):
return {
"id": conv.id,
"timestamp": conv.created_at,
"title": conv.title,
"messages": [
{"role": m.role, "content": m.content, "timestamp": m.timestamp, "model": m.model, "tokens": m.tokens}
for m in conv.messages
@@ -710,6 +917,25 @@ async def get_conversation(conversation_id: str):
raise HTTPException(status_code=500, detail=str(e))
@app.patch("/conversations/{conversation_id}")
async def rename_conversation(conversation_id: str, payload: Dict[str, Any] = Body(...)):
"""Manually set a conversation's title."""
try:
conv = store.get_conversation(conversation_id)
if not conv:
raise HTTPException(status_code=404, detail="Conversation not found")
title = (payload.get("title") or "").strip()
if not title:
raise HTTPException(status_code=400, detail="Missing 'title'")
title = " ".join(title.split())[:120]
store.set_conversation_title(conversation_id, title)
return {"id": conversation_id, "title": title}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.delete("/conversations/{conversation_id}")
async def delete_conversation(conversation_id: str):
try:
@@ -723,6 +949,80 @@ async def delete_conversation(conversation_id: str):
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# ── Icon branding routes ──────────────────────────────────────────────────────
_REPO_ASSETS = str(Path(__file__).resolve().parents[1] / "assets")
_ALLOWED_ICON_ROOTS = [
"/usr/share/icons",
"/usr/share/pixmaps",
"/usr/local/share/icons",
"/opt",
_os.path.expanduser("~/.local/share/icons"),
_os.path.expanduser("~/.icons"),
_REPO_ASSETS,
]
@app.get("/icons/apps")
async def list_icon_apps():
"""Return all installed applications with their icon paths."""
try:
from .icons.resolver import scan_apps
loop = _asyncio.get_event_loop()
apps = await loop.run_in_executor(None, scan_apps)
return {"apps": apps}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/icons/image")
async def get_icon_image(path: str):
"""Serve an icon file after verifying it's in an allowed root."""
real = _os.path.realpath(path)
if not any(real.startswith(r) for r in _ALLOWED_ICON_ROOTS):
raise HTTPException(status_code=403, detail="Path not allowed")
if not _os.path.isfile(real):
raise HTTPException(status_code=404, detail="Icon not found")
return FileResponse(real)
@app.post("/icons/brand")
async def brand_app_icon(payload: Dict[str, Any] = Body(...)):
"""Composite an app icon onto the NexusOS underlay tile."""
src_path = payload.get("src_path", "")
output_name = payload.get("output_name", "")
if not src_path or not output_name:
raise HTTPException(status_code=400, detail="src_path and output_name required")
frac = float(payload.get("frac", 0.60))
round_mask = bool(payload.get("round_mask", False))
nexus_ring = bool(payload.get("nexus_ring", False))
reload = bool(payload.get("reload", True))
category = str(payload.get("category", "apps"))
try:
from .icons.compositor import brand_icon
loop = _asyncio.get_event_loop()
await loop.run_in_executor(
None,
lambda: brand_icon(src_path, output_name, frac, round_mask, nexus_ring, reload, category),
)
return {"status": "ok", "output_name": output_name}
except ValueError as e:
raise HTTPException(status_code=403, detail=str(e))
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/icons/apply")
async def apply_icon_cache_route():
"""Rebuild the GTK icon cache and reload the panel."""
try:
from .icons.compositor import apply_icon_cache
loop = _asyncio.get_event_loop()
await loop.run_in_executor(None, apply_icon_cache)
return {"status": "ok"}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# -------------------------
# End of file
# -------------------------