|
#!/usr/bin/env python3 |
|
"""Small, local-only Ollama launcher for coding agents.""" |
|
import argparse |
|
from concurrent.futures import ThreadPoolExecutor, as_completed |
|
import hashlib |
|
import json |
|
import os |
|
import re |
|
import shlex |
|
import shutil |
|
import subprocess |
|
import sys |
|
import time |
|
import urllib.error |
|
import urllib.request |
|
from pathlib import Path |
|
import webbrowser |
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent)) |
|
import local_catalog as catalog |
|
import local_hardware as hardware |
|
import local_explore as explore |
|
|
|
ENDPOINT = "http://127.0.0.1:11434" |
|
CONTEXTS = (8192, 16384, 32768, 65536, 131072) |
|
DEFAULT_SETTINGS = {"comfortable_percent": 40.0} |
|
|
|
def settings_path(): |
|
return Path(os.environ.get("LOCAL_CODE_SETTINGS", "~/.config/local-code/settings.json")).expanduser() |
|
|
|
def load_settings(): |
|
path = settings_path() |
|
try: |
|
data = json.loads(path.read_text()) |
|
value = float(data.get("comfortable_percent", DEFAULT_SETTINGS["comfortable_percent"])) |
|
if not 1 <= value < 100: |
|
raise ValueError |
|
return {"comfortable_percent": value} |
|
except (OSError, ValueError, TypeError, json.JSONDecodeError): |
|
return dict(DEFAULT_SETTINGS) |
|
|
|
def save_settings(settings): |
|
path = settings_path() |
|
path.parent.mkdir(parents=True, exist_ok=True) |
|
path.write_text(json.dumps(settings, indent=2) + "\n") |
|
|
|
def settings_flow(): |
|
"""Interactively edit persistent launcher heuristics without touching models.""" |
|
settings = load_settings() |
|
while True: |
|
current = settings["comfortable_percent"] |
|
print(f"\nlocal-code settings ({settings_path()})") |
|
print(f"Comfortable threshold: {current:.0f}% of conservative weight budget") |
|
action = choose("Settings", [("threshold", "Change comfortable threshold"), |
|
("reset", "Reset defaults (40%)"), |
|
("done", "Done")]) |
|
if action in (None, "done"): |
|
return 0 |
|
if action == "reset": |
|
settings = dict(DEFAULT_SETTINGS) |
|
save_settings(settings) |
|
hardware.set_comfortable_fraction(settings["comfortable_percent"] / 100) |
|
print("Reset and saved.") |
|
continue |
|
try: |
|
value = float(input("Comfortable threshold percent (1–99): ").strip()) |
|
if not 1 <= value < 100: |
|
raise ValueError |
|
except (EOFError, ValueError): |
|
print("Enter a number from 1 to 99.") |
|
continue |
|
settings = {"comfortable_percent": value} |
|
save_settings(settings) |
|
hardware.set_comfortable_fraction(value / 100) |
|
print(f"Saved {value:.0f}%. This affects fit labels and recommendations, not model weights.") |
|
|
|
def api(path, method="GET", payload=None, timeout=2): |
|
data = None if payload is None else json.dumps(payload).encode() |
|
req = urllib.request.Request(ENDPOINT + path, data=data, method=method, |
|
headers={"Content-Type": "application/json"}) |
|
# Never let ambient proxy settings route the loopback daemon externally. |
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) |
|
with opener.open(req, timeout=timeout) as response: |
|
raw = response.read() |
|
result = json.loads(raw or b"{}") |
|
if not isinstance(result, dict) or result.get("error"): |
|
raise RuntimeError(f"Ollama returned an error for {path}: {result}") |
|
return result |
|
|
|
def version_tuple(v): |
|
m = re.search(r"(\d+)\.(\d+)\.(\d+)", str(v)) |
|
return tuple(map(int, m.groups())) if m else (0, 0, 0) |
|
|
|
def server_info(): |
|
# Ollama can keep the TCP listener open while it is loading a large model. |
|
# A single short request during that window is not evidence that the daemon |
|
# is stopped (and telling users to run `ollama serve` then produces a |
|
# confusing "address already in use" error). |
|
last = None |
|
for attempt in range(3): |
|
try: |
|
version = api("/api/version", timeout=5).get("version", "unknown") |
|
tags = api("/api/tags", timeout=5).get("models", []) |
|
return version, tags, None |
|
except (urllib.error.URLError, OSError, ValueError, RuntimeError) as exc: |
|
last = exc |
|
if attempt < 2: |
|
time.sleep(0.5) |
|
return None, [], last |
|
|
|
def model_names(tags): |
|
return {m.get("name") or m.get("model") for m in tags if m.get("name") or m.get("model")} |
|
|
|
def model_dir(): |
|
configured = os.environ.get("OLLAMA_MODELS") |
|
if configured: |
|
return Path(configured).expanduser() |
|
return Path.home() / ".ollama" / "models" |
|
|
|
def free_bytes(path): |
|
path = Path(path).expanduser() |
|
while not path.exists() and path != path.parent: |
|
path = path.parent |
|
return shutil.disk_usage(path).free |
|
|
|
def executable(name): |
|
"""Find a user-installed harness even when the shell omitted its bin dir.""" |
|
found = shutil.which(name) |
|
if found: |
|
return found |
|
roots = [Path.home() / ".local" / "bin", |
|
Path.home() / ".npm-global" / "bin", |
|
Path.home() / ".nvm" / "versions" / "node"] |
|
candidates = [] |
|
for root in roots: |
|
if root.name == "node": |
|
candidates.extend(root.glob("*/bin/" + name)) |
|
else: |
|
candidates.append(root / name) |
|
for path in candidates: |
|
if path.is_file() and os.access(path, os.X_OK): |
|
return str(path) |
|
return None |
|
|
|
def harness_env(name): |
|
path = executable(name) |
|
if not path: |
|
return {} |
|
directory = str(Path(path).parent) |
|
current = os.environ.get("PATH", "") |
|
return {"PATH": directory + (os.pathsep + current if current else "")} |
|
|
|
def need_space(size_bytes, path=None): |
|
required = int(size_bytes + 3 * 1024**3) |
|
p = path or model_dir() |
|
return free_bytes(p), required |
|
|
|
def pull(tag): |
|
# Keep Ollama's progress display and let a large download take its time. |
|
subprocess.run(["ollama", "pull", tag], env=child_env({}), check=True) |
|
|
|
def agent_command(agent, alias, context, extras): |
|
forbidden = {"--model", "--local-provider", "--oss", "--base-url", "--api-key", |
|
"--provider", "--endpoint", "--fallback-model", "--settings", |
|
"--setting-sources", "--model-provider", "--config", "--agents", |
|
"--profile", "--remote", "--remote-auth-token-env"} |
|
if any(x.split("=", 1)[0] in forbidden for x in extras): |
|
raise ValueError("extra arguments cannot override local provider/model routing") |
|
if agent == "codex": |
|
# Let Ollama's maintained integration own provider setup. Codex's |
|
# native OSS adapter is intentionally limited to compatible reasoning |
|
# models; arbitrary completion/tool models can look launched but then |
|
# reconnect or fall back to the user's hosted account. |
|
return (["ollama", "launch", "codex", "--model", alias, "--"] + extras, {}) |
|
if agent == "claude": |
|
# Ollama ships and maintains the Claude Code adapter. It configures |
|
# the Anthropic-compatible endpoint and avoids Claude Code's private |
|
# model-catalog assumptions about synthetic local names. |
|
return (["ollama", "launch", "claude", "--model", alias, "--"] + extras, {}) |
|
if agent == "aider": |
|
return (["aider", "--model", f"ollama_chat/{alias}"] + extras, |
|
{"OLLAMA_API_BASE": ENDPOINT, "OLLAMA_API_KEY": "ollama"}) |
|
if agent == "claudish": |
|
# Claudish adapts Ollama's OpenAI-compatible interface to Claude |
|
# Code's harness. Transport support does not make every model a good |
|
# tool user; use_flow labels that distinction before launch. |
|
return (["claudish", "--model", f"ollama@{alias}"] + extras, |
|
harness_env("claudish")) |
|
if agent == "opencode": |
|
return (["ollama", "launch", "opencode", "--model", alias, "--"] + extras, |
|
harness_env("opencode")) |
|
if agent == "goose": |
|
return (["goose"] + extras, {**harness_env("goose"), "GOOSE_PROVIDER": "ollama", |
|
"GOOSE_MODEL": alias, "GOOSE_PROVIDER__HOST": ENDPOINT, |
|
"GOOSE_PROVIDER__API_KEY": "ollama"}) |
|
return (["ollama", "run", alias] + extras, {}) |
|
|
|
def child_env(changes): |
|
env = os.environ.copy() |
|
env.update(changes) |
|
env["OLLAMA_HOST"] = ENDPOINT |
|
for key in ("NO_PROXY", "no_proxy"): |
|
values = [x for x in env.get(key, "").split(",") if x] |
|
env[key] = ",".join(dict.fromkeys(values + ["127.0.0.1", "localhost"])) |
|
return env |
|
|
|
def check_ready(version): |
|
if not version: |
|
raise RuntimeError("Ollama is unavailable at 127.0.0.1:11434. Start it with: ollama serve") |
|
|
|
def _align_table(options, header): |
|
"""Pad compact pipe-delimited labels so fzf and fallback menus line up.""" |
|
lines = (header or "").splitlines() |
|
if not lines or "|" not in lines[0]: |
|
return options, header |
|
header_cols = [part.strip() for part in lines[0].split("|")] |
|
cells = [[str(key)] + [part.strip() for part in str(label).split("|")] |
|
for key, label in options] |
|
count = max([len(header_cols)] + [len(row) for row in cells]) |
|
header_cols += [""] * (count - len(header_cols)) |
|
cells = [row + [""] * (count - len(row)) for row in cells] |
|
widths = [max([len(header_cols[i])] + [len(row[i]) for row in cells]) for i in range(count)] |
|
limits = [4, 27, 11, 20, 14, 20] + [20] * count |
|
widths = [min(widths[i], limits[i] if i < len(limits) else 20) for i in range(count)] |
|
available = max(60, shutil.get_terminal_size((100, 20)).columns - 8) |
|
while sum(widths) + 3 * (count - 1) > available and max(widths[1:], default=0) > 8: |
|
index = max(range(1, count), key=lambda i: widths[i]) |
|
widths[index] -= 1 |
|
|
|
def trim(value, width): |
|
value = str(value) |
|
return value if len(value) <= width else value[:max(1, width - 1)] + "…" |
|
|
|
def line(parts, start=0): |
|
return " | ".join(trim(parts[i], widths[i]).ljust(widths[i]) |
|
for i in range(start, count)).rstrip() |
|
|
|
aligned = [(row[0], line(row, 1)) for row in cells] |
|
return aligned, "\n".join([line(header_cols)] + lines[1:]) |
|
|
|
|
|
def choose(prompt, options, header=None, table=False): |
|
"""Choose from (stable key, visible description) rows without interpreting shell text.""" |
|
if not sys.stdin.isatty(): |
|
raise RuntimeError("menus need a terminal; use --list, --scan, or --model NAME:TAG --agent claude") |
|
header = header or "Enter selects / Esc cancels" |
|
if table: |
|
options, header = _align_table(options, header) |
|
key_width = max((len(str(key)) for key, _ in options), default=1) if table else 0 |
|
rows = [f"{str(key).ljust(key_width)}\t{label}" if table else f"{key}\t{label}" |
|
for key, label in options] |
|
if shutil.which("fzf"): |
|
env = os.environ.copy() |
|
for key in ("FZF_DEFAULT_OPTS", "FZF_DEFAULT_COMMAND", "FZF_DEFAULT_OPTS_FILE"): |
|
env.pop(key, None) |
|
result = subprocess.run( |
|
["fzf", "--no-multi", "--no-sort", "--height=70%", "--layout=reverse", |
|
"--border", "--header", header, "--prompt", prompt + ": "], |
|
input="\n".join(rows) + "\n", text=True, stdout=subprocess.PIPE, env=env) |
|
if result.returncode in (1, 130): |
|
return None |
|
if result.returncode: |
|
raise RuntimeError("fzf could not open its menu; check that this is an interactive terminal") |
|
selected = result.stdout.strip().split("\t", 1)[0].strip() |
|
return selected if selected in {key for key, _ in options} else None |
|
print(header) |
|
print(prompt) |
|
for i, (key, label) in enumerate(options, 1): |
|
display_key = str(key).ljust(key_width) if table else str(key) |
|
print(f" {i}) {display_key} | {label}" if table else f" {i}) {key}: {label}") |
|
try: |
|
n = int(input("Choose (blank cancels): ").strip() or "0") |
|
except (EOFError, ValueError): |
|
return None |
|
return options[n - 1][0] if 1 <= n <= len(options) else None |
|
|
|
def command_output(command): |
|
try: |
|
result = subprocess.run(command, capture_output=True, text=True, timeout=5, |
|
env=child_env({})) |
|
return (result.stdout or result.stderr).strip() if result.returncode == 0 else "unavailable" |
|
except (OSError, subprocess.TimeoutExpired): |
|
return "unavailable" |
|
|
|
def hardware_snapshot(override=None): |
|
device = hardware.host() |
|
weight_budget = hardware.budget(device, override) |
|
print(hardware.describe(device, weight_budget)) |
|
if device.get("swap"): |
|
print("Swap: " + device["swap"]) |
|
try: |
|
print(f"Disk free ({model_dir()}): {free_bytes(model_dir()) / 1024**3:.1f} GiB") |
|
except OSError as exc: |
|
print(f"Disk free: unavailable ({exc})") |
|
return device, weight_budget |
|
|
|
def doctor(device=None, weight_budget=None): |
|
version, tags, err = server_info() |
|
print("Ollama endpoint: 127.0.0.1:11434") |
|
print("Server: " + (version if version else f"unavailable ({err})")) |
|
print("Installed models: " + (", ".join(sorted(model_names(tags))) or "none")) |
|
if device is None: |
|
hardware_snapshot() |
|
for binary in ("ollama", "claudish", "opencode", "goose", "aider", "claude", "codex", "fzf"): |
|
location = executable(binary) |
|
print(f"{binary}: {command_output([location, '--version']) if location else 'not installed'}") |
|
if version: |
|
try: |
|
running = api("/api/ps").get("models", []) |
|
print("Running models: " + (", ".join(sorted(model_names(running))) or "none")) |
|
except (OSError, urllib.error.URLError, ValueError, RuntimeError): |
|
print("Running models: unavailable") |
|
print("Runtime updates: https://github.com/ollama/ollama/releases/latest") |
|
print("New model architectures may require updating Ollama and restarting its server.") |
|
|
|
def validate_tag(tag): |
|
if not re.fullmatch(r"(?:[A-Za-z0-9][A-Za-z0-9_.-]*/)?[A-Za-z0-9][A-Za-z0-9_.-]*(?::[A-Za-z0-9][A-Za-z0-9_.-]*)?", tag): |
|
raise ValueError("use an Ollama model name or namespace/name:tag; URLs and file paths are not model names") |
|
if "cloud" in tag.rsplit(":", 1)[-1].lower(): |
|
raise ValueError("this launcher runs local models; use --free to explore hosted alternatives") |
|
return tag if ":" in tag else tag + ":latest" |
|
|
|
|
|
def reject_remote(details): |
|
if details.get("remote") or details.get("remote_model") or details.get("remote_host"): |
|
raise RuntimeError("this model delegates inference to a remote service; local launch refused") |
|
|
|
|
|
def launch(args, extras, budget): |
|
tag = validate_tag(args.model) |
|
context = args.context or 32768 |
|
agent = args.agent or "claude" |
|
cmd, changes = agent_command(agent, tag, context, extras) |
|
if args.dry_run: |
|
print("DRY RUN: no network, cache writes, downloads, alias creation, or launch.") |
|
print(f"Would inspect {tag}, verify local weights, and check disk space before a missing-model download.") |
|
print("argv:", shlex.join(["ollama", "pull", tag] if args.pull else cmd)) |
|
if not args.pull: |
|
print("Working directory:", os.getcwd()) |
|
print("env overrides:", shlex.join([f"{key}={value}" for key, value in {"OLLAMA_HOST": ENDPOINT, **changes}.items()])) |
|
return 0 |
|
for binary in ({"ollama"} if args.pull else {"ollama", cmd[0]}): |
|
if not executable(binary): |
|
raise RuntimeError(f"{binary} is not installed/on PATH; use --doctor or --dry-run") |
|
version, tags, _ = server_info() |
|
check_ready(version) |
|
installed = next((row for row in tags if (row.get("name") or row.get("model")) == tag), None) |
|
if installed: |
|
details = api("/api/show", "POST", {"model": tag}) |
|
reject_remote(details) |
|
size = installed.get("size") |
|
else: |
|
details = catalog.manifest(tag, refresh=True) |
|
reject_remote(details) |
|
size = details.get("size_bytes") |
|
if not size or not details.get("weight_bytes"): |
|
raise RuntimeError("manifest does not identify downloadable local weights; download manually and inspect the model") |
|
print(f"Model: {tag} | {size / 10**9:.2f} GB | {hardware.fit(size, budget)}" if size else f"Model: {tag} | size unknown") |
|
if not args.pull: |
|
print(f"Context: {context:,} tokens; actual context/runtime overhead must be checked with ollama ps.") |
|
if not installed: |
|
free, required = need_space(size) |
|
print(f"Disk now: {free / hardware.GIB:.1f} GiB free on {model_dir()} " |
|
f"(need {required / hardware.GIB:.1f} GiB including 3 GiB buffer)") |
|
if free < required: |
|
raise RuntimeError(f"not enough disk space right now: need {required / hardware.GIB:.1f} GiB including buffer; " |
|
f"have {free / hardware.GIB:.1f} GiB on {model_dir()}") |
|
if not args.yes: |
|
if not sys.stdin.isatty(): |
|
raise RuntimeError("model is missing; rerun interactively or use --yes to consent to download") |
|
if input(f"Download {tag} (~{size / 10**9:.2f} GB)? [y/N] ").strip().lower() not in ("y", "yes"): |
|
return 0 |
|
pull(tag) |
|
try: |
|
print(f"Disk after download: {free_bytes(model_dir()) / hardware.GIB:.1f} GiB free") |
|
except OSError: |
|
pass |
|
if args.pull: |
|
print(f"Ready: {tag}. Already-installed tags were left unchanged; use ollama pull explicitly to update weights.") |
|
return 0 |
|
details = api("/api/show", "POST", {"model": tag}) |
|
reject_remote(details) |
|
caps = details.get("capabilities") |
|
if agent != "chat" and caps and "tools" not in caps: |
|
raise RuntimeError("this installed model does not advertise tool support; choose --agent chat or another model") |
|
maxima = [value for key, value in details.get("model_info", {}).items() |
|
if key.endswith(".context_length") and isinstance(value, int)] |
|
if maxima and context > max(maxima): |
|
raise RuntimeError(f"requested context exceeds advertised limit {max(maxima):,}; pass a smaller --context") |
|
# Use the installed tag directly. Synthetic wrappers previously caused |
|
# local-code-local-code-* aliases to accumulate on every launch. |
|
if agent == "codex" and "thinking" not in (details.get("capabilities") or []): |
|
raise RuntimeError("Codex's Ollama adapter requires a thinking-capable model; " |
|
"use gpt-oss with Codex, or choose Aider/Chat for this model") |
|
cmd, changes = agent_command(agent, tag, context, extras) |
|
print(f"Launching {agent} in {os.getcwd()} using {tag}", flush=True) |
|
os.execvpe(cmd[0], cmd, child_env(changes)) |
|
return 0 |
|
|
|
|
|
def print_external(items, kind): |
|
for item in items: |
|
if kind == "free": |
|
print(f"{item['id']} | ctx {item.get('context')} | tools {item.get('tools')} | {item['url']}") |
|
else: |
|
print(f"{item['id']} | downloads {item.get('downloads')} | updated {item.get('updated')} | {item['url']}") |
|
print("Discovery only: popularity and zero listed token prices do not establish coding quality, availability, or unlimited access.") |
|
|
|
|
|
def install_flow(args, device, budget): |
|
"""Recommend, explain, install, and return to this menu.""" |
|
while True: |
|
print("\nInstall new models") |
|
print("This host: " + hardware.describe(device, budget)) |
|
print("Recommendations use Arena text-overall ratings plus Ollama local-size/tool metadata.") |
|
comfort = hardware.comfortable_percent() |
|
print(f"Fit labels: comfortable (<= {comfort:.0f}% of conservative budget), " |
|
f"near limit ({comfort:.0f}–100%), too large (>100%).") |
|
live_disk = free_bytes(model_dir()) |
|
print(f"Disk available right now: {live_disk / hardware.GIB:.1f} GiB (model download requires a 3 GiB buffer).") |
|
print("Checking the current Ollama catalog and benchmark source...", flush=True) |
|
items = explore.recommend(budget, device, limit=5, refresh=args.refresh, |
|
disk_available=live_disk) |
|
if items: |
|
print("\nFive highest-rated benchmark matches within your comfortable threshold:") |
|
else: |
|
print("\nNo benchmark-matched local candidates were found; browse the live catalog instead.") |
|
options = explore.recommendation_rows(items, budget) |
|
options.append(("browse", "Browse all local models | — | — | — | search by family or capability")) |
|
selected = choose("Install", options, |
|
header=("# | Model | Download | Benchmark | Fit | Role\n" |
|
f"Fit: comfortable <= {comfort:.0f}% budget; near limit {comfort:.0f}-100%; too large >100%"), |
|
table=True) |
|
if selected is None: |
|
return 0 |
|
if selected == "browse": |
|
item = explore.helpful_browse(choose, budget, args.refresh, live_disk) |
|
if not item: |
|
continue |
|
else: |
|
item = items[int(selected) - 1] |
|
explore.show_model(item, budget, device) |
|
action = choose("Model", [("install", "Install this model and return to Install"), |
|
("back", "Back to the five recommendations")]) |
|
if action != "install": |
|
continue |
|
if item.get("cloud"): |
|
print("That is a hosted variant, not a local download. Pick another variant.") |
|
continue |
|
args.model, args.pull = item["tag"], True |
|
launch(args, [], budget) |
|
args.model, args.pull = None, False |
|
print(f"\nInstalled {item['tag']}. Returning to Install.") |
|
|
|
|
|
def _installed_items(tags, budget): |
|
return [{"tag": row.get("name") or row.get("model"), "size_bytes": row.get("size"), |
|
"context": None, "format": "installed"} for row in tags |
|
if row.get("name") or row.get("model")] |
|
|
|
def _launch_items(tags, budget): |
|
"""Hide legacy profiles made by older local-code versions from Use.""" |
|
items = _installed_items(tags, budget) |
|
base = [item for item in items if not item["tag"].lower().startswith("local-code-")] |
|
return base or items |
|
|
|
|
|
def remove_flow(budget): |
|
"""Interactively remove one explicitly selected local Ollama model.""" |
|
while True: |
|
version, tags, _ = server_info() |
|
check_ready(version) |
|
items = _installed_items(tags, budget) |
|
if not items: |
|
print("No local models installed.") |
|
return 0 |
|
selected = choose("Remove", explore.variant_rows(items, budget), |
|
header="# | Model | Download | Fit | Context | Format | Updated", table=True) |
|
if selected is None: |
|
return 0 |
|
item = items[int(selected)] |
|
size = item.get("size_bytes") |
|
size_text = f"{size / 10**9:.1f} GB" if size else "unknown size" |
|
if input(f"Remove {item['tag']} ({size_text}) permanently from Ollama? [y/N] ").strip().lower() not in ("y", "yes"): |
|
continue |
|
if not executable("ollama"): |
|
raise RuntimeError("ollama is not installed/on PATH") |
|
subprocess.run(["ollama", "rm", item["tag"]], env=child_env({}), check=True) |
|
try: |
|
print(f"Removed {item['tag']}. Disk now: {free_bytes(model_dir()) / hardware.GIB:.1f} GiB free.") |
|
except OSError: |
|
print(f"Removed {item['tag']}.") |
|
|
|
|
|
def use_flow(args, extras, budget): |
|
"""Launch or remove installed models, preserving arbitrary agent flags.""" |
|
version, tags, _ = server_info() |
|
check_ready(version) |
|
while True: |
|
mode = choose("Use", [("launch", "Launch an installed model"), |
|
("remove", "Remove an installed model")]) |
|
if mode is None: |
|
return 0 |
|
if mode == "remove": |
|
if extras: |
|
raise ValueError("agent flags cannot be used with Remove") |
|
remove_flow(budget) |
|
continue |
|
version, tags, _ = server_info() |
|
check_ready(version) |
|
items = _launch_items(tags, budget) |
|
if not items: |
|
print("No local models installed yet. Choose Install new models first.") |
|
continue |
|
selected = choose("Use", explore.variant_rows(items, budget), |
|
header="# | Model | Download | Fit | Context | Format | Updated", table=True) |
|
if selected is None: |
|
continue |
|
item = items[int(selected)] |
|
details = api("/api/show", "POST", {"model": item["tag"]}) |
|
caps = details.get("capabilities") or [] |
|
tools = "tools" in caps |
|
tool_note = "advertises tools; reliability varies by model" if tools else "not offered: model does not advertise tools" |
|
available = [ |
|
("claudish", f"Claudish / Claude Code | {tool_note}") if executable("claudish") and tools else None, |
|
("opencode", f"OpenCode | {tool_note}") if executable("opencode") and tools else None, |
|
("goose", f"Goose | {tool_note}") if executable("goose") and tools else None, |
|
("aider", f"Aider | {tool_note}") if executable("aider") and tools else None, |
|
("claude", f"Claude Code via Ollama | {tool_note}") if executable("ollama") and tools else None, |
|
("codex", "Codex via Ollama | thinking + tools advertised") |
|
if executable("ollama") and tools and "thinking" in caps else None, |
|
("chat", "Plain Ollama chat | compatible; no coding tools"), |
|
] |
|
agent = choose("Launch in", [row for row in available if row]) |
|
if not agent: |
|
continue |
|
selected_extras = list(extras) |
|
if not selected_extras: |
|
example = ("-y" if agent == "claudish" |
|
else "--dangerously-skip-permissions" if agent == "claude" |
|
else "--yolo" if agent == "codex" else "--help") |
|
typed = input(f"Extra {agent} flags (blank for defaults; e.g. {example}): ").strip() |
|
if typed: |
|
try: |
|
selected_extras = shlex.split(typed) |
|
except ValueError as exc: |
|
print(f"Could not parse flags: {exc}. Try again.") |
|
continue |
|
break |
|
args.model, args.agent, args.pull = item["tag"], agent, False |
|
args.context = args.context or 32768 |
|
if selected_extras: |
|
print("Passing through: " + shlex.join(selected_extras)) |
|
return launch(args, selected_extras, budget) |
|
|
|
|
|
def interactive(args, extras, device, budget): |
|
while True: |
|
action = choose("local-code", [("install", "Install new models"), |
|
("use", "Use installed models"), |
|
("metrics", "Metrics and diagnostics"), |
|
("settings", "Settings")]) |
|
if action is None: |
|
return 0 |
|
if action == "install": |
|
if extras: |
|
raise ValueError("agent flags after -- belong in Use installed models, not Install") |
|
install_flow(args, device, budget) |
|
elif action == "use": |
|
use_flow(args, extras, budget) |
|
elif action == "metrics": |
|
doctor(device, budget) |
|
elif action == "settings": |
|
settings_flow() |
|
|
|
|
|
def main(argv=None): |
|
raw = list(sys.argv[1:] if argv is None else argv) |
|
extras = [] |
|
if "--" in raw: |
|
split = raw.index("--") |
|
raw, extras = raw[:split], raw[split + 1:] |
|
ap = argparse.ArgumentParser(description="Explore current local models, compare hardware fit, and launch local coding harnesses.", |
|
epilog="Pass agent arguments after --. No model names or device presets are baked in.") |
|
actions = ap.add_mutually_exclusive_group() |
|
actions.add_argument("--list", action="store_true", help="discover current Ollama families") |
|
actions.add_argument("--scan", action="store_true", help="compare variants across current model families") |
|
actions.add_argument("--variants", metavar="FAMILY", help="all available tags, sizes, and fit labels") |
|
actions.add_argument("--inspect", metavar="MODEL", help="inspect exact registry manifest sizes") |
|
actions.add_argument("--free", action="store_true", help="live free hosted catalog; reference only") |
|
actions.add_argument("--hf", metavar="QUERY", help="search trending GGUF repositories; use '' for all") |
|
actions.add_argument("--installed", action="store_true", help="list all installed local models") |
|
actions.add_argument("--doctor", action="store_true", help="detect host and runtime setup") |
|
actions.add_argument("--install", action="store_true", help="interactive install flow") |
|
actions.add_argument("--use", action="store_true", help="interactive installed-model launcher") |
|
actions.add_argument("--metrics", action="store_true", help="host/runtime metrics and diagnostics") |
|
actions.add_argument("--pull", action="store_true", help="download --model without launching") |
|
ap.add_argument("--query", default="", help="search text for family discovery") |
|
ap.add_argument("--sort", choices=("newest", "popular"), default="newest") |
|
ap.add_argument("--limit", type=int, default=20, help="maximum families inspected by --scan (default 20)") |
|
ap.add_argument("--include-nontools", action="store_true", help="scan even without an advertised tools badge") |
|
ap.add_argument("--fit-only", action="store_true", help="hide hosted/over-budget/unknown-size variants") |
|
ap.add_argument("--refresh", action="store_true", help="force current catalog fetch; never update model weights") |
|
ap.add_argument("--budget-gib", type=float, help="override heuristic model-weight budget for this host") |
|
ap.add_argument("--comfortable-percent", type=float, |
|
help="one-off comfortable-fit cutoff (1-99); persistent value: local-code settings") |
|
ap.add_argument("--dry-run", action="store_true", help="preview --model launch without network or writes") |
|
ap.add_argument("--model", help="any Ollama model or namespace/name:tag; no fixed shortlist") |
|
ap.add_argument("--agent", choices=("aider", "claudish", "opencode", "goose", "claude", "codex", "chat"), default=None) |
|
ap.add_argument("--context", type=int, help="positive context token count; default 32768") |
|
ap.add_argument("--yes", action="store_true", help="consent to a missing-model download") |
|
ap.add_argument("command", nargs="?", choices=("settings",), help="edit persistent launcher settings") |
|
args = ap.parse_args(raw) |
|
if args.limit < 1 or args.limit > 500: |
|
ap.error("--limit must be between 1 and 500") |
|
if args.budget_gib is not None and not (0 < args.budget_gib < 100000): |
|
ap.error("--budget-gib must be a finite positive number") |
|
if args.comfortable_percent is not None and not 1 <= args.comfortable_percent < 100: |
|
ap.error("--comfortable-percent must be between 1 and 99") |
|
settings = load_settings() |
|
if os.environ.get("LOCAL_CODE_COMFORTABLE_PERCENT") is None: |
|
hardware.set_comfortable_fraction(settings["comfortable_percent"] / 100) |
|
if args.comfortable_percent is not None: |
|
hardware.set_comfortable_fraction(args.comfortable_percent / 100) |
|
if args.command == "settings": |
|
return settings_flow() |
|
if args.context is not None and not 512 <= args.context <= 2097152: |
|
ap.error("--context must be between 512 and 2097152") |
|
if args.pull and (not args.model or args.agent or extras): |
|
ap.error("--pull requires --model and cannot be combined with agent arguments") |
|
if args.dry_run: |
|
if not args.model: |
|
ap.error("--dry-run needs --model NAME:TAG; choose a model with --list or --scan") |
|
return launch(args, extras, None) |
|
if args.doctor: |
|
doctor() |
|
return 0 |
|
if args.free: |
|
print_external(catalog.free_models(refresh=args.refresh), "free") |
|
return 0 |
|
if args.hf is not None: |
|
print_external(catalog.hf_models(query=args.hf, refresh=args.refresh), "hf") |
|
return 0 |
|
device, budget = hardware_snapshot(args.budget_gib) |
|
if args.model: |
|
return launch(args, extras, budget) |
|
if (extras or args.agent or args.context) and not args.use: |
|
ap.error("launch options require --model") |
|
if args.inspect: |
|
explore.inspect(args.inspect, args.refresh) |
|
elif args.installed: |
|
version, tags, _ = server_info() |
|
check_ready(version) |
|
for row in tags: |
|
size = row.get("size") |
|
print(f"{row.get('name') or row.get('model')} | " |
|
f"{size / 10**9:.2f} GB | {hardware.fit(size, budget)}" if size else str(row)) |
|
return 0 |
|
elif args.list: |
|
for _, row in explore.family_rows(catalog.families(args.query, args.sort, args.refresh)): |
|
print(row) |
|
elif args.scan or args.variants: |
|
items = (catalog.variants(args.variants, args.refresh) if args.variants else |
|
explore.scan(args.query, args.sort, args.limit, budget, args.refresh, args.include_nontools)) |
|
if args.fit_only: |
|
items = [row for row in items if not row.get("cloud") and |
|
hardware.fit(row.get("size_bytes"), budget) in ("headroom", "tight")] |
|
for _, row in explore.variant_rows(items, budget): |
|
print(row) |
|
elif args.install: |
|
return install_flow(args, device, budget) |
|
elif args.use: |
|
return use_flow(args, extras, budget) |
|
elif args.metrics: |
|
doctor(device, budget) |
|
return 0 |
|
else: |
|
return interactive(args, extras, device, budget) |
|
explore.status() |
|
return 0 |
|
|
|
if __name__ == "__main__": |
|
try: |
|
raise SystemExit(main()) |
|
except (KeyboardInterrupt, EOFError): |
|
print("\nCancelled.", file=sys.stderr) |
|
raise SystemExit(130) |
|
except (RuntimeError, ValueError, OSError, urllib.error.URLError, subprocess.SubprocessError) as exc: |
|
print(f"error: {exc}", file=sys.stderr) |
|
raise SystemExit(2) |
