Athena/NexusOS
1
0
Fork 0
forked from enderofwings/NexusOS
Code Pull Requests Activity

fix(runtime): improve local service reliability

Browse Source 
Close SQLite handles safely on Windows, clean orphaned vectors, normalize Ollama endpoints, and surface model errors without leaking reasoning tags.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Athena
2026-08-20 00:52:08 -05:00
committed by Athena
co-authored by Cursor
parent 00bd43d32e
commit d45ce69b38
6 changed files with 414 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/ollama_manager.py
+155 -9
View File
@@ -200,6 +200,140 @@ def _chat_options(temperature: float | None, num_gpu: int | None, num_ctx: int |
return opts
_THINK_OPEN = "<think>"
_THINK_CLOSE = "</think>"
def _partial_tag_tail(text: str, tag: str) -> int:
"""Length of the longest suffix of `text` that could be the start of `tag`.
A tag can arrive split across stream chunks ("<thi" then "nk>"), so that much
of the tail has to be held back rather than emitted.
"""
for n in range(min(len(tag) - 1, len(text)), 0, -1):
if text.endswith(tag[:n]):
return n
return 0
class ThinkStripper:
"""Removes a reasoning model's <think> spans from a token stream.
Ollama routes reasoning into `message.thinking` only when asked to think.
We ask for `think: False` because the reasoning is pure latency here — but
deepseek-r1 and friends emit `<think>` inline in `content` anyway, so the
whole internal monologue reached the chat window, closing tags and all.
Two shapes show up in practice. A well-formed span is dropped whole. A
*stray* closing tag with no opening — which is what actually shipped — is at
least removed, so the user does not see a literal `</think>` in the reply.
Text already streamed before it cannot be recalled; `strip_think()` handles
that case properly for callers that have the complete message.
"""
def __init__(self) -> None:
self._buf = ""
self._inside = False
def feed(self, chunk: str) -> str:
self._buf += chunk
out: list[str] = []
while True:
if self._inside:
end = self._buf.find(_THINK_CLOSE)
if end == -1:
keep = _partial_tag_tail(self._buf, _THINK_CLOSE)
self._buf = self._buf[len(self._buf) - keep:] if keep else ""
break
self._buf = self._buf[end + len(_THINK_CLOSE):]
self._inside = False
continue
start = self._buf.find(_THINK_OPEN)
stray = self._buf.find(_THINK_CLOSE)
# A stray close before any open: drop the tag, keep going.
if stray != -1 and (start == -1 or stray < start):
out.append(self._buf[:stray])
self._buf = self._buf[stray + len(_THINK_CLOSE):]
continue
if start == -1:
keep = max(
_partial_tag_tail(self._buf, _THINK_OPEN),
_partial_tag_tail(self._buf, _THINK_CLOSE),
)
if keep:
out.append(self._buf[:len(self._buf) - keep])
self._buf = self._buf[len(self._buf) - keep:]
else:
out.append(self._buf)
self._buf = ""
break
out.append(self._buf[:start])
self._buf = self._buf[start + len(_THINK_OPEN):]
self._inside = True
return "".join(out)
def flush(self) -> str:
"""Whatever is left once the stream ends. An unterminated <think> is
reasoning that never closed, so it is dropped rather than shown."""
rest = "" if self._inside else self._buf
self._buf = ""
return rest
def strip_think(text: str) -> str:
"""Remove reasoning from a complete message.
Unlike the streaming case this sees everything, so an unmatched closing tag
can be handled the way it was meant: every token before it was reasoning,
and the answer is what follows.
"""
if not text or _THINK_CLOSE not in text and _THINK_OPEN not in text:
return text
import re
cleaned = re.sub(r"<think>.*?</think>", "", text, flags=re.S)
if _THINK_CLOSE in cleaned: # stray close: the answer follows it
cleaned = cleaned.rsplit(_THINK_CLOSE, 1)[1]
cleaned = re.sub(r"<think>.*\Z", "", cleaned, flags=re.S) # never closed
return cleaned.strip()
async def _raise_for_ollama(r: httpx.Response) -> None:
"""raise_for_status(), but say what Ollama actually said.
Ollama answers every failure with {"error": "..."} — a model that isn't
pulled, a request too large for VRAM, a cloud model retired upstream — and
httpx's default message discards the body, leaving the user with:
Client error '410 Gone' for url 'http://127.0.0.1:11434/api/chat'
when the body held "glm-4.6 was retired at 2026-06-16". Same exception type
as before so existing handlers are unaffected; only the message improves.
"""
if r.is_success:
return
# A streamed response has no body loaded yet; reading it is what makes the
# error message available at all.
try:
await r.aread()
except Exception:
pass
detail = ""
try:
body = r.json()
if isinstance(body, dict):
detail = str(body.get("error") or "").strip()
except Exception:
detail = (r.text or "").strip()
if not detail:
r.raise_for_status() # nothing to add — keep httpx's wording
raise httpx.HTTPStatusError(
f"Ollama {r.status_code} from {r.request.url.path}: {detail[:400]}",
request=r.request,
response=r,
)
class OllamaManager:
def __init__(self, runtime_dir=None):
self.process = None
@@ -258,7 +392,9 @@ class OllamaManager:
duplicated verbatim in two methods, so the Windows carve-out below had
to be fixed in both places or the two paths would disagree."""
env = os.environ.copy()
env["OLLAMA_HOST"] = self._api_base
# The bind value, not the connect value: a user who set 0.0.0.0 to reach
# Ollama from another machine must still get a server that listens there.
env["OLLAMA_HOST"] = settings.ollama_bind
# Every platform uses the project's own model store. Windows used to be
# exempt, because the installer pulled with a bare `ollama pull` into
# %USERPROFILE%\.ollama and a NexusOS-spawned serve pointed elsewhere
@@ -459,7 +595,7 @@ class OllamaManager:
elapsed = time.perf_counter() - start
_log.info("generate completed model=%s status=%d elapsed=%.3fs", model, r.status_code, elapsed)
r.raise_for_status()
await _raise_for_ollama(r)
return r.json().get("response", "")
except Exception as e:
@@ -484,7 +620,7 @@ class OllamaManager:
"stream": True,
}),
) as response:
response.raise_for_status()
await _raise_for_ollama(response)
async for line in response.aiter_lines():
if not line.strip():
continue
@@ -547,8 +683,12 @@ class OllamaManager:
async with httpx.AsyncClient(timeout=300.0) as client:
r = await client.post(f"{self._api_base}/api/chat", json=body)
elapsed = time.perf_counter() - start
r.raise_for_status()
await _raise_for_ollama(r)
message = r.json().get("message", {})
# A reasoning model puts its monologue in `content` even with
# think off, so strip it before anyone reads the answer.
if isinstance(message, dict) and message.get("content"):
message["content"] = strip_think(message["content"])
# Tool callers need the whole message (tool_calls); others want content.
return message if tools else message.get("content", "")
except Exception as e:
@@ -576,7 +716,7 @@ class OllamaManager:
f"{self._api_base}/api/embeddings",
json=body,
)
r.raise_for_status()
await _raise_for_ollama(r)
vec = r.json().get("embedding")
return vec if vec else None
except Exception as e:
@@ -588,7 +728,7 @@ class OllamaManager:
try:
async with httpx.AsyncClient(timeout=5.0) as client:
r = await client.get(f"{self._api_base}/api/tags")
r.raise_for_status()
await _raise_for_ollama(r)
return [m["name"] for m in r.json().get("models", [])]
except Exception:
return []
@@ -629,7 +769,7 @@ class OllamaManager:
try:
async with httpx.AsyncClient(timeout=10.0) as client:
r = await client.post(f"{self._api_base}/api/show", json={"model": model})
r.raise_for_status()
await _raise_for_ollama(r)
info = r.json().get("model_info", {}) or {}
block_count = next(
(v for k, v in info.items() if k.endswith(".block_count")), None
@@ -679,7 +819,8 @@ class OllamaManager:
f"{self._api_base}/api/chat",
json=body,
) as response:
response.raise_for_status()
await _raise_for_ollama(response)
thinking = ThinkStripper()
async for line in response.aiter_lines():
if not line.strip():
continue
@@ -687,8 +828,13 @@ class OllamaManager:
data = json.loads(line)
token = data.get("message", {}).get("content", "")
if token:
yield token
token = thinking.feed(token)
if token:
yield token
if data.get("done", False):
tail = thinking.flush() # held-back partial tag
if tail:
yield tail
elapsed = time.perf_counter() - start
_log.info("chat stream completed model=%s elapsed=%.3fs", model, elapsed)
eval_count = data.get("eval_count", 0)