AthenaandCursor
1f6beed0d2
Two tool/agent-layer hardening changes: * fetch_url now resolves the target host and refuses to connect if any resolved address is loopback, private (RFC1918/ULA), link-local (incl. the 169.254.169.254 cloud-metadata endpoint), multicast, reserved, or unspecified. IPv4-mapped IPv6 is unwrapped first, and the guard re-runs on every redirect hop so a public URL cannot 302 its way to an internal target. * /chat/approve now requires a single-use token minted when the stream pauses for approval and delivered only in that stream's tool_request event, compared in constant time. Previously the pending approval was keyed solely on a client-supplied conversation_id, so anyone who could enumerate a conversation_id could approve another client's pending action. The frontend threads the token from the tool_request event into the approve call. Co-authored-by: Cursor <cursoragent@cursor.com>
1392 lines
55 KiB
Python
1392 lines
55 KiB
Python
# main.py
|
|
from __future__ import annotations
|
|
|
|
import asyncio as _asyncio
|
|
import json as _json
|
|
import secrets as _secrets
|
|
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
|
|
import platform as _platform
|
|
from pathlib import Path
|
|
from fastapi import FastAPI, HTTPException, Body
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from starlette.middleware.trustedhost import TrustedHostMiddleware
|
|
from fastapi.responses import StreamingResponse, FileResponse
|
|
|
|
from .nexus_config import settings, VERSION, DEFAULT_CHAT_MODEL, ALLOWED_HOSTS, ALLOWED_ORIGINS
|
|
from .chat import generate_chat_response, stream_chat_response, _synapse_trace
|
|
from . import chat as _chat
|
|
from .ollama_manager import initialize_ollama, initialize_ollama_async, get_ollama_manager
|
|
from . import frontend_manager as _frontend_manager
|
|
from .playbook_manager import PlaybookManager
|
|
from . import tools as _tools
|
|
|
|
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 the user" 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)
|
|
|
|
|
|
# Header for the injected memory facts. No anti-recite restriction: NexusOS is a
|
|
# memory-first assistant, so facts are framed as freely usable knowledge about
|
|
# the user. (A minimal header is kept so the facts have context; drop it entirely
|
|
# to inject the raw facts block with no framing.)
|
|
_MEMORY_PREAMBLE = (
|
|
"\n\n---\nWhat you know about the user — use these facts freely and naturally to "
|
|
"inform and personalize your replies:\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"
|
|
# Auto-mode remap: a configured model for this intent fires first;
|
|
# otherwise fall back to the built-in preference list.
|
|
remap = s.get(f"auto_{intent}_model")
|
|
if remap:
|
|
return remap
|
|
return await get_ollama_manager().select_best_model(intent)
|
|
except Exception:
|
|
return DEFAULT_CHAT_MODEL
|
|
|
|
|
|
|
|
_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
|
|
|
|
# --- Local-access guard ---
|
|
# These APIs are unauthenticated and meant for this machine only. Two layers:
|
|
# 1. TrustedHostMiddleware rejects a foreign Host header, which is what defeats
|
|
# DNS-rebinding — the browser treats a rebound attacker domain as
|
|
# same-origin, so CORS alone can't stop it.
|
|
# 2. CORS is scoped to known local origins (not "*"), so a malicious page can't
|
|
# read responses cross-origin from the victim's browser.
|
|
# NEXUS_ALLOWED_HOSTS=* / a custom NEXUS_ALLOWED_ORIGINS opts into LAN exposure,
|
|
# and should only be used once real authentication is in front of these apps.
|
|
if "*" not in ALLOWED_HOSTS:
|
|
app.add_middleware(TrustedHostMiddleware, allowed_hosts=ALLOWED_HOSTS)
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=ALLOWED_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
|
|
# Manual AI control: build the manager but DON'T launch Ollama or warm a model.
|
|
# Backend serves logs/playbooks/memory/models with the AI off; the user brings
|
|
# it up from the UI (POST /ollama/start). get_ollama_manager() is the same
|
|
# singleton the rest of the app uses, so no divergent instance. is_running()
|
|
# probes the HTTP API, so an already-running Ollama (e.g. the Windows service)
|
|
# still reports as running.
|
|
ollama = get_ollama_manager()
|
|
ollama.keep_alive = store.get_settings().get("keep_alive") or ollama.keep_alive
|
|
print("[Synapse] Ready. Ollama not auto-started (manual control).")
|
|
|
|
|
|
# -------------------------
|
|
# Status
|
|
# -------------------------
|
|
# Was GET / — moved so the built frontend (mounted at / below) owns the root.
|
|
@app.get("/status")
|
|
async def root():
|
|
try:
|
|
# to_thread, not a direct call: get_status() does blocking IO (an httpx
|
|
# request and, once, a subprocess). Awaiting it inline stalled the whole
|
|
# event loop on every poll - and the UI polls /status continuously, so
|
|
# the server froze in lockstep with its own health check.
|
|
status = (await _asyncio.to_thread(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, "platform": _platform.system().lower()}
|
|
|
|
|
|
# -------------------------
|
|
# 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 precedence: explicit request > active playbook's pinned model > auto-select.
|
|
_active_pb = playbooks.get_main_playbook()
|
|
_pb_model = _active_pb.model if (_active_pb and _active_pb.model) else ""
|
|
model = payload.get("model") or _pb_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"))
|
|
num_ctx = payload.get("num_ctx", app_settings.get("num_ctx", 0))
|
|
think = payload.get("think", app_settings.get("think", False))
|
|
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
|
|
|
|
# Resolve the RAG scope: an existing conversation keeps its bound project;
|
|
# a brand-new one inherits the current workspace (active_project setting).
|
|
_conv_proj = store.conversation_project(conversation_id)
|
|
rag_scope = _conv_proj if _conv_proj is not None else app_settings.get("active_project", "")
|
|
|
|
# Retrieve relevant uploaded documents (RAG) and inject the top chunks.
|
|
doc_hits = await store.search_documents(
|
|
message, get_ollama_manager().embed,
|
|
limit=app_settings.get("rag_top_k", 3),
|
|
min_score=app_settings.get("rag_min_score", 0.6),
|
|
project_id=rag_scope or None,
|
|
)
|
|
doc_titles: list = []
|
|
if doc_hits:
|
|
doc_block = "\n\n".join(f"[{d['title']}]\n{d['text']}" for d in doc_hits)
|
|
separator = "\n\n---\nRelevant documents (cite as source material):\n\n"
|
|
system_prompt = (system_prompt + separator + doc_block) if system_prompt else doc_block
|
|
doc_titles = list(dict.fromkeys(d["title"] for d in doc_hits)) # unique, order-preserving
|
|
|
|
# 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, "num_ctx": num_ctx, "think": think}
|
|
|
|
# Vision: base64 images (data: prefix stripped by the client) ride on the user turn.
|
|
images = payload.get("images")
|
|
if images:
|
|
metadata["images"] = images
|
|
|
|
# Tool-using playbook: advertise the active playbook's allowlisted tools.
|
|
# Action tools follow action_tool_policy: off (withheld) / ask (per-call
|
|
# approval, handled in the tool loop) / allow (run freely).
|
|
_policy = app_settings.get("action_tool_policy", "off")
|
|
if _main_pb and getattr(_main_pb, "tools", None):
|
|
allow_actions = _policy != "off"
|
|
schemas = _tools.schemas_for(_main_pb.tools, allow_actions)
|
|
if schemas:
|
|
metadata["tools"] = schemas
|
|
metadata["action_tool_policy"] = _policy
|
|
metadata["conversation_id"] = conversation_id
|
|
_granted = [t for t in _main_pb.tools if not _tools.is_action(t) or allow_actions]
|
|
_withheld = [t for t in _main_pb.tools if _tools.is_action(t) and not allow_actions]
|
|
_synapse_trace(f" TOOLS : {', '.join(_granted)} [actions: {_policy}]\n")
|
|
if _withheld:
|
|
_synapse_trace(f" WITHHELD: {', '.join(_withheld)} (action tools off)\n")
|
|
|
|
# Persist conversation and user message before streaming
|
|
store.create_conversation(conversation_id, rag_scope or "")
|
|
store.add_message(conversation_id, "user", rendered_message)
|
|
|
|
async def event_stream() -> AsyncGenerator[str, None]:
|
|
response_chunks: list[str] = []
|
|
meta: dict = {}
|
|
final_model = model
|
|
|
|
# Tell the client which documents fed this answer (RAG citations).
|
|
if doc_titles:
|
|
yield f"event: sources\ndata: {_json.dumps({'sources': doc_titles})}\n\n"
|
|
|
|
# ── 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
|
|
if chunk.startswith("__status__"):
|
|
yield f"event: status\ndata: {_json.dumps({'tool': chunk[10:]})}\n\n"
|
|
continue
|
|
if chunk.startswith("__approve__"):
|
|
# Loop is paused awaiting the user; forward the pending actions.
|
|
yield f"event: tool_request\ndata: {chunk[11:]}\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))
|
|
|
|
|
|
@app.post("/chat/approve")
|
|
async def chat_approve(payload: Dict[str, Any] = Body(...)):
|
|
"""Resolve a pending per-call tool approval. `decisions` maps tool name ->
|
|
bool; the awaiting chat stream resumes and runs the approved actions.
|
|
|
|
The `token` (issued in the stream's tool_request event) is required: without
|
|
it, anyone who can guess/enumerate a conversation_id could approve another
|
|
client's pending action. Compared in constant time."""
|
|
conversation_id = payload.get("conversation_id") or ""
|
|
token = payload.get("token") or ""
|
|
decisions = payload.get("decisions") or {}
|
|
waiter = _chat.pending_approvals.get(conversation_id)
|
|
if not waiter:
|
|
raise HTTPException(status_code=404, detail="no pending approval for this conversation")
|
|
expected = waiter.get("token") or ""
|
|
if not token or not _secrets.compare_digest(str(token), str(expected)):
|
|
raise HTTPException(status_code=403, detail="invalid or missing approval token")
|
|
waiter["decisions"] = {k: bool(v) for k, v in decisions.items()}
|
|
waiter["event"].set()
|
|
return {"status": "resumed"}
|
|
|
|
# -------------------------
|
|
# Settings
|
|
# -------------------------
|
|
# -------------------------
|
|
# Logs
|
|
# -------------------------
|
|
|
|
# Service log files surfaced in the web UI Logs tab. Fixed dict = no path
|
|
# traversal from the {name} param.
|
|
_LOG_FILES = {
|
|
"backend": settings.runtime_dir / "backend.log",
|
|
"memory": settings.runtime_dir / "memory.log",
|
|
"frontend": settings.runtime_dir / "frontend.log",
|
|
"ollama": settings.logs_dir / "ollama.log",
|
|
"chat": settings.logs_dir / "chat.log",
|
|
}
|
|
|
|
|
|
def _tail(path: Path, n: int) -> str:
|
|
# ponytail: deque reads the whole file, keeps last n lines. Fine at current
|
|
# sizes (<200KB). Switch to seek-from-end if a log ever runs into MBs.
|
|
from collections import deque
|
|
with path.open("r", errors="replace") as f:
|
|
return "".join(deque(f, maxlen=n))
|
|
|
|
|
|
@app.get("/logs")
|
|
def list_logs():
|
|
return {"logs": list(_LOG_FILES.keys())}
|
|
|
|
|
|
@app.get("/logs/{name}")
|
|
def read_log(name: str, lines: int = 400):
|
|
path = _LOG_FILES.get(name)
|
|
if path is None:
|
|
raise HTTPException(status_code=404, detail="unknown log")
|
|
lines = max(1, min(lines, 5000))
|
|
if not path.exists():
|
|
return {"name": name, "content": "", "missing": True}
|
|
return {"name": name, "content": _tail(path, lines)}
|
|
|
|
|
|
@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/recommended")
|
|
async def models_recommended():
|
|
"""Curated models annotated with whether they fit this machine's VRAM/RAM."""
|
|
from . import hardware
|
|
return hardware.recommend()
|
|
|
|
|
|
@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
|
|
# -------------------------
|
|
_warm_tasks: set = set()
|
|
|
|
|
|
async def _warm_default_model():
|
|
"""Load the default model into RAM/VRAM so the first chat isn't a cold read.
|
|
Returns the model name. Raises are the caller's to swallow."""
|
|
s = store.get_settings()
|
|
warm_model = s.get("model") or await ollama.select_best_model()
|
|
num_gpu = await ollama.resolve_num_gpu(s.get("gpu_offload", -1), warm_model)
|
|
await ollama.warm(warm_model, num_gpu)
|
|
return warm_model
|
|
|
|
|
|
@app.post("/ollama/start")
|
|
async def ollama_start_endpoint(background: bool = False):
|
|
try:
|
|
global ollama
|
|
if ollama is None:
|
|
ollama = initialize_ollama()
|
|
|
|
# start_async, NOT start: the sync one polls with time.sleep(1) up to 30
|
|
# times, which blocks the event loop — the whole backend (including the
|
|
# UI's 5s status poll) goes dead while Ollama boots, so a slow start
|
|
# looks like a frozen app rather than a slow one.
|
|
if hasattr(ollama, "start_async"):
|
|
await ollama.start_async()
|
|
|
|
is_running = ollama.is_running() if hasattr(ollama, "is_running") else True
|
|
|
|
# Warm the default model so the first chat isn't a cold load.
|
|
# background=False (UI "Start AI"): block until resident, so the button
|
|
# finishes only once the AI can actually answer.
|
|
# background=True (`ncp start`): fire-and-forget so boot returns fast and
|
|
# the model warms concurrently — Ollama serialises the load, so a first
|
|
# chat that arrives mid-warm simply waits on the same load.
|
|
warmed = None
|
|
if is_running:
|
|
if background:
|
|
task = _asyncio.create_task(_warm_default_model())
|
|
_warm_tasks.add(task) # hold a ref (asyncio only weak-refs tasks)
|
|
task.add_done_callback(_warm_tasks.discard)
|
|
warmed = "background"
|
|
else:
|
|
try:
|
|
warmed = await _warm_default_model()
|
|
except Exception:
|
|
pass
|
|
|
|
return {"status": "started" if is_running else "failed", "running": is_running, "warmed": warmed}
|
|
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))
|
|
|
|
# -------------------------
|
|
# Vite Dev Server (Settings toggle) - dev-only convenience, not part of the
|
|
# single-process production path. Runs via a thread so npm's own startup time
|
|
# doesn't block the event loop, matching the Ollama start/stop pattern above.
|
|
# -------------------------
|
|
@app.get("/frontend/status")
|
|
async def frontend_status_endpoint():
|
|
return {"running": await _asyncio.to_thread(_frontend_manager.is_running)}
|
|
|
|
|
|
@app.post("/frontend/start")
|
|
async def frontend_start_endpoint():
|
|
try:
|
|
return await _asyncio.to_thread(_frontend_manager.start)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@app.post("/frontend/stop")
|
|
async def frontend_stop_endpoint():
|
|
try:
|
|
return await _asyncio.to_thread(_frontend_manager.stop)
|
|
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", []),
|
|
"tools": getattr(p, "tools", []),
|
|
"model": getattr(p, "model", ""),
|
|
}
|
|
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", []),
|
|
tools=playbook_dict.get("tools", []),
|
|
model=playbook_dict.get("model", ""),
|
|
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,
|
|
"tools": playbook_item.tools,
|
|
"model": playbook_item.model,
|
|
}
|
|
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 []),
|
|
"tools": getattr(pb, "tools", []) or (pb.get("tools") if isinstance(pb, dict) else []),
|
|
"model": getattr(pb, "model", "") or (pb.get("model") 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))
|
|
|
|
|
|
# -------------------------
|
|
# Speech-to-text (local Whisper)
|
|
# -------------------------
|
|
from . import stt as _stt
|
|
|
|
|
|
@app.get("/stt/status")
|
|
async def stt_status():
|
|
return {"available": _stt.available()}
|
|
|
|
|
|
@app.post("/stt")
|
|
async def stt_transcribe(payload: Dict[str, Any] = Body(...)):
|
|
if not _stt.available():
|
|
raise HTTPException(status_code=503, detail="local STT (faster-whisper) not installed")
|
|
audio = payload.get("audio") or ""
|
|
if not audio:
|
|
raise HTTPException(status_code=400, detail="audio (base64) is required")
|
|
try:
|
|
text = await _asyncio.to_thread(_stt.transcribe_b64, audio)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"transcription failed: {e}")
|
|
return {"text": text}
|
|
|
|
|
|
# -------------------------
|
|
# Projects / workspaces
|
|
# -------------------------
|
|
def _active_project() -> str:
|
|
"""The active project id, or '' for the unscoped 'All' view."""
|
|
return store.get_settings().get("active_project", "") or ""
|
|
|
|
|
|
@app.get("/projects")
|
|
async def list_projects():
|
|
return {"projects": store.list_projects(), "active": _active_project()}
|
|
|
|
|
|
@app.post("/projects")
|
|
async def create_project(payload: Dict[str, Any] = Body(...)):
|
|
name = (payload.get("name") or "").strip()
|
|
if not name:
|
|
raise HTTPException(status_code=400, detail="name is required")
|
|
return store.create_project(name)
|
|
|
|
|
|
@app.delete("/projects/{project_id}")
|
|
async def delete_project(project_id: str):
|
|
if not store.delete_project(project_id):
|
|
raise HTTPException(status_code=404, detail="Not Found")
|
|
# If the deleted project was active, fall back to the "All" view.
|
|
if _active_project() == project_id:
|
|
store.update_settings({"active_project": ""})
|
|
return {"status": "deleted"}
|
|
|
|
|
|
# -------------------------
|
|
# Documents (RAG)
|
|
# -------------------------
|
|
@app.get("/documents")
|
|
async def list_documents():
|
|
# Scope to the active project; "" (All) lists everything.
|
|
active = _active_project()
|
|
return {"documents": store.list_documents(active if active else None)}
|
|
|
|
|
|
@app.post("/documents")
|
|
async def add_document(payload: Dict[str, Any] = Body(...)):
|
|
title = (payload.get("title") or "").strip()
|
|
content = (payload.get("content") or "").strip()
|
|
if not title or not content:
|
|
raise HTTPException(status_code=400, detail="title and content are required")
|
|
result = await store.add_document(title, content, get_ollama_manager().embed, _active_project())
|
|
if result["chunks"] == 0:
|
|
raise HTTPException(status_code=400, detail="no text to index")
|
|
return result
|
|
|
|
|
|
def _extract_text(filename: str, data: bytes) -> str:
|
|
"""Pull plain text from an uploaded file by extension. PDF/DOCX use
|
|
pure-Python parsers; anything else is decoded as UTF-8."""
|
|
import io
|
|
name = (filename or "").lower()
|
|
if name.endswith(".pdf"):
|
|
from pypdf import PdfReader
|
|
reader = PdfReader(io.BytesIO(data))
|
|
return "\n\n".join((page.extract_text() or "") for page in reader.pages)
|
|
if name.endswith(".docx"):
|
|
import docx
|
|
doc = docx.Document(io.BytesIO(data))
|
|
return "\n\n".join(p.text for p in doc.paragraphs if p.text.strip())
|
|
return data.decode("utf-8", errors="replace")
|
|
|
|
|
|
@app.post("/documents/upload")
|
|
async def upload_document(payload: Dict[str, Any] = Body(...)):
|
|
"""Ingest a file (pdf/docx/txt/md) sent as base64. Extracts text, then runs
|
|
the same chunk/embed pipeline as a pasted document."""
|
|
import base64
|
|
import os as _os
|
|
filename = (payload.get("filename") or "").strip()
|
|
b64 = payload.get("data") or ""
|
|
if not filename or not b64:
|
|
raise HTTPException(status_code=400, detail="filename and data are required")
|
|
try:
|
|
raw = base64.b64decode(b64)
|
|
except Exception:
|
|
raise HTTPException(status_code=400, detail="data must be base64")
|
|
try:
|
|
text = _extract_text(filename, raw).strip()
|
|
except Exception as e:
|
|
raise HTTPException(status_code=400, detail=f"could not read {filename}: {e}")
|
|
if not text:
|
|
raise HTTPException(status_code=400, detail="no extractable text in file")
|
|
title = _os.path.splitext(_os.path.basename(filename))[0] or filename
|
|
return await store.add_document(title, text, get_ollama_manager().embed, _active_project())
|
|
|
|
|
|
@app.get("/documents/{doc_id}")
|
|
async def get_document(doc_id: str):
|
|
chunks = store.get_document(doc_id)
|
|
if not chunks:
|
|
raise HTTPException(status_code=404, detail="Not Found")
|
|
return {"doc_id": doc_id, "chunks": chunks}
|
|
|
|
|
|
@app.delete("/documents/{doc_id}")
|
|
async def delete_document(doc_id: str):
|
|
if not store.delete_document(doc_id):
|
|
raise HTTPException(status_code=404, detail="Not Found")
|
|
return {"status": "deleted"}
|
|
|
|
|
|
# -------------------------
|
|
# 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, conversation_id: Optional[str] = None):
|
|
"""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)
|
|
conversation_id — export just this one conversation (default: all)
|
|
"""
|
|
from fastapi.responses import Response
|
|
import datetime
|
|
|
|
if conversation_id:
|
|
one = store.get_conversation(conversation_id)
|
|
conversations = [one] if one else []
|
|
else:
|
|
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))
|
|
|
|
# -------------------------
|
|
# Built frontend (single-process mode)
|
|
# -------------------------
|
|
# Serve the Vite build so the backend IS the whole app — one process on :8000
|
|
# serves API + UI, no separate Node/Vite at runtime. Mounted LAST so every API
|
|
# route above wins; the SPA only catches what's left. Dev (no dist) skips this
|
|
# and uses the Vite dev server as before.
|
|
from fastapi.staticfiles import StaticFiles # noqa: E402
|
|
|
|
_DIST = Path(__file__).resolve().parent.parent / "interface" / "web" / "dist"
|
|
if _DIST.is_dir():
|
|
app.mount("/", StaticFiles(directory=str(_DIST), html=True), name="ui")
|
|
|
|
# -------------------------
|
|
# End of file
|
|
# -------------------------
|
GitNexus · part of the Nexus stack