hanzo/hanzo.vim
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(local): detect Hanzo engine + Ollama + LM Studio; local-aware :AILogin

Browse Source 
:AILogin lists detected local backends first; Enter uses the running local node
with NO login (cloud login optional, fails with a clean one-line message). Adds
Ollama (:11434) + LM Studio (:1234) probes alongside the Hanzo engine (:36900);
:AIStatus lists all detected backends + active route. 59 python tests pass.
This commit is contained in:
hanzo-dev
2026-06-27 22:58:27 +00:00
parent 6454595b0c
commit 7892358d5c
5 changed files with 793 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
autoload/hanzo.vim
+178 -43
View File
@@ -27,6 +27,9 @@ function! hanzo#GetConfig() abort
\ 'route': get(g:, 'hanzo_route', 'auto'),
\ 'local_url': get(g:, 'hanzo_local_url', 'http://127.0.0.1:36900'),
\ 'local_model': get(g:, 'hanzo_local_model', 'default'),
\ 'ollama_url': get(g:, 'hanzo_ollama_url', 'http://127.0.0.1:11434'),
\ 'lmstudio_url': get(g:, 'hanzo_lmstudio_url', 'http://127.0.0.1:1234'),
\ 'local_backends': get(g:, 'hanzo_local_backends', ['hanzo', 'ollama', 'lmstudio']),
\ 'cloud_url': get(g:, 'hanzo_cloud_url', 'https://api.hanzo.ai'),
\ 'llm_gateway': get(g:, 'hanzo_llm_gateway', ''),
\}
@@ -47,6 +50,9 @@ function! hanzo#NeuralProvider() abort
\ 'route': l:config.route,
\ 'local_url': l:config.local_url,
\ 'local_model': l:config.local_model,
\ 'ollama_url': l:config.ollama_url,
\ 'lmstudio_url': l:config.lmstudio_url,
\ 'local_backends': l:config.local_backends,
\ 'cloud_url': l:config.cloud_url,
\ 'llm_gateway': l:config.llm_gateway,
\}
@@ -607,7 +613,26 @@ function! hanzo#LoginArgv(vendor) abort
return []
endfunction
" One clean line when a cloud (OAuth/device-code) login fails -- e.g. hanzo.id
" returning a Cloudflare 525. We never dump the raw multi-line error/stack;
" local AI keeps working regardless.
function! s:CloudLoginFailed() abort
echohl WarningMsg
echomsg 'Cloud login failed (Hanzo ID/IAM unreachable). '
\ . 'Local AI still works if a local engine is running.'
echohl None
endfunction
" Terminal/job exit callback for a cloud login. Vim's exit_cb(job, status) and
" Neovim's on_exit(job, status, event) both put the exit status at arg 1.
function! s:OnLoginExit(...) abort
let l:status = a:0 >= 2 ? a:000[1] : 0
if type(l:status) == v:t_number && l:status != 0
call s:CloudLoginFailed()
return
endif
call hanzo#Status()
endfunction
@@ -626,7 +651,11 @@ function! s:RunTerminal(argv) abort
else
" No +terminal: run synchronously as a last resort.
call system(join(map(copy(a:argv), 'shellescape(v:val)'), ' '))
call hanzo#Status()
if v:shell_error != 0
call s:CloudLoginFailed()
else
call hanzo#Status()
endif
endif
endfunction
@@ -721,51 +750,136 @@ function! s:LoginOAuth(vendor) abort
call s:RunTerminal(hanzo#LoginArgv(a:vendor))
endfunction
" :AILogin [vendor] -- interactive menu when no vendor is given.
function! hanzo#Login(...) abort
let l:vendor = a:0 > 0 ? s:NormalizeVendor(a:1) : ''
" The detected (running) local backends, in probe order. [] when none are up
" or the route could not be resolved.
function! hanzo#DetectedLocal() abort
let l:route = hanzo#ResolveRoute()
if empty(l:vendor)
let l:choice = inputlist([
\ 'Select AI login (Enter = Hanzo ID):',
\ '1) Hanzo ID (hanzo.id OAuth, device-code) [default]',
\ '2) ChatGPT (OpenAI OAuth, device-code)',
\ '3) Claude (Anthropic API key)',
\ '4) API key (active provider)',
\ '5) Cancel',
\])
echo "\n"
return filter(copy(get(l:route, 'backends', [])),
\ 'get(v:val, "up", v:false)')
endfunction
if l:choice == 0 || l:choice == 1
" Enter / no selection defaults to Hanzo ID.
let l:vendor = 'hanzo'
elseif l:choice == 2
let l:vendor = 'chatgpt'
elseif l:choice == 3
let l:vendor = 'claude'
elseif l:choice == 4
let l:vendor = 'apikey'
else
echo 'AILogin cancelled.'
" Confirm a local backend is active. No login runs -- a local engine needs
" none. Clearing the explicit-provider flag lets auto-routing prefer it.
function! s:LocalActiveMsg(backend) abort
let l:label = type(a:backend) == v:t_dict
\ ? get(a:backend, 'label', 'local engine') : 'local engine'
let l:host = type(a:backend) == v:t_dict ? get(a:backend, 'host', '') : ''
let l:where = empty(l:host) ? l:label : printf('%s (%s)', l:label, l:host)
return
endif
echo printf('Local AI active via %s - no login needed.', l:where)
endfunction
" Menu choice of a detected local backend: confirm, no login. Local-first auto
" routing already prefers it, so just drop any explicit cloud choice.
function! s:UseLocal(backend) abort
let g:hanzo_route = 'auto'
let g:hanzo_provider_explicit = 0
call s:LocalActiveMsg(a:backend)
endfunction
" :AILogin local -- force the local route regardless of probe latency.
function! s:ForceLocal() abort
let g:hanzo_route = 'local'
let g:hanzo_provider_explicit = 0
let l:detected = hanzo#DetectedLocal()
if empty(l:detected)
echo 'Local route forced. No local engine detected yet - start the '
\ . 'Hanzo engine, Ollama, or LM Studio.'
else
call s:LocalActiveMsg(l:detected[0])
endif
endfunction
if l:vendor ==# 'hanzo' || l:vendor ==# 'chatgpt'
call s:LoginOAuth(l:vendor)
elseif l:vendor ==# 'claude'
" Run a cloud login for a vendor. A device-code/OAuth failure surfaces via
" s:OnLoginExit as one clean line, never a raw stack.
function! s:LoginCloud(vendor) abort
if a:vendor ==# 'hanzo' || a:vendor ==# 'chatgpt'
call s:LoginOAuth(a:vendor)
elseif a:vendor ==# 'claude'
call s:LoginApiKey('anthropic')
elseif l:vendor ==# 'apikey'
elseif a:vendor ==# 'apikey'
call s:LoginApiKey(get(g:, 'hanzo_provider', 'anthropic'))
else
echohl ErrorMsg
echomsg 'Unknown AILogin vendor: ' . l:vendor
\ . ' (use claude|chatgpt|hanzo|apikey)'
echomsg 'Unknown AILogin vendor: ' . a:vendor
\ . ' (use local|hanzo|chatgpt|claude|apikey)'
echohl None
endif
endfunction
" :AILogin [target] -- local-aware. With no argument it shows a menu that lists
" any detected local backends first (Enter uses the first one, no login) and
" the cloud options after. With an argument it acts directly:
" local|hanzo|chatgpt|claude|apikey
function! hanzo#Login(...) abort
let l:vendor = a:0 > 0 ? s:NormalizeVendor(a:1) : ''
if !empty(l:vendor)
return l:vendor ==# 'local'
\ ? s:ForceLocal()
\ : s:LoginCloud(l:vendor)
endif
" Interactive menu: probe local backends once, list them first.
let l:route = hanzo#ResolveRoute()
let l:detected = filter(copy(get(l:route, 'backends', [])),
\ 'get(v:val, "up", v:false)')
let l:default = empty(l:detected)
\ ? 'Hanzo ID'
\ : 'Local ' . get(l:detected[0], 'label', 'engine')
let l:display = ['Select AI login (Enter = ' . l:default . '):']
let l:actions = []
for l:backend in l:detected
call add(l:display, printf('%d) Local: %s (%s) [running, no login]',
\ len(l:actions) + 1, get(l:backend, 'label', ''),
\ get(l:backend, 'host', '')))
call add(l:actions, {'kind': 'local', 'backend': l:backend})
endfor
for [l:label, l:vend] in [
\ ['Hanzo ID (hanzo.id)', 'hanzo'],
\ ['ChatGPT', 'chatgpt'],
\ ['Claude', 'claude'],
\ ['API key', 'apikey'],
\ ['Cancel', 'cancel'],
\]
call add(l:display, printf('%d) %s', len(l:actions) + 1, l:label))
call add(l:actions, {'kind': 'cloud', 'vendor': l:vend})
endfor
let l:choice = inputlist(l:display)
echo "\n"
" Enter / Esc (0): first detected local backend if any, else Hanzo ID.
if l:choice <= 0
return empty(l:detected)
\ ? s:LoginCloud('hanzo')
\ : s:UseLocal(l:detected[0])
endif
if l:choice > len(l:actions)
echo 'AILogin cancelled.'
return
endif
let l:action = l:actions[l:choice - 1]
if l:action.kind ==# 'local'
return s:UseLocal(l:action.backend)
endif
if l:action.vendor ==# 'cancel'
echo 'AILogin cancelled.'
return
endif
return s:LoginCloud(l:action.vendor)
endfunction
" :AILogout -- clear the shared credential stores via `dev`.
function! hanzo#Logout() abort
if s:HasDev()
@@ -800,20 +914,41 @@ function! hanzo#ResolveRoute() abort
return type(l:parsed) == v:t_dict ? l:parsed : {}
endfunction
" :AIStatus / :AIWhoami -- show the active route, login state, and provider.
" :AIStatus / :AIWhoami -- list every detected local backend (model + up/down),
" the active route, and the login state.
function! hanzo#Status() abort
let l:route = hanzo#ResolveRoute()
if empty(l:route)
echo 'route: unknown (could not probe local engine)'
elseif get(l:route, 'route', '') ==# 'local'
echo printf('route=local engine %s (%s) [UP]',
\ get(l:route, 'base_url', ''), get(l:route, 'model', ''))
echo 'route: unknown (could not probe local engines)'
else
let l:note = get(l:route, 'authenticated', v:false)
\ ? '' : ' [no credential]'
echo printf('route=cloud provider=%s (cloud %s)%s',
\ get(l:route, 'provider', ''), get(l:route, 'base_url', ''), l:note)
let l:backends = get(l:route, 'backends', [])
if !empty(l:backends)
let l:parts = []
for l:b in l:backends
if get(l:b, 'up', v:false)
call add(l:parts, printf('%s %s (%s) [UP]',
\ get(l:b, 'label', ''), get(l:b, 'host', ''),
\ get(l:b, 'model', '')))
else
call add(l:parts,
\ printf('%s [down]', get(l:b, 'label', '')))
endif
endfor
echo 'Local backends: ' . join(l:parts, ' ; ')
endif
if get(l:route, 'route', '') ==# 'local'
echo printf('Active route: local -> %s (%s)',
\ get(l:route, 'base_url', ''), get(l:route, 'model', ''))
else
let l:note = get(l:route, 'authenticated', v:false)
\ ? '' : ' [no credential]'
echo printf('Active route: cloud -> provider=%s %s%s',
\ get(l:route, 'provider', ''),
\ get(l:route, 'base_url', ''), l:note)
endif
endif
if s:HasDev()
@@ -830,7 +965,7 @@ endfunction
function! hanzo#LoginComplete(arglead, cmdline, cursorpos) abort
let l:matches = []
for l:vendor in ['claude', 'chatgpt', 'hanzo', 'apikey']
for l:vendor in ['local', 'hanzo', 'chatgpt', 'claude', 'apikey']
if stridx(l:vendor, a:arglead) == 0
call add(l:matches, l:vendor)
endif
lua/hanzo/init.lua
+7
View File
@@ -19,6 +19,10 @@ M.config = {
route = "auto", -- auto | local | cloud
local_url = "http://127.0.0.1:36900",
local_model = "default",
-- Additional local backends detected (in order); first one up wins.
ollama_url = "http://127.0.0.1:11434",
lmstudio_url = "http://127.0.0.1:1234",
local_backends = { "hanzo", "ollama", "lmstudio" },
cloud_url = "https://api.hanzo.ai",
-- Keybinds
set_keymaps = false,
@@ -34,6 +38,9 @@ function M.setup(opts)
vim.g.hanzo_route = M.config.route
vim.g.hanzo_local_url = M.config.local_url
vim.g.hanzo_local_model = M.config.local_model
vim.g.hanzo_ollama_url = M.config.ollama_url
vim.g.hanzo_lmstudio_url = M.config.lmstudio_url
vim.g.hanzo_local_backends = M.config.local_backends
vim.g.hanzo_cloud_url = M.config.cloud_url
if M.config.set_keymaps then
plugin/hanzo.vim
+6
View File
@@ -51,6 +51,12 @@ let g:hanzo_mode = get(g:, 'hanzo_mode', 'api')
let g:hanzo_route = get(g:, 'hanzo_route', 'auto')
let g:hanzo_local_url = get(g:, 'hanzo_local_url', 'http://127.0.0.1:36900')
let g:hanzo_local_model = get(g:, 'hanzo_local_model', 'default')
" Additional local backends probed (in order) for the local/auto routes.
" The first one that is up wins; all speak the OpenAI-compatible API and need
" no auth. Set g:hanzo_local_backends to reorder or disable detection.
let g:hanzo_ollama_url = get(g:, 'hanzo_ollama_url', 'http://127.0.0.1:11434')
let g:hanzo_lmstudio_url = get(g:, 'hanzo_lmstudio_url', 'http://127.0.0.1:1234')
let g:hanzo_local_backends = get(g:, 'hanzo_local_backends', ['hanzo', 'ollama', 'lmstudio'])
let g:hanzo_cloud_url = get(g:, 'hanzo_cloud_url', 'https://api.hanzo.ai')
" Optional override of the cloud base (e.g. a local gateway on :4000). Empty
" means use g:hanzo_cloud_url.
src/neural/provider/hanzo.py
+295 -34
View File
@@ -18,6 +18,7 @@ import sys
import time
import urllib.error
import urllib.request
from collections.abc import Callable
from typing import Any, NamedTuple, cast
# Try to import websockets for MCP/ZAP bridge
@@ -29,22 +30,33 @@ except ImportError:
# Constants
#
# Local-first routing endpoints. The native Hanzo engine (kept alive by the
# Hanzo desktop app's node manager) speaks the OpenAI-compatible API on :36900
# and needs no auth. The cloud account is the Hanzo gateway, reached with the
# resolved account credentials.
HANZO_LOCAL_URL = "http://127.0.0.1:36900" # native local engine
HANZO_LOCAL_MODEL = "default" # model id the local engine serves
# Local-first routing endpoints. Several OpenAI-compatible local engines may be
# running; we detect them in order and route to the first that answers (no
# auth). The Hanzo engine (kept alive by the Hanzo desktop app's node manager)
# serves on :36900, Ollama on :11434, LM Studio on :1234. The cloud account is
# the Hanzo gateway, reached with the resolved account credentials.
HANZO_LOCAL_URL = "http://127.0.0.1:36900" # native Hanzo engine
HANZO_LOCAL_MODEL = "default" # model id the Hanzo engine serves
HANZO_OLLAMA_URL = "http://127.0.0.1:11434" # Ollama OpenAI-compatible server
HANZO_LMSTUDIO_URL = "http://127.0.0.1:1234" # LM Studio server
HANZO_CLOUD_URL = "https://api.hanzo.ai" # cloud account gateway
HANZO_MCP_BRIDGE = "ws://localhost:9228" # Vim bridge port
DATA_HEADER = "data: "
DONE_MARKER = "[DONE]"
ANTHROPIC_VERSION = "2023-06-01"
# A /health probe result is trusted for this long (per URL), so a burst of
# requests/keystrokes does not re-probe the local engine each time.
_HEALTH_TTL_SECONDS = 5.0
_health_cache: "dict[str, tuple[float, bool]]" = {}
# Order local backends are probed in; the first one that is up wins for the
# local/auto routes. Mirrors the g:hanzo_local_backends vim default.
_DEFAULT_LOCAL_BACKENDS = ("hanzo", "ollama", "lmstudio")
# A probe result is trusted for this long (per URL), so a burst of
# requests/keystrokes does not re-probe a local engine each time. The probe is
# a single GET whose (status, body) is cached: the status tells us a backend is
# up, the body carries its model list.
_PROBE_TIMEOUT = 0.5
_PROBE_TTL_SECONDS = 5.0
_MAX_PROBE_BYTES = 1 << 20 # cap a probe body read at 1 MiB
_probe_cache: "dict[str, tuple[float, tuple[int, bytes]]]" = {}
# Cloud vendors whose explicit selection (with a resolved credential) wins
# over the local engine even in auto mode.
@@ -239,6 +251,9 @@ class HanzoConfig:
route: str = "auto", # auto | local | cloud
local_url: str = HANZO_LOCAL_URL,
local_model: str = HANZO_LOCAL_MODEL,
ollama_url: str = HANZO_OLLAMA_URL,
lmstudio_url: str = HANZO_LMSTUDIO_URL,
local_backends: "tuple[str, ...] | None" = None,
cloud_url: str = HANZO_CLOUD_URL,
gateway: str = "", # optional cloud-base override (e.g. :4000)
@@ -262,6 +277,13 @@ class HanzoConfig:
self.route = route
self.local_url = local_url
self.local_model = local_model
self.ollama_url = ollama_url
self.lmstudio_url = lmstudio_url
self.local_backends = (
local_backends
if local_backends is not None
else _DEFAULT_LOCAL_BACKENDS
)
self.cloud_url = cloud_url
self.gateway = gateway
self.temperature = temperature
@@ -271,6 +293,25 @@ class HanzoConfig:
self.system_prompt = system_prompt
def _parse_local_backends(raw: object) -> "tuple[str, ...]":
"""Return the probe order for local backends from raw config.
Accepts a list of backend names (e.g. from g:hanzo_local_backends),
keeping only non-empty strings and preserving order. Falls back to the
default order when nothing usable is given.
"""
if isinstance(raw, list):
names = tuple(
item for item in cast("list[object]", raw)
if isinstance(item, str) and item
)
if names:
return names
return _DEFAULT_LOCAL_BACKENDS
def load_config(raw_config: dict[str, Any]) -> HanzoConfig:
"""Load and validate configuration."""
if not isinstance(raw_config, dict):
@@ -337,6 +378,9 @@ def load_config(raw_config: dict[str, Any]) -> HanzoConfig:
local_url = raw_config.get("local_url", "") or HANZO_LOCAL_URL
local_model = raw_config.get("local_model", "") or HANZO_LOCAL_MODEL
ollama_url = raw_config.get("ollama_url", "") or HANZO_OLLAMA_URL
lmstudio_url = raw_config.get("lmstudio_url", "") or HANZO_LMSTUDIO_URL
local_backends = _parse_local_backends(raw_config.get("local_backends"))
cloud_url = raw_config.get("cloud_url", "") or HANZO_CLOUD_URL
gateway = (
raw_config.get("llm_gateway", "")
@@ -372,6 +416,9 @@ def load_config(raw_config: dict[str, Any]) -> HanzoConfig:
route=route,
local_url=local_url,
local_model=local_model,
ollama_url=ollama_url,
lmstudio_url=lmstudio_url,
local_backends=local_backends,
cloud_url=cloud_url,
gateway=gateway,
temperature=temperature,
@@ -401,35 +448,222 @@ class Endpoint(NamedTuple):
headers: dict[str, str]
def _probe_health(url: str, timeout: float = 0.7) -> bool:
"""Return True if ``GET {url}/health`` answers 2xx.
def _http_probe(
url: str,
timeout: float = _PROBE_TIMEOUT,
) -> "tuple[int, bytes]":
"""``GET url``; return ``(status, body)``, or ``(0, b"")`` on any error.
The result is cached per URL for a few seconds so a burst of requests
does not re-probe the engine on every keystroke.
This is the single network primitive behind local-backend detection. The
result is cached per URL for a few seconds so a burst of requests does not
re-probe an engine on every keystroke. The body is read (capped) so a probe
that also lists models (Ollama ``/api/tags``, LM Studio ``/v1/models``)
does not need a second request.
"""
now = time.monotonic()
cached = _health_cache.get(url)
cached = _probe_cache.get(url)
if cached is not None and now - cached[0] < _HEALTH_TTL_SECONDS:
if cached is not None and now - cached[0] < _PROBE_TTL_SECONDS:
return cached[1]
ok = False
result: tuple[int, bytes] = (0, b"")
try:
req = urllib.request.Request(f"{url}/health", method="GET")
req = urllib.request.Request(url, method="GET")
with urllib.request.urlopen(req, timeout=timeout) as resp:
status = getattr(resp, "status", 0)
ok = isinstance(status, int) and 200 <= status < 300
raw_status = getattr(resp, "status", 0)
status = raw_status if isinstance(raw_status, int) else 0
result = (status, resp.read(_MAX_PROBE_BYTES))
except (urllib.error.URLError, OSError, ValueError):
ok = False
result = (0, b"")
_health_cache[url] = (now, ok)
_probe_cache[url] = (now, result)
return ok
return result
def _probe_ok(url: str, timeout: float = _PROBE_TIMEOUT) -> bool:
"""Return True if ``GET url`` answers 2xx (cached, see ``_http_probe``)."""
status, _ = _http_probe(url, timeout)
return 200 <= status < 300
def _first_ollama_model(body: bytes) -> str:
"""First model name from an Ollama ``/api/tags`` body, else ``""``."""
parsed: object = _loads_or_none(body)
if not isinstance(parsed, dict):
return ""
models = cast("dict[str, object]", parsed).get("models")
if not isinstance(models, list):
return ""
for item in cast("list[object]", models):
if isinstance(item, dict):
entry = cast("dict[str, object]", item)
name = _str_field(entry, "name") or _str_field(entry, "model")
if name:
return name
return ""
def _first_openai_model(body: bytes) -> str:
"""First model id from an OpenAI-style ``/v1/models`` body, else ``""``."""
parsed: object = _loads_or_none(body)
if not isinstance(parsed, dict):
return ""
items = cast("dict[str, object]", parsed).get("data")
if not isinstance(items, list):
return ""
for item in cast("list[object]", items):
if isinstance(item, dict):
model_id = _str_field(cast("dict[str, object]", item), "id")
if model_id:
return model_id
return ""
def _loads_or_none(body: bytes) -> object:
"""Parse JSON from ``body``; return None on any decode error."""
try:
return json.loads(body.decode("utf-8", errors="replace"))
except ValueError:
return None
class BackendStatus(NamedTuple):
"""A probed local backend: where it is, what it serves, and if it's up."""
name: str # "hanzo" | "ollama" | "lmstudio"
label: str # human label, e.g. "Hanzo engine"
host: str # host:port for display, e.g. "127.0.0.1:36900"
base_url: str # OpenAI-compatible base URL
model: str # model id to use ("" when down/unknown)
up: bool # answered its probe
def _host(url: str) -> str:
"""Strip the scheme from ``url`` for display: ``host:port``."""
return url.split("://", 1)[-1].rstrip("/")
def _detect_hanzo(config: HanzoConfig) -> BackendStatus:
"""Probe the native Hanzo engine: ``GET {base}/health`` == 2xx."""
base = config.local_url
up = _probe_ok(f"{base}/health")
return BackendStatus(
name="hanzo",
label="Hanzo engine",
host=_host(base),
base_url=base,
model=config.local_model if up else "",
up=up,
)
def _detect_ollama(config: HanzoConfig) -> BackendStatus:
"""Probe Ollama: ``GET {base}/api/tags`` == 2xx; model = first tag.
Ollama exposes the OpenAI-compatible API at ``{base}/v1`` and serves no
auth. It is considered usable only when a model name resolves (from
``/api/tags``, falling back to ``/v1/models``).
"""
base = config.ollama_url
status, body = _http_probe(f"{base}/api/tags")
up = 200 <= status < 300
model = ""
if up:
model = _first_ollama_model(body) or _first_openai_model(
_http_probe(f"{base}/v1/models")[1],
)
return BackendStatus(
name="ollama",
label="Ollama",
host=_host(base),
base_url=base,
model=model,
up=up and bool(model),
)
def _detect_lmstudio(config: HanzoConfig) -> BackendStatus:
"""Probe LM Studio: ``GET {base}/v1/models`` == 2xx; model = first id."""
base = config.lmstudio_url
status, body = _http_probe(f"{base}/v1/models")
up = 200 <= status < 300
model = _first_openai_model(body) if up else ""
return BackendStatus(
name="lmstudio",
label="LM Studio",
host=_host(base),
base_url=base,
model=model,
up=up and bool(model),
)
_LOCAL_DETECTORS: "dict[str, Callable[[HanzoConfig], BackendStatus]]" = {
"hanzo": _detect_hanzo,
"ollama": _detect_ollama,
"lmstudio": _detect_lmstudio,
}
def inspect_local_backends(config: HanzoConfig) -> "list[BackendStatus]":
"""Probe every configured local backend (in order) and report each.
Includes backends that are down, so the UI can show their status. Use
``detect_local_backends`` when only the running ones are needed.
"""
report: list[BackendStatus] = []
for name in config.local_backends:
detector = _LOCAL_DETECTORS.get(name)
if detector is not None:
report.append(detector(config))
return report
def detect_local_backends(config: HanzoConfig) -> "list[BackendStatus]":
"""Return the running local backends, in probe order (first = preferred).
Degrades gracefully: a backend whose port is closed simply does not appear,
and no probe blocks longer than ``_PROBE_TIMEOUT``.
"""
return [
backend for backend in inspect_local_backends(config) if backend.up
]
def _endpoint_for(backend: BackendStatus) -> Endpoint:
"""Build a no-auth local Endpoint for a detected backend."""
return Endpoint(
route="local",
base_url=backend.base_url,
model=backend.model,
headers={"Content-Type": "application/json"},
)
def _local_endpoint(config: HanzoConfig) -> Endpoint:
"""Native local engine endpoint. It needs no auth header."""
"""Native Hanzo engine endpoint (unconditional); needs no auth header."""
return Endpoint(
route="local",
base_url=config.local_url,
@@ -461,20 +695,25 @@ def _cloud_endpoint(config: HanzoConfig) -> Endpoint:
def resolve_endpoint(config: HanzoConfig) -> Endpoint:
"""Pick the local engine or the cloud account for this request.
"""Pick a local backend or the cloud account for this request.
- ``local``: always the native engine (no auth).
- ``local``: the first detected local backend (no auth); falls back to the
configured Hanzo engine when none answer, so the route stays local.
- ``cloud``: always the cloud account (resolved account creds).
- ``auto`` (default): an explicitly chosen cloud vendor with a resolved
credential wins; else the local engine when its /health is up; else
the cloud account.
credential wins; else the first detected local backend (Hanzo engine,
Ollama, LM Studio, in order); else the cloud account.
"""
if config.route == "local":
return _local_endpoint(config)
if config.route == "cloud":
return _cloud_endpoint(config)
if config.route == "local":
detected = detect_local_backends(config)
return _endpoint_for(detected[0]) if detected else _local_endpoint(
config,
)
explicit_cloud = (
config.provider_explicit
and config.provider in _CLOUD_VENDORS
@@ -484,8 +723,10 @@ def resolve_endpoint(config: HanzoConfig) -> Endpoint:
if explicit_cloud:
return _cloud_endpoint(config)
if _probe_health(config.local_url):
return _local_endpoint(config)
detected = detect_local_backends(config)
if detected:
return _endpoint_for(detected[0])
return _cloud_endpoint(config)
@@ -691,22 +932,42 @@ def _print_resolution() -> None:
"""Print the resolved route as JSON for :AIStatus. Reads config on stdin.
Input: {"config": {...}} on a single line.
Output: {"route", "base_url", "model", "provider", "authenticated"}.
Output: {"route", "base_url", "model", "provider", "authenticated",
"active_backend", "backends": [{name,label,host,model,up}, ...]}.
"""
payload = json.loads(sys.stdin.readline() or "{}")
config = load_config(payload.get("config", {}))
backends = inspect_local_backends(config)
target = resolve_endpoint(config)
authenticated = bool(
target.headers.get("Authorization")
or target.headers.get("x-api-key"),
)
active = ""
if target.route == "local":
for backend in backends:
if backend.up and backend.base_url == target.base_url:
active = backend.name
break
print(json.dumps({
"route": target.route,
"base_url": target.base_url,
"model": target.model,
"provider": config.provider,
"authenticated": authenticated,
"active_backend": active,
"backends": [
{
"name": backend.name,
"label": backend.label,
"host": backend.host,
"model": backend.model,
"up": backend.up,
}
for backend in backends
],
}))
test/python/provider/test_hanzo.py
+307 -9
View File
@@ -1,5 +1,6 @@
import json
import socket
from collections.abc import Callable
from pathlib import Path
import pytest
@@ -32,12 +33,45 @@ def _closed_port() -> int:
return port
def _always_up(_url: str, timeout: float = 0.7) -> bool:
return True
_Probe = Callable[..., "tuple[int, bytes]"]
def _always_down(_url: str, timeout: float = 0.7) -> bool:
return False
def _probe_up(_url: str, _timeout: float = 0.5) -> tuple[int, bytes]:
"""Fake _http_probe: every backend answers 2xx (empty body)."""
return (200, b"")
def _probe_down(_url: str, _timeout: float = 0.5) -> tuple[int, bytes]:
"""Fake _http_probe: nothing is listening anywhere."""
return (0, b"")
def _routed_probe(routes: dict[str, tuple[int, bytes]]) -> _Probe:
"""Fake _http_probe routing by URL substring (e.g. "11434/api/tags").
The first matching fragment wins; unmatched URLs report down. This lets a
test bring exactly one backend up by its host:port + path.
"""
def probe(url: str, _timeout: float = 0.5) -> tuple[int, bytes]:
for fragment, response in routes.items():
if fragment in url:
return response
return (0, b"")
return probe
def _ollama_tags(*names: str) -> bytes:
return json.dumps(
{"models": [{"name": name} for name in names]},
).encode("utf-8")
def _openai_models(*ids: str) -> bytes:
return json.dumps(
{"data": [{"id": model_id} for model_id in ids]},
).encode("utf-8")
def test_anthropic_prefers_env(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -222,7 +256,7 @@ def test_auto_picks_local_when_engine_up(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_clear_cloud_env(monkeypatch)
monkeypatch.setattr(hanzo, "_probe_health", _always_up)
monkeypatch.setattr(hanzo, "_http_probe", _probe_up)
endpoint = hanzo.resolve_endpoint(hanzo.load_config({"route": "auto"}))
@@ -236,7 +270,7 @@ def test_auto_falls_back_to_cloud_when_engine_down(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_clear_cloud_env(monkeypatch)
monkeypatch.setattr(hanzo, "_probe_health", _always_down)
monkeypatch.setattr(hanzo, "_http_probe", _probe_down)
endpoint = hanzo.resolve_endpoint(hanzo.load_config({"route": "auto"}))
@@ -249,7 +283,7 @@ def test_auto_explicit_cloud_vendor_wins_over_local(
) -> None:
_clear_cloud_env(monkeypatch)
# Engine is up, but the user explicitly chose a cloud vendor with a key.
monkeypatch.setattr(hanzo, "_probe_health", _always_up)
monkeypatch.setattr(hanzo, "_http_probe", _probe_up)
endpoint = hanzo.resolve_endpoint(hanzo.load_config({
"route": "auto",
@@ -269,7 +303,7 @@ def test_auto_default_provider_stays_local_even_with_key(
# A key is present, but the anthropic default was not explicitly chosen,
# so local-first must still win when the engine is up.
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant")
monkeypatch.setattr(hanzo, "_probe_health", _always_up)
monkeypatch.setattr(hanzo, "_http_probe", _probe_up)
endpoint = hanzo.resolve_endpoint(hanzo.load_config({
"route": "auto",
@@ -296,13 +330,15 @@ def test_auto_dead_port_falls_back_to_cloud_no_crash(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Real probe (not monkeypatched) against a closed port: must refuse fast,
# report unhealthy, and fall back to cloud without raising.
# report unhealthy, and fall back to cloud without raising. Constrain to
# the hanzo backend so the test does not depend on a dev's local Ollama.
_clear_cloud_env(monkeypatch)
dead = f"http://127.0.0.1:{_closed_port()}"
endpoint = hanzo.resolve_endpoint(hanzo.load_config({
"route": "auto",
"local_url": dead,
"local_backends": ["hanzo"],
}))
assert endpoint.route == "cloud"
@@ -311,3 +347,265 @@ def test_auto_dead_port_falls_back_to_cloud_no_crash(
def test_invalid_route_defaults_to_auto() -> None:
assert hanzo.load_config({"route": "nonsense"}).route == "auto"
# ---------------------------------------------------------------------------
# Multi-backend local detection (detect_local_backends + inspect_*)
# ---------------------------------------------------------------------------
def test_config_default_backends_order() -> None:
assert hanzo.load_config({}).local_backends == (
"hanzo", "ollama", "lmstudio",
)
def test_config_custom_backends_order() -> None:
config = hanzo.load_config({"local_backends": ["ollama", "hanzo"]})
assert config.local_backends == ("ollama", "hanzo")
@pytest.mark.parametrize("raw", [[], "nope", None, [123, ""]])
def test_config_backends_fall_back_to_default(raw: object) -> None:
config = hanzo.load_config({"local_backends": raw})
assert config.local_backends == ("hanzo", "ollama", "lmstudio")
def _ollama_model(monkeypatch: pytest.MonkeyPatch, body: bytes) -> str:
"""Resolve the model the Ollama detector picks for a /api/tags body."""
monkeypatch.setattr(
hanzo, "_http_probe", _routed_probe({"11434/api/tags": (200, body)}),
)
detected = hanzo.detect_local_backends(
hanzo.load_config({"local_backends": ["ollama"]}),
)
return detected[0].model if detected else ""
@pytest.mark.parametrize(
("body", "expected"),
[
(_ollama_tags("a", "b"), "a"),
(b'{"models": [{"model": "fallback"}]}', "fallback"),
(_ollama_tags(), ""),
(b"not json", ""),
(b"[]", ""),
],
)
def test_ollama_model_parsing_via_detection(
monkeypatch: pytest.MonkeyPatch,
body: bytes,
expected: str,
) -> None:
assert _ollama_model(monkeypatch, body) == expected
def _lmstudio_model(monkeypatch: pytest.MonkeyPatch, body: bytes) -> str:
"""Resolve the model the LM Studio detector picks for a /v1/models body."""
monkeypatch.setattr(
hanzo, "_http_probe", _routed_probe({"1234/v1/models": (200, body)}),
)
detected = hanzo.detect_local_backends(
hanzo.load_config({"local_backends": ["lmstudio"]}),
)
return detected[0].model if detected else ""
@pytest.mark.parametrize(
("body", "expected"),
[
(_openai_models("x", "y"), "x"),
(_openai_models(), ""),
(b"garbage", ""),
(b'{"data": "nope"}', ""),
],
)
def test_lmstudio_model_parsing_via_detection(
monkeypatch: pytest.MonkeyPatch,
body: bytes,
expected: str,
) -> None:
assert _lmstudio_model(monkeypatch, body) == expected
def test_detect_hanzo_up_others_absent(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Only the Hanzo engine (:36900 /health) answers; Ollama and LM Studio
# are not running and must be reported absent without hanging.
monkeypatch.setattr(
hanzo, "_http_probe", _routed_probe({"36900/health": (200, b"")}),
)
detected = hanzo.detect_local_backends(hanzo.load_config({}))
assert [backend.name for backend in detected] == ["hanzo"]
assert detected[0].model == "default"
assert detected[0].host == "127.0.0.1:36900"
assert detected[0].up is True
def test_inspect_reports_all_backends_with_status(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(
hanzo, "_http_probe", _routed_probe({"36900/health": (200, b"")}),
)
report = hanzo.inspect_local_backends(hanzo.load_config({}))
assert [backend.name for backend in report] == [
"hanzo", "ollama", "lmstudio",
]
assert report[0].up is True
assert report[1].up is False
assert report[2].up is False
assert report[1].label == "Ollama"
assert report[2].label == "LM Studio"
def test_auto_resolves_to_local_hanzo_when_only_hanzo_up(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_clear_cloud_env(monkeypatch)
monkeypatch.setattr(
hanzo, "_http_probe", _routed_probe({"36900/health": (200, b"")}),
)
endpoint = hanzo.resolve_endpoint(hanzo.load_config({"route": "auto"}))
assert endpoint.route == "local"
assert "36900" in endpoint.base_url
assert endpoint.model == "default"
assert "Authorization" not in endpoint.headers
def test_detect_skips_ollama_when_up_but_no_models(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Ollama answering /api/tags with an empty list is not usable for routing.
monkeypatch.setattr(
hanzo,
"_http_probe",
_routed_probe({"11434/api/tags": (200, _ollama_tags())}),
)
assert hanzo.detect_local_backends(hanzo.load_config({})) == []
def test_auto_picks_ollama_when_hanzo_down(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Hanzo engine down, Ollama up: auto must route to Ollama with its model.
_clear_cloud_env(monkeypatch)
monkeypatch.setattr(
hanzo,
"_http_probe",
_routed_probe({"11434/api/tags": (200, _ollama_tags("llama3.2"))}),
)
detected = hanzo.detect_local_backends(hanzo.load_config({}))
endpoint = hanzo.resolve_endpoint(hanzo.load_config({"route": "auto"}))
assert [backend.name for backend in detected] == ["ollama"]
assert endpoint.route == "local"
assert "11434" in endpoint.base_url
assert endpoint.model == "llama3.2"
assert "Authorization" not in endpoint.headers
def test_auto_picks_lmstudio_when_only_it_is_up(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_clear_cloud_env(monkeypatch)
monkeypatch.setattr(
hanzo,
"_http_probe",
_routed_probe({"1234/v1/models": (200, _openai_models("qwen-coder"))}),
)
endpoint = hanzo.resolve_endpoint(hanzo.load_config({"route": "auto"}))
assert endpoint.route == "local"
assert "1234" in endpoint.base_url
assert endpoint.model == "qwen-coder"
def test_backend_order_decides_winner_when_several_up(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# Both Hanzo and Ollama up, but the configured order prefers Ollama.
_clear_cloud_env(monkeypatch)
monkeypatch.setattr(
hanzo,
"_http_probe",
_routed_probe({
"36900/health": (200, b""),
"11434/api/tags": (200, _ollama_tags("llama3.2")),
}),
)
endpoint = hanzo.resolve_endpoint(hanzo.load_config({
"route": "auto",
"local_backends": ["ollama", "hanzo"],
}))
assert endpoint.route == "local"
assert "11434" in endpoint.base_url
def test_auto_all_local_down_falls_back_to_cloud(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_clear_cloud_env(monkeypatch)
monkeypatch.setattr(hanzo, "_http_probe", _probe_down)
config = hanzo.load_config({"route": "auto"})
assert hanzo.detect_local_backends(config) == []
assert hanzo.resolve_endpoint(config).route == "cloud"
def test_route_local_falls_back_to_hanzo_engine_when_none_up(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# An explicit local route must stay local even if nothing answers a probe.
_clear_cloud_env(monkeypatch)
monkeypatch.setattr(hanzo, "_http_probe", _probe_down)
endpoint = hanzo.resolve_endpoint(hanzo.load_config({"route": "local"}))
assert endpoint.route == "local"
assert "36900" in endpoint.base_url
assert "Authorization" not in endpoint.headers
def test_probe_is_cached_across_requests(
monkeypatch: pytest.MonkeyPatch,
) -> None:
# The probe is cached per URL so a burst of requests hits the network once.
# Use a unique closed port so this URL is absent from the shared cache, and
# drive it through the public resolve_endpoint (no private access).
_clear_cloud_env(monkeypatch)
calls = {"n": 0}
def counting_open(*_args: object, **_kwargs: object) -> object:
calls["n"] += 1
raise OSError("refused")
monkeypatch.setattr(hanzo.urllib.request, "urlopen", counting_open)
dead = f"http://127.0.0.1:{_closed_port()}"
config = hanzo.load_config({
"route": "local",
"local_url": dead,
"local_backends": ["hanzo"],
})
hanzo.resolve_endpoint(config)
hanzo.resolve_endpoint(config)
assert calls["n"] == 1