jon
1029 lines
40 KiB
Python
Executable File
1029 lines
40 KiB
Python
Executable File
# main.py
|
|
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, FileResponse
|
|
|
|
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.
|
|
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():
|
|
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)
|
|
|
|
|
|
# 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"]
|
|
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=VERSION)
|
|
|
|
# Alias for startup scripts
|
|
sio_app = app
|
|
|
|
# --- CORS ---
|
|
# allow_credentials=True is incompatible with allow_origins=["*"] — browsers
|
|
# reject such responses. Since this is a local-only service with no cookies/auth,
|
|
# wildcard origins without credentials is correct.
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=False,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# --- GLOBALS ---
|
|
playbooks = PlaybookManager()
|
|
ollama = None
|
|
|
|
|
|
# -------------------------
|
|
# Startup: Initialize Ollama service
|
|
# -------------------------
|
|
@app.on_event("startup")
|
|
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
|
|
# is brought up, but the backend itself stays alive so the control panel
|
|
# and other endpoints remain reachable.
|
|
print(f"[Synapse] WARNING: Ollama unavailable at startup: {e}")
|
|
from .ollama_manager import OllamaManager
|
|
ollama = OllamaManager()
|
|
|
|
|
|
# -------------------------
|
|
# Root
|
|
# -------------------------
|
|
@app.get("/")
|
|
async def root():
|
|
try:
|
|
status = ollama.get_status() if (ollama is not None and hasattr(ollama, "get_status")) else None
|
|
except Exception:
|
|
status = None
|
|
return {"status": "online", "version": VERSION, "ollama": status}
|
|
|
|
|
|
# -------------------------
|
|
# Chat (streaming)
|
|
# -------------------------
|
|
@app.post("/chat/stream")
|
|
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(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 = 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", "")
|
|
|
|
# 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}"
|
|
for pb in context_pbs
|
|
)
|
|
separator = "\n\n---\nReference playbooks (read these as additional context):\n\n"
|
|
system_prompt = (system_prompt + separator + refs) if system_prompt else refs
|
|
if memory_facts:
|
|
facts_block = _render_memory_block(memory_facts)
|
|
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.
|
|
# 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:
|
|
exchange = "\n".join(
|
|
f" {m['role'].upper()}: {m['content']}"
|
|
for m in conv["matches"]
|
|
)
|
|
snippets.append(exchange)
|
|
memory_block = "\n\n".join(snippets)
|
|
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
|
|
|
|
# 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)
|
|
store.add_message(conversation_id, "user", rendered_message)
|
|
|
|
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,
|
|
metadata=metadata,
|
|
history=history,
|
|
):
|
|
if chunk.startswith("__meta__"):
|
|
try:
|
|
meta = _json.loads(chunk[8:])
|
|
except Exception:
|
|
pass
|
|
yield f"event: meta\ndata: {chunk[8:]}\n\n"
|
|
continue
|
|
response_chunks.append(chunk)
|
|
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=310.0) as _mc:
|
|
r = await _mc.post(
|
|
f"{MEMORY_SERVICE}/memories/extract",
|
|
json={
|
|
"user_message": rendered_message,
|
|
"assistant_response": "".join(response_chunks),
|
|
},
|
|
)
|
|
if r.status_code == 200:
|
|
data = r.json()
|
|
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 as e:
|
|
_synapse_trace(f"\n⚠ memory extraction call failed: {e}\n")
|
|
|
|
return StreamingResponse(event_stream(), media_type="text/event-stream")
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
# -------------------------
|
|
# Settings
|
|
# -------------------------
|
|
@app.get("/settings")
|
|
async def get_settings_endpoint():
|
|
return store.get_settings()
|
|
|
|
|
|
@app.put("/settings")
|
|
async def put_settings_endpoint(payload: Dict[str, Any] = Body(...)):
|
|
store.update_settings(payload)
|
|
merged = store.get_settings()
|
|
if "keep_alive" in payload:
|
|
get_ollama_manager().keep_alive = merged.get("keep_alive") or None
|
|
return merged
|
|
|
|
|
|
# -------------------------
|
|
# Memory
|
|
# -------------------------
|
|
@app.get("/memory")
|
|
async def get_memory():
|
|
return {"items": [{"id": m.id, "section": m.section, "text": m.text, "tags": m.tags} for m in store.all()]}
|
|
|
|
|
|
@app.post("/memory")
|
|
async def add_memory(payload: Dict[str, Any] = Body(...)):
|
|
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
|
|
item = MemoryItem(
|
|
id=str(_mem_uuid.uuid4()),
|
|
section=(payload.get("section") or "General").strip(),
|
|
text=text,
|
|
tags=payload.get("tags", []),
|
|
)
|
|
store.add(item)
|
|
return {"id": item.id, "section": item.section, "text": item.text, "tags": item.tags}
|
|
|
|
|
|
@app.patch("/memory/{item_id}")
|
|
async def update_memory(item_id: str, payload: Dict[str, Any] = Body(...)):
|
|
existing = store.get(item_id)
|
|
if not existing:
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
from .memory.store import MemoryItem
|
|
updated = MemoryItem(
|
|
id=item_id,
|
|
section=(payload.get("section") or existing.section or "General").strip(),
|
|
text=(payload.get("text") or existing.text).rstrip(),
|
|
tags=payload.get("tags", existing.tags),
|
|
)
|
|
store.update(updated)
|
|
return {"id": updated.id, "section": updated.section, "text": updated.text, "tags": updated.tags}
|
|
|
|
|
|
@app.delete("/memory/{item_id}")
|
|
async def delete_memory(item_id: str):
|
|
if not store.get(item_id):
|
|
raise HTTPException(status_code=404, detail="Not found")
|
|
store.delete(item_id)
|
|
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
|
|
# -------------------------
|
|
@app.get("/models")
|
|
async def get_models():
|
|
try:
|
|
mgr = get_ollama_manager()
|
|
# 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))
|
|
|
|
|
|
@app.get("/models/details")
|
|
async def get_model_details():
|
|
"""Return full model objects (name, size, modified_at) from Ollama."""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=5.0) as client:
|
|
r = await client.get(f"{settings.ollama_host.rstrip('/')}/api/tags")
|
|
r.raise_for_status()
|
|
return r.json()
|
|
except Exception as e:
|
|
raise HTTPException(status_code=502, detail=f"Ollama unreachable: {e}")
|
|
|
|
|
|
@app.post("/models/pull")
|
|
async def pull_model(payload: Dict[str, Any] = Body(...)):
|
|
"""Proxy a streaming model pull from Ollama."""
|
|
name = payload.get("name", "").strip()
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="Missing model name")
|
|
|
|
async def _stream():
|
|
try:
|
|
async with httpx.AsyncClient(timeout=600.0) as client:
|
|
async with client.stream(
|
|
"POST", f"{settings.ollama_host.rstrip('/')}/api/pull",
|
|
json={"name": name},
|
|
) as resp:
|
|
async for line in resp.aiter_lines():
|
|
if line:
|
|
yield line + "\n"
|
|
except Exception as e:
|
|
yield f'{{"error": "{e}"}}\n'
|
|
finally:
|
|
# Bust the model cache so new model is visible immediately
|
|
get_ollama_manager().invalidate_model_cache()
|
|
|
|
return StreamingResponse(_stream(), media_type="application/x-ndjson")
|
|
|
|
|
|
@app.delete("/models/{name:path}")
|
|
async def delete_model(name: str):
|
|
"""Proxy a model deletion to Ollama."""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
r = await client.request(
|
|
"DELETE", f"{settings.ollama_host.rstrip('/')}/api/delete",
|
|
json={"name": name},
|
|
)
|
|
if not r.is_success:
|
|
raise HTTPException(status_code=r.status_code, detail=r.text)
|
|
get_ollama_manager().invalidate_model_cache()
|
|
return {"status": "deleted", "name": name}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=502, detail=f"Ollama unreachable: {e}")
|
|
|
|
|
|
# -------------------------
|
|
# Ollama Status
|
|
# -------------------------
|
|
@app.get("/ollama/status")
|
|
async def ollama_status_endpoint():
|
|
try:
|
|
if ollama is not None and hasattr(ollama, "get_status"):
|
|
status = ollama.get_status()
|
|
else:
|
|
status = None
|
|
return {"status": status}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
# -------------------------
|
|
# Ollama Start
|
|
# -------------------------
|
|
@app.post("/ollama/start")
|
|
async def ollama_start_endpoint():
|
|
try:
|
|
global ollama
|
|
if ollama is None:
|
|
ollama = initialize_ollama()
|
|
|
|
# Try to start if it has a start method
|
|
if hasattr(ollama, "start"):
|
|
ollama.start()
|
|
|
|
is_running = ollama.is_running() if hasattr(ollama, "is_running") else True
|
|
return {"status": "started" if is_running else "failed", "running": is_running}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
# -------------------------
|
|
# Ollama Stop
|
|
# -------------------------
|
|
@app.post("/ollama/stop")
|
|
async def ollama_stop_endpoint():
|
|
try:
|
|
global ollama
|
|
if ollama is not None and hasattr(ollama, "stop"):
|
|
ollama.stop()
|
|
|
|
is_running = ollama.is_running() if (ollama is not None and hasattr(ollama, "is_running")) else False
|
|
return {"status": "stopped" if not is_running else "still_running", "running": is_running}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
# -------------------------
|
|
# Playbooks List (existing)
|
|
# -------------------------
|
|
@app.get("/playbooks")
|
|
async def get_playbooks():
|
|
try:
|
|
playbook_list = playbook_store.all_playbooks()
|
|
return {
|
|
"playbooks": [
|
|
{
|
|
"id": p.id,
|
|
"title": p.title,
|
|
"goal": p.goal,
|
|
"instructions": getattr(p, "instructions", ""),
|
|
"tags": getattr(p, "tags", []),
|
|
}
|
|
for p in playbook_list
|
|
]
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
# -------------------------
|
|
# Playbook retrieve / update / replace (new)
|
|
# -------------------------
|
|
def _find_playbook_by_id(playbook_id: str) -> Tuple[Optional[Any], Optional[str]]:
|
|
"""
|
|
Try common store getters, then fall back to scanning store.all_playbooks().
|
|
Returns (playbook_obj, key) where key is the identifier used by store if applicable.
|
|
"""
|
|
try:
|
|
pb = playbook_store.get_playbook(playbook_id)
|
|
if pb:
|
|
return pb, playbook_id
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
for p in playbook_store.all_playbooks():
|
|
pid = str(getattr(p, "id", None) or "")
|
|
if pid == str(playbook_id):
|
|
return p, pid
|
|
except Exception:
|
|
pass
|
|
|
|
return None, None
|
|
|
|
|
|
def _persist_playbook(playbook_dict: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Persist a playbook dict to the store by converting to PlaybookItem.
|
|
"""
|
|
try:
|
|
# Preserve existing order on update; use provided order (or tail) on create
|
|
existing = playbook_store.get_playbook(str(playbook_dict["id"]))
|
|
order = existing.order if existing else playbook_dict.get("order", len(playbook_store.all_playbooks()))
|
|
|
|
playbook_item = PlaybookItem(
|
|
id=str(playbook_dict["id"]),
|
|
title=playbook_dict.get("title", ""),
|
|
goal=playbook_dict.get("goal", ""),
|
|
instructions=playbook_dict.get("instructions", ""),
|
|
tags=playbook_dict.get("tags", []),
|
|
order=order
|
|
)
|
|
|
|
playbook_store.add_playbook(playbook_item)
|
|
|
|
# Return as dict
|
|
return {
|
|
"id": playbook_item.id,
|
|
"title": playbook_item.title,
|
|
"goal": playbook_item.goal,
|
|
"instructions": playbook_item.instructions,
|
|
"tags": playbook_item.tags,
|
|
}
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"Failed to persist playbook: {str(e)}")
|
|
|
|
@app.post("/playbooks/reorder")
|
|
async def reorder_playbooks(payload: Dict[str, Any] = Body(...)):
|
|
"""
|
|
Accepts {"ids": ["id1", "id2", "id3"]} in the desired order.
|
|
"""
|
|
try:
|
|
ids = payload.get("ids", [])
|
|
if not ids:
|
|
raise HTTPException(status_code=400, detail="Missing ids list")
|
|
playbook_store.reorder_playbooks(ids)
|
|
return {"status": "reordered"}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.post("/playbooks")
|
|
async def create_playbook(payload: Dict[str, Any] = Body(...)):
|
|
try:
|
|
if not payload.get("title") or not payload.get("goal") or not payload.get("instructions"):
|
|
raise HTTPException(status_code=400, detail="Missing required fields: title, goal, instructions")
|
|
payload["id"] = str(_uuid.uuid4())
|
|
payload.setdefault("tags", [])
|
|
created = _persist_playbook(payload)
|
|
return created
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@app.get("/playbooks/{id}")
|
|
async def get_playbook(id: UUID):
|
|
try:
|
|
pb, _ = _find_playbook_by_id(str(id))
|
|
if not pb:
|
|
raise HTTPException(status_code=404, detail="Not Found")
|
|
|
|
# Build dict in the exact order you want
|
|
result = {
|
|
"id": getattr(pb, "id", None) or (pb.get("id") if isinstance(pb, dict) else None),
|
|
"title": getattr(pb, "title", None) or (pb.get("title") if isinstance(pb, dict) else None),
|
|
"goal": getattr(pb, "goal", None) or (pb.get("goal") if isinstance(pb, dict) else None),
|
|
"instructions": getattr(pb, "instructions", "") or (pb.get("instructions") if isinstance(pb, dict) else ""),
|
|
"tags": getattr(pb, "tags", []) or (pb.get("tags") if isinstance(pb, dict) else []),
|
|
}
|
|
return result
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.patch("/playbooks/{id}")
|
|
async def patch_playbook(id: UUID, payload: Dict[str, Any] = Body(...)):
|
|
"""
|
|
Partial update: accepts a JSON object with fields to update (e.g., {"instructions":"..."}).
|
|
"""
|
|
try:
|
|
pb, key = _find_playbook_by_id(str(id))
|
|
if not pb:
|
|
raise HTTPException(status_code=404, detail="Not Found")
|
|
|
|
# Normalize existing to dict
|
|
if isinstance(pb, dict):
|
|
existing = dict(pb)
|
|
else:
|
|
existing = {k: getattr(pb, k) for k in ("id", "title", "goal", "instructions", "tags") if hasattr(pb, k)}
|
|
|
|
merged = {**existing, **payload}
|
|
|
|
# Try to persist via store API
|
|
updated = _persist_playbook(merged)
|
|
if not isinstance(updated, dict):
|
|
return {k: getattr(updated, k) for k in ("id", "title", "goal", "instructions", "tags") if hasattr(updated, k)}
|
|
return updated
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.put("/playbooks/{id}")
|
|
async def put_playbook(id: UUID, payload: Dict[str, Any] = Body(...)):
|
|
"""
|
|
Full replace: replace the playbook with the provided payload (payload should include title, goal, instructions, tags).
|
|
"""
|
|
try:
|
|
pb, key = _find_playbook_by_id(str(id))
|
|
if not pb:
|
|
raise HTTPException(status_code=404, detail="Not Found")
|
|
|
|
payload["id"] = str(id)
|
|
|
|
replaced = _persist_playbook(payload)
|
|
if not isinstance(replaced, dict):
|
|
return {k: getattr(replaced, k) for k in ("id", "title", "goal", "instructions", "tags") if hasattr(replaced, k)}
|
|
return replaced
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
@app.delete("/playbooks/{id}")
|
|
async def delete_playbook_endpoint(id: UUID):
|
|
try:
|
|
pb, _ = _find_playbook_by_id(str(id))
|
|
if not pb:
|
|
raise HTTPException(status_code=404, detail="Not Found")
|
|
playbook_store.delete_playbook(str(id))
|
|
return {"status": "deleted"}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
# -------------------------
|
|
# Unified Search
|
|
# -------------------------
|
|
# -------------------------
|
|
# Conversations
|
|
# -------------------------
|
|
@app.get("/conversations")
|
|
async def get_conversations(q: Optional[str] = None):
|
|
try:
|
|
conversations = store.all_conversations()
|
|
if q:
|
|
q_lower = q.lower()
|
|
conversations = [
|
|
c for c in conversations
|
|
if any(q_lower in m.content.lower() for m in c.messages)
|
|
or q_lower in c.preview.lower()
|
|
]
|
|
return {
|
|
"conversations": [
|
|
{
|
|
"id": c.id,
|
|
"timestamp": c.created_at,
|
|
"updated_at": c.updated_at,
|
|
"preview": c.preview,
|
|
"title": c.title,
|
|
}
|
|
for c in conversations
|
|
]
|
|
}
|
|
except Exception as e:
|
|
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:
|
|
conv = store.get_conversation(conversation_id)
|
|
if not conv:
|
|
raise HTTPException(status_code=404, detail="Conversation not found")
|
|
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
|
|
]
|
|
}
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
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:
|
|
conv = store.get_conversation(conversation_id)
|
|
if not conv:
|
|
raise HTTPException(status_code=404, detail="Conversation not found")
|
|
store.delete_conversation(conversation_id)
|
|
return {"status": "deleted"}
|
|
except HTTPException:
|
|
raise
|
|
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
|
|
# -------------------------
|
GitNexus · part of the Nexus stack