ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

CLI: Rename unsloth connect to unsloth start (#6613)

Browse source 
* replaced connect with start

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix

* Studio: build the coding-agent command from the selected server

The API keys panel showed a hardcoded `unsloth start claude`. `unsloth start`
defaults to 127.0.0.1:8888 and only mints a key for a loopback server, so a
non-default port or a tunnel/remote base would target the wrong server or fail
to mint. Build the command from the panel base/key (and emit a key for
non-loopback), matching the other snippets in the panel.

* CLI: keep `unsloth connect` as a hidden alias for `unsloth start`

Avoids breaking existing scripts and docs that still call `unsloth connect`.

* Tests: stub _unstarted_cleanup in same-task disconnect test

The test builds _SameTaskStreamingResponse via __new__, so set the attribute
that __call__ now reads.

* Match coding-agent command loopback check to the CLI 127.0.0.0/8 rule (#6613)

* Keep unsloth_cli.commands.connect importable as a deprecated shim (#6613)

* Format the new coding-agents panel strings and import per biome (#6613)

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Drop the unsloth connect alias and shim; unsloth start is the only command (#6613)

* Route unsloth connect to unsloth start as a hidden backward-compatible alias (#6613)

* Forward unsloth run model-load flags to unsloth start (gguf-variant, context-length, load-in-4bit, tensor-parallel) (#6613)

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Session-scope coding agent config in unsloth start

Configure each agent for the current session instead of writing the Studio endpoint, key, and default model into the user's own config. Codex, OpenCode, OpenClaw, and Hermes get a private config relocated through their config-path env vars (CODEX_HOME, OPENCODE_CONFIG overlay, OPENCLAW_CONFIG_PATH plus OPENCLAW_STATE_DIR, HERMES_HOME). Claude Code suppresses the attribution header for the session via the CLAUDE_CODE_ATTRIBUTION_HEADER env var plus a --settings overlay, with no ~/.claude write. --launch uses an ephemeral temp dir removed after the agent exits; --no-launch uses a stable Unsloth-owned dir and prints the matching export lines.

* Read relocated agent session config in Local Agent Guides CI

The contract crosscheck and the openclaw/hermes patch helpers now read each agent's config from the relocated path printed by unsloth start --no-launch (CODEX_HOME, OPENCODE_CONFIG, OPENCLAW_CONFIG_PATH, HERMES_HOME) instead of fixed home paths. The Claude attribution A/B toggles the header for the session only (shipped-config HIT vs vanilla MISS) instead of editing ~/.claude/settings.json.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Skip the POSIX-only --no-launch parser test on Windows

test_no_launch_output_is_parseable mirrors the #6547 bash CI parser, which greps export/unset lines and only runs on Linux/macOS runners. On Windows --no-launch prints PowerShell ($env: / Remove-Item), so the export-line assertion does not apply there. Cross-OS staging CI surfaced this.

* Size Claude Code's auto-compact window to the loaded model's context

Claude Code auto-compacts against its native (~600k token) window, so against a smaller local model it overflows the server's context (silent truncation) long before it compacts. Set CLAUDE_CODE_AUTO_COMPACT_WINDOW to the loaded model's real context length (the value codex/openclaw already get via model_context_window / contextWindow). Omitted when the model reports no context length.

* Pin OpenCode/Hermes context window and set 90% compaction across agents

Feed every agent the server-determined sequence length (the value /v1/models reports from runtime_context_length) and a ~90% compaction threshold. OpenCode: a custom-provider model with no limit defaults to context 0, which silently disables auto-compaction, so set limit.context/output and scale the compaction buffer to 10% of the window. Hermes: pin model.context_length (it otherwise falls back to a 256k default when the server's /v1/models omits the field) and set compression.threshold 0.9. Claude: add CLAUDE_AUTOCOMPACT_PCT_OVERRIDE=90 alongside the window. Codex (model_context_window) and OpenClaw (contextWindow) already carried the window and auto-manage off it.

* Add `unsloth start pi` recipe

Pi was the only agent without a built-in recipe, so the agent-guides CI
hand-wrote ~/.pi/agent/models.json. Add a first-class `pi` command mirroring
the others:

- write_pi_config writes the session-scoped OpenAI-compatible provider config
  (key in the config, like openclaw/opencode).
- pi() launches `pi --provider unsloth --model <id>` (Pi defaults to the google
  provider, so the provider/model are pinned on the command line) with HOME
  relocated for the session. Pi has no config-dir env var and resolves ~/.pi off
  $HOME, so HOME-scoping keeps the user's ~/.pi untouched.

Migrate the agent-guides CI off the hand-written config onto the
`unsloth start pi --no-launch` path (connection + file-edit), with a crosscheck
for the provider api, so the documented recipe is exercised.

* Harden unsloth start for Windows and WSL agent launches

Address the Codex review on PR 6613:
- write_pi_config now pins the loaded contextWindow and a sane maxTokens so Pi
  compacts instead of overflowing a small Studio context (it otherwise assumes
  its 128000 default), matching the other agents.
- pi() sets USERPROFILE (and HOMEDRIVE/HOMEPATH when present) alongside HOME on
  native Windows, where Node resolves ~/.pi via USERPROFILE rather than HOME, so
  the session no longer reads or writes the user's real ~/.pi.
- The WSLENV bridge flags path-valued vars with /p so a Windows npm shim under
  /mnt receives translated paths, while scalar vars (the numeric context window)
  pass through untranslated. WSLENV is deduped on the bare name.
- _print_env prints the launch command with PowerShell-safe quoting so the inline
  --settings JSON survives copy-paste on native Windows --no-launch.

Add tests for the WSLENV path flagging, PowerShell quoting, the Pi context
window, and the Pi USERPROFILE relocation.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Set CLAUDE_CODE_NO_FLICKER for the Claude session

A local server streams in bursts, so Claude Code's full-screen TUI redraw
flickers between tokens. Disable it for the session via CLAUDE_CODE_NO_FLICKER,
alongside the other CLAUDE_CODE_* session env knobs.

* Add a normalized --yolo flag routed to each agent's auto-approve mode

It is easy to forget which agent spells "run tools without prompting" which way,
so `unsloth start` now accepts all three spellings as one option (--yolo,
--dangerously-skip-permissions, --dangerously-bypass-approvals-and-sandbox) and
routes to the agent's own mechanism:

- claude:   --dangerously-skip-permissions
- codex:    --dangerously-bypass-approvals-and-sandbox
- hermes:   --yolo
- pi:       --approve (Pi's only approval gate is project trust)
- opencode: a permission allow block in opencode.json (no CLI flag exists)
- openclaw: tools.exec security=full / ask=off / host=gateway (no CLI flag exists)

Because the option is parsed by `unsloth start`, the "wrong" spelling for an
agent still routes correctly instead of leaking through to the agent and erroring.
IS_SANDBOX is deliberately left unset for Claude so its root/sandbox safety gate
still applies. Adds routing, cross-routing, and per-config tests.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix review findings: IPv6 loopback command, pi USERPROFILE under WSL, yolo guard

From a 10-reviewer pass over the PR:

- studio/frontend agent-command.ts: normalize bracketed IPv6 hosts. URL.hostname
  returns "[::1]" for http://[::1]:8888, which never matched the "::1" loopback
  checks, so the copied command embedded the placeholder API key for a local IPv6
  server instead of the bare auto-minting command. Now [::1] is treated as loopback
  like the CLI's is_loopback_url, so the command matches the CLI contract.

- pi(): also relocate USERPROFILE (and HOMEDRIVE/HOMEPATH) when running under WSL
  against a /mnt Windows shim, not just on native Windows. Windows Node resolves
  ~/.pi via USERPROFILE, and the WSLENV bridge translates the path, so pi no longer
  falls back to the user's real ~/.pi in that case.

- _yolo_command_flags: use .get so a config-based agent (or a typo) yields no flag
  instead of a latent KeyError.

Adds tests for the WSL pi USERPROFILE relocation, the yolo unmapped-agent guard,
and that opencode/openclaw --yolo stays config-only (no argv flag).

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix round-2 review findings: WSLENV /p upgrade, agent help text

- _merge_wslenv now upgrades a user's pre-existing unflagged WSLENV entry (e.g. a
  bare HOME or USERPROFILE) to the path-translated form (HOME/p) instead of leaving
  it as-is, so a Windows agent shim under WSL receives the translated session path
  rather than the raw Linux path.
- Generalize the `unsloth start` registration help to list all six agents (was only
  "Claude Code, Codex").

Adds a test for the WSLENV unflagged-entry upgrade.

* Fix round-3 review findings: complete openclaw --yolo, refresh stale copy

- openclaw --yolo now also writes the host approvals file (exec-approvals.json with
  defaults security=full / ask=off / askFallback=full) alongside the tools.exec
  config. OpenClaw gates tool execution on both layers (the stricter wins), so the
  config alone could still leave it prompting or denying. Mirrors `openclaw
  exec-policy preset yolo`. ask=off means nothing is ever prompted, so the runtime
  socket block is unnecessary.
- Studio API panel copy: clarify that a local server auto-mints the key while a
  remote one embeds it in the command, and add pi to the swap hint.
- Local Agent Guides CI: drop the stale "pi has no start.py recipe" note now that
  all six agents are driven via `unsloth start <agent> --no-launch`.

Adds the openclaw approvals-file assertions and a no-yolo openclaw test.

* start: parse claude --version with a regex so a format change does not drop optimization flags

* start: offer to install a missing agent (prompt then run its install command)

* start: auto-start a Studio server for --model when none is running, and stop it on exit

* inference: surface an actionable message when llama-server cannot compile a tool grammar

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Fix review findings: kill the auto-started server tree on Windows; apply the tool-grammar message to the OpenAI passthrough too

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* start: split --model org/repo:variant so a running session is not evicted

`unsloth start <agent> --model org/repo:QUANT` failed against an already-running
Studio server and, worse, killed whatever model another session had loaded.

/v1/models lists a loaded GGUF under its bare repo id (e.g. unsloth/Qwen3-1.7B-GGUF),
so _resolve_model never matched the `:QUANT`-suffixed request. It then POSTed
/api/inference/load with model_path=org/repo:QUANT, which (a) Hugging Face rejects
("Repo id must use alphanumeric chars, '-', '_' or '.'") and (b) evicts the model the
other session was using, so a second 'unsloth start' in a new tmux/terminal tore down
the first. Re-running the command then attached to the now-empty server, which is why
it 'worked the second time'.

Mirror the org/repo:QUANT -> org/repo + --gguf-variant QUANT shorthand that
'unsloth run' and llama.cpp already accept, splitting it in _connect before we match or
serve. Matching now resolves against the loaded bare repo id (no spurious reload, no
eviction), and any real load uses a valid repo id plus gguf_variant. An explicit
--gguf-variant still wins; local paths and Windows drive letters pass through untouched.
The auto-serve path likewise spawns 'unsloth run --model org/repo --gguf-variant QUANT'.

* start: harden auth-key handling, codex teardown, and CI transcript redaction

Three review findings:

1. CI could leak a live key. agent-guides-drive.sh printed the raw
   'unsloth start --no-launch' transcript (which carries export UNSLOTH_API_KEY /
   ANTHROPIC_AUTH_TOKEN lines) to the Actions log on both the failure path and the
   success path before redact() ran. Add cat_redacted() and use it for those two
   prints, so the key is scrubbed on the way to the log while the on-disk file stays
   intact for the env parsing that follows.

2. Outages masqueraded as bad keys. _key_accepted caught a broad Exception and
   returned False, so a 5xx or timeout while checking a cached key looked like a
   rejection: it discarded a good key and minted extra ones (local) or reported 'no
   saved key' (remote). Only treat HTTP 401/403 as a rejection; let other errors
   propagate so a real outage surfaces.

3. Codex preflight could leave the auto-started server up. _require_gguf_for_codex
   runs after _connect may have auto-started Studio but before _run installs its
   teardown finally, so a preflight rejection (e.g. a transformers-backend model) left
   the server holding the port/GPU until the atexit backstop. Tear it down explicitly
   at the point of failure.

Tests: a 5xx on a saved key surfaces without minting; a non-GGUF codex preflight
tears down the auto-served server.

* start: fix IPv6/portless studio URLs, Pi config-dir isolation, and Pi install recipe

Four review findings:

1. Pi ignored the session config when PI_CODING_AGENT_DIR was already set. Pi's
   getAgentDir() reads process.env.PI_CODING_AGENT_DIR before falling back to
   $HOME/.pi/agent, so a value inherited from the user's shell sent Pi to their real
   config and skipped our provider/key (the HOME relocation alone was not enough). Pin
   PI_CODING_AGENT_DIR at the session's .pi/agent dir; it is path-valued so the WSL
   bridge translates it automatically.

2. Pi install hint dropped Pi's documented --ignore-scripts. Pi's README installs with
   'npm install -g --ignore-scripts @earendil-works/pi-coding-agent' and notes it needs
   no install scripts, so accepting the prompt now follows that safe recipe.

3. Auto-start ignored a portless UNSLOTH_STUDIO_URL. unsloth run binds to
   'parsed.port or 8888', so http://127.0.0.1 launched the child on 8888 but the health
   poll (and the returned base) still used port 80, stalling until the startup timeout.
   Normalize the base to host:8888 (IPv6-safe) before starting and polling.

4. API-panel command mistook IPv6 loopback for the bare default. The bare 'unsloth
   start' only probes 127.0.0.1:8888 on the IPv4 stack, so http://[::1]:8888 must carry
   an explicit UNSLOTH_STUDIO_URL. Drop ::1 from the bare-default host set while keeping
   it a loopback host (URL emitted, no key needed).

Tests: PI_CODING_AGENT_DIR is set to the session dir; _effective_base normalizes
portless/IPv6 bases; a portless UNSLOTH_STUDIO_URL auto-serves on :8888.

* start: apply fresh-review findings across CLI, CI, and the API-panel command

From a fresh multi-reviewer pass over the merged head plus the latest Codex bot review:

1. Load knobs now always consult the server. _resolve_model matched on model id alone,
   so --gguf-variant / --context-length / --no-load-in-4bit / --tensor-parallel were
   silently ignored whenever the id was already loaded (asking for UD-Q4_K_XL kept a
   Q8_0 serving). With any explicit knob the CLI defers to /api/inference/load, whose
   already-loaded dedup answers without reloading when variant and settings match, so a
   second session running the same command still attaches without evicting the first.

2. OpenCode --yolo and the session model pin now ride in OPENCODE_CONFIG_CONTENT. A
   project's own opencode.json outranks OPENCODE_CONFIG, so a repo config could silently
   override the session model and the --yolo permission block; OPENCODE_CONFIG_CONTENT
   outranks project config. The API key stays in the private file, never in printed env.

3. The --no-launch recipe's last line is a self-contained one-liner (inline VAR=value
   assignments before the command, conflicting vars blanked). People copy just the last
   line, and a bare codex/claude there ran against the user's real ~/.codex or Anthropic
   credentials with zero isolation, e.g. inheriting a pre-existing damaged ~/.codex
   state DB and blaming the recipe. The CI drive script scrubs the key from the one
   'invoking:' echo this adds.

4. The auto-serve log is 0600 and the parent handle is closed. It sat world-readable in
   the shared tempdir under a predictable name while carrying the minted sk-unsloth-
   key from the unsloth run banner.

5. _key_accepted fails with a clean message on outages. Non-auth errors (5xx, network,
   timeout) surfaced as a raw traceback; 401/403 still mean a rejected key.

6. _effective_base strips URL paths, and https loopback targets never auto-serve.
   http://127.0.0.1:8888/studio polled /studio/api/health (404) and https://127.0.0.1
   polled the wrong scheme, both spinning until the 15-minute startup timeout.

7. API-panel command: only literal 127.0.0.1:8888 earns the bare command. localhost can
   resolve to ::1, which the bare CLI never probes, so it keeps UNSLOTH_STUDIO_URL.

8. CI artifact sweep covers redacted-configs/ and agent-workdir/, not just logs/.

Tests: 125 CLI tests pass (new coverage for each fix), 156 backend tests pass, ruff
clean. Adds an unsloth connect alias regression test.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* start: hand Pi a clean screen at launch

Pi paints inline from wherever the cursor sits: its first render assumes a
clean screen instead of clearing or entering the alternate screen itself
(current Pi never emits a clear at startup). Launched under unsloth start,
that left the session starting mid-scroll beneath the connection output.
Clear the screen (click.clear, cross-platform, no-op without a TTY) right
before the Studio banner so Pi opens exactly one line down on a clean
viewport. Launch path only: --no-launch recipes and piped output are never
wiped, and alternate-screen agents are left alone.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* start: auto-override hermes' 64K context floor for small model windows

Hermes refuses to initialize when the served model's context window is
under 64,000 tokens, and a second copy of the same check rejects the
compression model mid-session. write_hermes_config previously pinned the
real window, so any small local model (e.g. 40,960) failed at startup
with manual config.yaml instructions.

For windows below the floor the recipe now claims 65,536 in
model.context_length, scales compression.threshold so compaction still
fires at 90% of the real window, and sets
auxiliary.compression.context_length to cover the mid-session check.
Windows at or above the floor keep the exact previous behavior.

* ci: install pi with --ignore-scripts, matching the start.py hint

The pi cell predates the pi recipe in start.py and still installed the
package with lifecycle scripts enabled, so CI stopped exercising the
exact command users are prompted to run. npm_retry now passes extra
flags through, the pi branch mirrors the install hint verbatim, and the
stale no-recipe comment is refreshed.

* ci: fail loudly when a relocation var is missing from connect output

The empty-string guards ran after appending /config.toml or /config.yaml,
so they could never fire: crosscheck_contract silently skipped its
contract checks and patch_hermes_tools died on the root path with a bare
traceback. Check the raw variable first and guide_fail with the real
cause.

* staging: 6613 round 6 (https elision, no-launch home reuse, auto-start key fallback)

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: shimmyshimmer <107991372+shimmyshimmer@users.noreply.github.com>
Co-authored-by: Daniel Han <danielhanchen@gmail.com>
Co-authored-by: Lee Jackson <130007945+Imagineer99@users.noreply.github.com>
Co-authored-by: Wasim Yousef Said <wasimysdev@gmail.com>
This commit is contained in:
Nilay 2026-07-03 20:47:27 +05:30 • committed by GitHub
commit b8400f40df
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 3722 additions and 1903 deletions
Download patch file Download diff file Expand all files Collapse all files

13
unsloth_cli/__init__.py
View file

@ -11,7 +11,7 @@ from importlib.metadata import version as package_version, PackageNotFoundError
from unsloth_cli.commands.train import train
from unsloth_cli.commands.inference import inference
from unsloth_cli.commands.chat import chat
from unsloth_cli.commands.connect import connect_app
from unsloth_cli.commands.start import start_app
from unsloth_cli.commands.export import export, list_checkpoints
from unsloth_cli.commands.studio import (
run as studio_run,
@ -79,9 +79,16 @@ app.command()(export)
app.command("list-checkpoints")(list_checkpoints)
app.add_typer(studio_app, name = "studio", help = "Unsloth Studio commands.")
app.add_typer(
connect_app,
start_app,
name = "start",
help = "Start a coding agent (Claude, Codex, OpenClaw, OpenCode, Hermes, Pi) against Studio.",
)
# Backwards-compatible hidden alias: `unsloth connect` routes to `unsloth start`.
app.add_typer(
start_app,
name = "connect",
help = "Connect a coding agent (Claude Code, Codex) to Studio.",
hidden = True,
help = "Deprecated alias for `unsloth start`.",
)
# Top-level `unsloth run` aliases `unsloth studio run`; same context

777
unsloth_cli/commands/connect.py
View file

@ -1,777 +0,0 @@
# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
"""`unsloth connect` — launch a coding agent against a running Studio server."""
import json
import os
import re
import shlex
import shutil
import signal
import subprocess
import urllib.error
import urllib.request
from pathlib import Path
from typing import NoReturn, Optional
import typer
from unsloth_cli._inference import (
_USER_AGENT,
_studio_token,
ensure_studio_backend_path,
find_studio_server,
is_loopback_url,
urlopen_no_redirect,
verify_studio_identity,
)
connect_app = typer.Typer(
help = "Connect a coding agent to a running Studio server.",
no_args_is_help = True,
context_settings = {"help_option_names": ["-h", "--help"]},
)
_CODEX_PROFILE = "unsloth_api"
_CODEX_ENV_KEY = "UNSLOTH_STUDIO_AUTH_TOKEN"
_HERMES_ENV_KEY = "UNSLOTH_API_KEY"
_HERMES_PROVIDER = "unsloth"
_PROVIDER_HEADER = f"[model_providers.{_CODEX_PROFILE}]"
_PASSTHROUGH = {"allow_extra_args": True, "ignore_unknown_options": True}
_CLAUDE_ENV_UNSET = ("ANTHROPIC_API_KEY", "CLAUDE_CODE_OAUTH_TOKEN")
# Shared by every agent command; only the config/env/command differ.
_MODEL_OPTION = typer.Option(
None, "--model", "-m", help = "Model for the agent; defaults to the one loaded in Studio."
)
_KEY_OPTION = typer.Option(
None,
"--api-key",
envvar = "UNSLOTH_API_KEY",
help = (
"Studio API key. For a local Studio it is minted automatically and "
"remembered per server. For a remote server, pass one with --api-key "
"(or UNSLOTH_API_KEY); it is remembered for next time."
),
)
_LAUNCH_OPTION = typer.Option(
True,
"--launch/--no-launch",
help = "--no-launch prints the env and command instead (remote shells, WSL).",
)
def _fail(message: str) -> NoReturn:
typer.echo(message, err = True)
raise typer.Exit(code = 1)
def _http_error_detail(exc: urllib.error.HTTPError) -> str:
try:
body = json.loads(exc.read().decode())
return body.get("detail") or body["error"]["message"]
except Exception:
return str(exc)
def _http_json(
method: str,
url: str,
token: str,
payload = None,
timeout = 30,
error = None,
):
"""On HTTPError: raise if `error` is None, else fail with `error` plus the server's detail."""
request = urllib.request.Request(
url,
data = None if payload is None else json.dumps(payload).encode(),
headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
"User-Agent": _USER_AGENT,
},
method = method,
)
try:
# No redirects: a 3xx would leak this bearer token to an unvetted base.
with urlopen_no_redirect(request, timeout = timeout) as response:
return json.loads(response.read().decode() or "{}")
except urllib.error.HTTPError as exc:
if error is None:
raise
_fail(f"{error}: {_http_error_detail(exc)}")
except (urllib.error.URLError, TimeoutError) as exc:
if error is None:
raise
_fail(f"{error}: {getattr(exc, 'reason', None) or exc}")
def _require_studio() -> str:
base = find_studio_server()
if base is None:
expected = os.environ.get("UNSLOTH_STUDIO_URL", "http://127.0.0.1:8888")
_fail(
f"No running Studio server found at {expected}. Start one with "
"`unsloth studio`, or point UNSLOTH_STUDIO_URL at a remote server."
)
return base
def _key_cache_path() -> Path:
ensure_studio_backend_path()
from utils.paths import auth_root
return auth_root() / "agent_api_key.json"
def _read_cache(cache: Path) -> dict:
try:
data = json.loads(cache.read_text(encoding = "utf-8"))
except Exception:
return {}
return data if isinstance(data, dict) else {}
def _server_buckets(servers: dict, base: str) -> dict:
# Normalise a server's entry to {"saved": [...], "minted": [...]}, tolerating a
# corrupt/legacy value (bare string/list -> treated as minted, behind the handshake).
entry = servers.get(base) if isinstance(servers, dict) else None
if isinstance(entry, list):
return {"saved": [], "minted": [k for k in entry if isinstance(k, str)]}
if not isinstance(entry, dict):
return {"saved": [], "minted": []}
def _strs(name: str) -> list:
value = entry.get(name)
return [k for k in value if isinstance(k, str)] if isinstance(value, list) else []
return {"saved": _strs("saved"), "minted": _strs("minted")}
def _cached_keys(cache: Path, base: str, source: str) -> list:
# Keys are scoped per server. `source` splits user-supplied --api-key keys
# ("saved", trusted for that base) from auto-minted ones ("minted", replayed
# only after the identity check). Legacy unscoped caches are ignored.
return _server_buckets(_read_cache(cache).get("servers", {}), base)[source]
def _write_private_json(path: Path, data: dict) -> None:
# O_CREAT with 0o600 so a file holding an API key is never world-readable,
# even briefly (existing files keep whatever perms the user set).
path.parent.mkdir(parents = True, exist_ok = True, mode = 0o700)
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
with os.fdopen(fd, "w") as handle:
handle.write(json.dumps(data, indent = 2) + "\n")
def _read_json_object(path: Path) -> Optional[dict]:
# {} when missing, None when it can't be parsed as an object (so the caller
# leaves a user-managed file untouched rather than clobbering it).
if not path.exists():
return {}
try:
data = json.loads(path.read_text(encoding = "utf-8"))
except (ValueError, OSError):
return None
return data if isinstance(data, dict) else None
def _subdict(parent: dict, key: str) -> dict:
child = parent.get(key)
if not isinstance(child, dict):
child = parent[key] = {}
return child
def _remember_key(cache: Path, base: str, key: str, source: str) -> None:
data = _read_cache(cache)
servers = data.get("servers")
if not isinstance(servers, dict):
servers = data["servers"] = {}
buckets = _server_buckets(servers, base)
other = "minted" if source == "saved" else "saved"
buckets[source] = ([key] + [k for k in buckets[source] if k != key])[:8]
buckets[other] = [k for k in buckets[other] if k != key] # a key has one provenance
new_entry = {"saved": buckets["saved"], "minted": buckets["minted"]}
if servers.get(base) == new_entry:
return
servers[base] = new_entry
# Collapse legacy unscoped fields.
data.pop("keys", None)
data.pop("key", None)
try:
_write_private_json(cache, data)
except OSError:
pass # worst case the next launch mints another key
def _key_accepted(base: str, key: str) -> bool:
try:
_http_json("GET", f"{base}/v1/models", key)
return True
except Exception:
return False
def _agent_api_key(base: str, explicit: Optional[str]) -> str:
cache = _key_cache_path()
if explicit:
_remember_key(cache, base, explicit, "saved")
return explicit
# Replay a key the user saved for *this exact* server first (scoped per base,
# so it only goes back there -- including a remote/SSH-tunnelled Studio whose
# secret the local handshake can't match). Skip ones the server rejects.
for key in _cached_keys(cache, base, "saved"):
if _key_accepted(base, key):
_remember_key(cache, base, key, "saved")
return key
# Beyond here we auto-mint or replay an auto-minted key. find_studio_server()
# trusts a base after only a health check, so both are limited to a loopback
# server we can cryptographically confirm is ours.
if not is_loopback_url(base):
_fail(
f"No saved API key for {base} and automatic minting only runs against "
"a local Studio. Create an API key in Studio → Settings → API and "
"pass it with --api-key (it is remembered per server), or set "
"UNSLOTH_API_KEY."
)
if not verify_studio_identity(base):
_fail(
f"Couldn't verify that {base} is your Studio (it may be running as a "
"different OS user, or another process took the port). Create an API "
"key in Studio → Settings → API and pass it with --api-key, or set "
"UNSLOTH_API_KEY."
)
# Identity verified: replay a previously auto-minted key, else mint a new one.
for key in _cached_keys(cache, base, "minted"):
if _key_accepted(base, key):
_remember_key(cache, base, key, "minted")
return key
# Self-issue a JWT (signed with the local secret) and mint a key.
token = _studio_token()
if token is None:
_fail(
"Couldn't authenticate with the Studio server automatically. Create "
"an API key in Studio → Settings → API and pass it with --api-key, "
"or set UNSLOTH_API_KEY."
)
key = _http_json(
"POST",
f"{base}/api/auth/api-keys",
token,
{"name": "Coding agents (unsloth connect)"},
error = "Couldn't create an API key",
)["key"]
_remember_key(cache, base, key, "minted")
return key
def _loaded_models(base: str, key: str) -> list:
return _http_json("GET", f"{base}/v1/models", key, error = "Couldn't list models").get("data", [])
def _resolve_model(base: str, key: str, requested: Optional[str]) -> dict:
models = _loaded_models(base, key)
match = next((m for m in models if m["id"] == requested), None)
if requested and match is None:
typer.echo(f"Loading {requested} on the Studio server (this can take a while)…")
loaded = _http_json(
"POST",
f"{base}/api/inference/load",
key,
{"model_path": requested},
timeout = 3600,
error = "Model load failed",
)
# Studio registers the model under a canonical id (resolved identifier,
# casing) that /v1/models echoes but which may differ from the path we
# passed; match on the id the load reports so we don't silently fall
# through to models[0] and connect to a different loaded model.
wanted = {requested}
if isinstance(loaded, dict):
wanted |= {loaded.get("model"), loaded.get("display_name")} - {None}
models = _loaded_models(base, key)
match = next((m for m in models if m["id"] in wanted), None)
if match is not None:
return match
if requested:
# We asked Studio to load it and it didn't surface in /v1/models; don't
# silently hand back an unrelated loaded model.
_fail(
f"Studio didn't report '{requested}' as loaded. Double-check the model "
"id, or load it from the model dropdown in the UI."
)
if not models:
_fail(
"No model is loaded in Studio. Load one from the model dropdown in "
"the UI, or pass --model <hf-id-or-path> to load it from here."
)
return models[0]
def _require_gguf_for_codex(base: str, key: str, model_id: str) -> None:
# Codex always streams, and Studio only streams /v1/responses from llama-server.
try:
status = _http_json("GET", f"{base}/api/inference/status", key)
except urllib.error.HTTPError as exc:
if exc.code == 404:
return # older server without the endpoint; don't block the launch
raise
if status.get("is_gguf"):
return
hint = model_id if "gguf" in model_id.lower() else f"{model_id}-GGUF"
_fail(
f"Codex needs a GGUF model served by llama-server, but {model_id} is on "
f"the transformers backend. Try: unsloth connect codex --model {hint}"
)
def claude_settings_path() -> Path:
return Path.home() / ".claude" / "settings.json"
def ensure_claude_attribution_header() -> None:
# The header invalidates the llama.cpp KV cache (~90% slower) and Claude
# Code only honors this setting from settings.json, not the env var.
path = claude_settings_path()
settings = {}
if path.exists():
try:
settings = json.loads(path.read_text(encoding = "utf-8"))
except (ValueError, OSError):
settings = None
if not isinstance(settings, dict):
typer.echo(
f"Warning: couldn't parse {path} — set CLAUDE_CODE_ATTRIBUTION_HEADER "
'to "0" in its "env" section yourself, or local inference will be much slower.',
err = True,
)
return
env = settings.get("env")
if not isinstance(env, dict):
env = settings["env"] = {}
if str(env.get("CLAUDE_CODE_ATTRIBUTION_HEADER")) == "0":
return
env["CLAUDE_CODE_ATTRIBUTION_HEADER"] = "0"
try:
path.parent.mkdir(parents = True, exist_ok = True)
path.write_text(json.dumps(settings, indent = 2) + "\n", encoding = "utf-8")
except OSError:
typer.echo(
f"Warning: couldn't write {path} — set CLAUDE_CODE_ATTRIBUTION_HEADER "
'to "0" in its "env" section yourself, or local inference will be much slower.',
err = True,
)
return
typer.echo(f"Disabled Claude Code's attribution header in {path} (it breaks KV-cache reuse).")
_DYNAMIC_SECTIONS_FLAG = "--exclude-dynamic-system-prompt-sections"
def _claude_cache_flags() -> list:
# The flag moves per-machine context (cwd, env info, git status) out of
# the system prompt, where it changes every session and defeats llama.cpp
# prefix caching. As of 2.1.175 it only takes effect in print mode (`-p`
# passed through ctx.args); interactive sessions accept and ignore it.
# Claude Code < 2.1.98 aborts on the unknown flag, so check the version
# first; no local binary means a --no-launch printout for another machine.
executable = shutil.which("claude")
if executable is None:
return [_DYNAMIC_SECTIONS_FLAG]
try:
result = subprocess.run(
[executable, "--version"], capture_output = True, text = True, timeout = 10
)
version = tuple(int(part) for part in result.stdout.split()[0].split("."))
except Exception:
return []
return [_DYNAMIC_SECTIONS_FLAG] if version >= (2, 1, 98) else []
def codex_home() -> Path:
return Path(os.environ.get("CODEX_HOME") or Path.home() / ".codex")
def _merge_codex_config(existing: str, base: str) -> str:
chunks = re.split(r"(?m)^(?=\[)", existing) # preamble, then one chunk per table
if not re.search(r"(?m)^\s*oss_provider\s*=", chunks[0]):
if chunks[0] and not chunks[0].endswith("\n"):
chunks[0] += "\n"
chunks[0] += f'oss_provider = "{_CODEX_PROFILE}"\n'
# Drop the provider table and any stale [model_providers.unsloth_api.*] subtables.
stale = (_PROVIDER_HEADER, _PROVIDER_HEADER[:-1] + ".")
text = "".join(c for c in chunks if not c.startswith(stale))
if not text.endswith("\n"):
text += "\n"
if not text.endswith("\n\n"):
text += "\n"
return text + (
f"{_PROVIDER_HEADER}\n"
'name = "Unsloth Studio"\n'
f"base_url = {json.dumps(base + '/v1')}\n"
f'env_key = "{_CODEX_ENV_KEY}"\n'
'wire_api = "responses"\n'
"requires_openai_auth = false\n"
)
def write_codex_config(base: str, model: dict) -> None:
home = codex_home()
home.mkdir(parents = True, exist_ok = True)
config = home / "config.toml"
existing = config.read_text(encoding = "utf-8") if config.exists() else ""
merged = _merge_codex_config(existing, base)
if merged != existing:
config.write_text(merged, encoding = "utf-8")
typer.echo(f"Updated {config}")
# oss_provider here too: codex --oss picks the provider from it, and the
# profile layer must beat a user-set value (e.g. "ollama") in config.toml.
profile_text = (
f'oss_provider = "{_CODEX_PROFILE}"\n'
f'model_provider = "{_CODEX_PROFILE}"\n'
f"model = {json.dumps(model['id'])}\n"
)
window = model.get("context_length") or model.get("max_context_length")
if window:
profile_text += f"model_context_window = {int(window)}\n"
profile = home / f"{_CODEX_PROFILE}.config.toml"
if not profile.exists() or profile.read_text(encoding = "utf-8") != profile_text:
profile.write_text(profile_text, encoding = "utf-8")
typer.echo(f"Updated {profile}")
def _wsl_windows_executable(command: list) -> Optional[str]:
if os.name == "nt" or not os.environ.get("WSL_DISTRO_NAME"):
return None
executable = shutil.which(command[0])
if executable and executable.startswith("/mnt/"):
return executable
return None
def _merge_wslenv(current: str, names: tuple) -> str:
entries = [entry for entry in current.split(":") if entry]
existing = {entry.split("/", 1)[0] for entry in entries}
entries.extend(name for name in names if name not in existing)
return ":".join(entries)
def _print_env(
env: dict,
command: list,
unset_env: tuple = (),
wsl_env_bridge: tuple = (),
) -> None:
if os.name == "nt":
for name in unset_env:
typer.echo(f"Remove-Item Env:{name} -ErrorAction SilentlyContinue")
for name, value in env.items():
# PowerShell: ` is the escape char, and $ triggers expansion inside "".
escaped = value.replace("`", "``").replace('"', '`"').replace("$", "`$")
typer.echo(f'$env:{name} = "{escaped}"')
typer.echo(subprocess.list2cmdline(command))
return
for name in unset_env:
typer.echo(f"export {name}=" if wsl_env_bridge else f"unset {name}")
for name, value in env.items():
typer.echo(f"export {name}={shlex.quote(value)}")
if wsl_env_bridge:
typer.echo(
f"export WSLENV={shlex.quote(_merge_wslenv(os.environ.get('WSLENV', ''), wsl_env_bridge))}"
)
typer.echo(shlex.join(command))
def _launch(
command: list,
env: dict,
install_hint: str,
unset_env: tuple = (),
) -> NoReturn:
executable = shutil.which(command[0])
if executable is None:
_fail(f"`{command[0]}` not found on PATH. Install it with: {install_hint}")
wsl_env_bridge = (
tuple(dict.fromkeys((*env.keys(), *unset_env))) if _wsl_windows_executable(command) else ()
)
child_env = dict(os.environ)
if wsl_env_bridge:
child_env["WSLENV"] = _merge_wslenv(child_env.get("WSLENV", ""), wsl_env_bridge)
for name in unset_env:
child_env[name] = ""
else:
for name in unset_env:
child_env.pop(name, None)
child_env.update(env)
# Ctrl+C cancels a turn inside the agent; don't let it kill this wrapper.
previous = signal.signal(signal.SIGINT, signal.SIG_IGN)
try:
code = subprocess.run([executable, *command[1:]], env = child_env).returncode
finally:
signal.signal(signal.SIGINT, previous)
# Negative returncode means killed by signal N; shells expect 128+N.
raise typer.Exit(code = code if code >= 0 else 128 - code)
def _connect(api_key: Optional[str], model: Optional[str]) -> tuple:
base = _require_studio()
key = _agent_api_key(base, api_key)
return base, key, _resolve_model(base, key, model)
def _run(
base: str,
entry: dict,
env: dict,
command: list,
*,
launch: bool,
install_hint: str,
unset_env: tuple = (),
) -> None:
typer.echo(f"Studio {base} · model {entry['id']}")
wsl_env_bridge = (
tuple(dict.fromkeys((*env.keys(), *unset_env))) if _wsl_windows_executable(command) else ()
)
if not launch:
_print_env(env, command, unset_env = unset_env, wsl_env_bridge = wsl_env_bridge)
return
_launch(command, env, install_hint = install_hint, unset_env = unset_env)
def openclaw_config_path() -> Path:
return Path.home() / ".openclaw" / "openclaw.json"
def write_openclaw_config(base: str, key: str, model: dict) -> None:
path = openclaw_config_path()
config = _read_json_object(path)
if config is None:
typer.echo(
f"Warning: couldn't parse {path} — add an 'unsloth' provider there "
"yourself, or move the file aside and re-run.",
err = True,
)
return
before = json.dumps(config, sort_keys = True)
# Studio is a generic OpenAI-compatible /v1 endpoint (the vLLM/LM Studio path).
provider_model = {"id": model["id"], "name": model["id"]}
window = model.get("context_length") or model.get("max_context_length")
if window:
provider_model["contextWindow"] = int(window)
models = _subdict(config, "models")
models.setdefault("mode", "merge")
_subdict(models, "providers")["unsloth"] = {
"baseUrl": f"{base}/v1",
"apiKey": key,
"api": "openai-completions",
"models": [provider_model],
}
# Pin a default model, else OpenClaw drops into its setup agent ("no models available").
defaults = _subdict(_subdict(config, "agents"), "defaults")
_subdict(defaults, "model")["primary"] = f"unsloth/{model['id']}"
# Unauthenticated loopback gateway: without auth.mode=none the client won't open
# the websocket. The daemon must still be started separately (`openclaw gateway`).
gateway = _subdict(config, "gateway")
gateway.setdefault("mode", "local")
_subdict(gateway, "auth").setdefault("mode", "none")
if json.dumps(config, sort_keys = True) != before:
_write_private_json(path, config)
typer.echo(f"Updated {path}")
def opencode_config_path() -> Path:
config_home = os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config"
return Path(config_home) / "opencode" / "opencode.json"
def write_opencode_config(base: str, key: str, model: dict) -> None:
path = opencode_config_path()
config = _read_json_object(path)
if config is None:
typer.echo(
f"Warning: couldn't parse {path} — add an 'unsloth' provider there "
"yourself, or move the file aside and re-run.",
err = True,
)
return
before = json.dumps(config, sort_keys = True)
config.setdefault("$schema", "https://opencode.ai/config.json")
_subdict(config, "provider")["unsloth"] = {
"npm": "@ai-sdk/openai-compatible",
"name": "Unsloth Studio",
"options": {"baseURL": f"{base}/v1", "apiKey": key},
"models": {model["id"]: {"name": model["id"]}},
}
# OpenCode selects a model by "<providerID>/<modelID>".
config["model"] = f"unsloth/{model['id']}"
if json.dumps(config, sort_keys = True) != before:
_write_private_json(path, config)
typer.echo(f"Updated {path}")
def hermes_config_path() -> Path:
return Path.home() / ".hermes" / "config.yaml"
def write_hermes_config(base: str, model: dict) -> None:
import yaml
path = hermes_config_path()
config: dict = {}
if path.exists():
try:
loaded = yaml.safe_load(path.read_text(encoding = "utf-8"))
except (yaml.YAMLError, OSError):
typer.echo(
f"Warning: couldn't parse {path} — configure the custom endpoint "
"there yourself, or move the file aside and re-run.",
err = True,
)
return
if isinstance(loaded, dict):
config = loaded
elif loaded is not None:
# Non-empty, non-mapping YAML is a user-managed file; leave it.
typer.echo(
f"Warning: couldn't parse {path} — configure the custom endpoint "
"there yourself, or move the file aside and re-run.",
err = True,
)
return
# Hermes only reads the key for a *named* custom provider (a bare
# `provider: custom` ignores it), so register it under providers.*.
_subdict(config, "model").update(
provider = f"custom:{_HERMES_PROVIDER}",
default = model["id"],
api_mode = "openai",
)
_subdict(config, "providers")[_HERMES_PROVIDER] = {
"base_url": f"{base}/v1",
"api_mode": "openai",
"key_env": _HERMES_ENV_KEY,
}
text = yaml.safe_dump(config, sort_keys = False)
if not path.exists() or path.read_text(encoding = "utf-8") != text:
path.parent.mkdir(parents = True, exist_ok = True)
path.write_text(text, encoding = "utf-8")
typer.echo(f"Updated {path}")
@connect_app.command("claude", context_settings = _PASSTHROUGH)
def claude(
ctx: typer.Context,
model: Optional[str] = _MODEL_OPTION,
api_key: Optional[str] = _KEY_OPTION,
launch: bool = _LAUNCH_OPTION,
):
"""Point Claude Code at the running Studio server and start it."""
base, key, entry = _connect(api_key, model)
model_id = entry["id"]
ensure_claude_attribution_header()
env = {
"ANTHROPIC_BASE_URL": base,
"ANTHROPIC_AUTH_TOKEN": key,
"ANTHROPIC_MODEL": model_id,
# Update checks, beta features, and other background requests either
# stall against a local server or evict the conversation from
# llama-server's KV-cache slots, so turn off everything nonessential.
"CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC": "1",
"CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS": "1",
}
command = ["claude", "--model", model_id, *_claude_cache_flags(), *ctx.args]
install_hint = (
"irm https://claude.ai/install.ps1 | iex"
if os.name == "nt"
else "curl -fsSL https://claude.ai/install.sh | bash"
)
_run(
base,
entry,
env,
command,
launch = launch,
install_hint = install_hint,
unset_env = _CLAUDE_ENV_UNSET,
)
@connect_app.command("codex", context_settings = _PASSTHROUGH)
def codex(
ctx: typer.Context,
model: Optional[str] = _MODEL_OPTION,
api_key: Optional[str] = _KEY_OPTION,
launch: bool = _LAUNCH_OPTION,
):
"""Point OpenAI Codex at the running Studio server and start it."""
base, key, entry = _connect(api_key, model)
_require_gguf_for_codex(base, key, entry["id"])
write_codex_config(base, entry)
env = {_CODEX_ENV_KEY: key}
command = ["codex", "--oss", "--profile", _CODEX_PROFILE, *ctx.args]
_run(base, entry, env, command, launch = launch, install_hint = "npm install -g @openai/codex")
@connect_app.command("openclaw", context_settings = _PASSTHROUGH)
def openclaw(
ctx: typer.Context,
model: Optional[str] = _MODEL_OPTION,
api_key: Optional[str] = _KEY_OPTION,
launch: bool = _LAUNCH_OPTION,
):
"""Point OpenClaw at the running Studio server and start it."""
base, key, entry = _connect(api_key, model)
write_openclaw_config(base, key, entry) # key lives in the config, not the env
command = ["openclaw", *ctx.args]
install_hint = (
"iwr -useb https://openclaw.ai/install.ps1 | iex"
if os.name == "nt"
else "curl -fsSL https://openclaw.ai/install.sh | bash"
)
_run(base, entry, {}, command, launch = launch, install_hint = install_hint)
@connect_app.command("opencode", context_settings = _PASSTHROUGH)
def opencode(
ctx: typer.Context,
model: Optional[str] = _MODEL_OPTION,
api_key: Optional[str] = _KEY_OPTION,
launch: bool = _LAUNCH_OPTION,
):
"""Point OpenCode at the running Studio server and start it."""
base, key, entry = _connect(api_key, model)
write_opencode_config(base, key, entry) # key lives in the config, not the env
command = ["opencode", *ctx.args]
_run(base, entry, {}, command, launch = launch, install_hint = "npm install -g opencode-ai")
@connect_app.command("hermes", context_settings = _PASSTHROUGH)
def hermes(
ctx: typer.Context,
model: Optional[str] = _MODEL_OPTION,
api_key: Optional[str] = _KEY_OPTION,
launch: bool = _LAUNCH_OPTION,
):
"""Point Hermes (Nous Research) at the running Studio server and start it."""
base, key, entry = _connect(api_key, model)
write_hermes_config(base, entry)
env = {_HERMES_ENV_KEY: key}
command = ["hermes", *ctx.args]
install_hint = (
"curl -fsSL https://raw.githubusercontent.com/NousResearch/hermes-agent"
"/main/scripts/install.sh | bash"
)
_run(base, entry, env, command, launch = launch, install_hint = install_hint)

1497
unsloth_cli/commands/start.py Normal file
View file

File diff suppressed because it is too large Load diff

954
unsloth_cli/tests/test_connect.py
View file

@ -1,954 +0,0 @@
# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
"""Tests for `unsloth connect` — config merging and launch env, no network."""
from __future__ import annotations
import json
import os
import sys
import urllib.error
from pathlib import Path
from types import SimpleNamespace
_REPO_ROOT = Path(__file__).resolve().parents[2]
if str(_REPO_ROOT) not in sys.path:
sys.path.insert(0, str(_REPO_ROOT))
import pytest
from typer.testing import CliRunner
import unsloth_cli.commands.connect as connect
BASE = "http://127.0.0.1:8888"
MODEL = {"id": "unsloth/gemma-4-26B-A4B-it-GGUF", "context_length": 131072}
# --no-launch prints shell setup as POSIX (export/unset) on Unix/WSL and
# PowerShell ($env:/Remove-Item) on native Windows; assert the host's form.
def _assert_env_set(output: str, name: str, value: str) -> None:
needle = f'$env:{name} = "{value}"' if os.name == "nt" else f"export {name}={value}"
assert needle in output, f"{needle!r} not found in:\n{output}"
def _assert_env_unset(output: str, name: str) -> None:
needle = f"Remove-Item Env:{name}" if os.name == "nt" else f"unset {name}"
assert needle in output, f"{needle!r} not found in:\n{output}"
@pytest.fixture()
def claude_settings(tmp_path, monkeypatch):
path = tmp_path / "claude" / "settings.json"
monkeypatch.setattr(connect, "claude_settings_path", lambda: path)
return path
def test_claude_settings_created_when_missing(claude_settings):
connect.ensure_claude_attribution_header()
settings = json.loads(claude_settings.read_text())
assert settings["env"]["CLAUDE_CODE_ATTRIBUTION_HEADER"] == "0"
def test_claude_settings_merge_preserves_existing(claude_settings):
claude_settings.parent.mkdir(parents = True)
claude_settings.write_text(
json.dumps({"effortLevel": "high", "env": {"CLAUDE_CODE_ENABLE_TELEMETRY": "0"}})
)
connect.ensure_claude_attribution_header()
settings = json.loads(claude_settings.read_text())
assert settings["effortLevel"] == "high"
assert settings["env"]["CLAUDE_CODE_ENABLE_TELEMETRY"] == "0"
assert settings["env"]["CLAUDE_CODE_ATTRIBUTION_HEADER"] == "0"
def test_claude_settings_already_set_untouched(claude_settings):
claude_settings.parent.mkdir(parents = True)
original = json.dumps({"env": {"CLAUDE_CODE_ATTRIBUTION_HEADER": "0"}})
claude_settings.write_text(original)
connect.ensure_claude_attribution_header()
assert claude_settings.read_text() == original
def test_claude_settings_bad_json_left_alone(claude_settings, capsys):
claude_settings.parent.mkdir(parents = True)
claude_settings.write_text("{not json")
connect.ensure_claude_attribution_header()
assert claude_settings.read_text() == "{not json"
assert "couldn't parse" in capsys.readouterr().err
def _fake_claude(monkeypatch, version_output: str) -> None:
monkeypatch.setattr(connect.shutil, "which", lambda _: "/usr/local/bin/claude")
monkeypatch.setattr(
connect.subprocess,
"run",
lambda *args, **kwargs: SimpleNamespace(stdout = version_output),
)
def test_cache_flags_passed_to_supported_claude(monkeypatch):
_fake_claude(monkeypatch, "2.1.98 (Claude Code)\n")
assert connect._claude_cache_flags() == ["--exclude-dynamic-system-prompt-sections"]
def test_cache_flags_skipped_on_old_claude(monkeypatch):
_fake_claude(monkeypatch, "2.0.14 (Claude Code)\n")
assert connect._claude_cache_flags() == []
def test_cache_flags_skipped_on_unparseable_version(monkeypatch):
_fake_claude(monkeypatch, "weird build string\n")
assert connect._claude_cache_flags() == []
def _parse_toml(text: str) -> dict:
tomllib = pytest.importorskip("tomllib")
return tomllib.loads(text)
def test_merge_codex_config_fresh():
merged = connect._merge_codex_config("", BASE)
parsed = _parse_toml(merged)
assert parsed["oss_provider"] == "unsloth_api"
provider = parsed["model_providers"]["unsloth_api"]
assert provider["base_url"] == f"{BASE}/v1"
assert provider["wire_api"] == "responses"
assert provider["requires_openai_auth"] is False
def test_merge_codex_config_replaces_stale_block():
existing = (
'model = "gpt-5"\n'
"\n"
"[model_providers.unsloth_api]\n"
'base_url = "http://old-host:9999/v1"\n'
'wire_api = "chat"\n'
"\n"
"[model_providers.unsloth_api.http_headers]\n"
'x-old = "1"\n'
"\n"
"[model_providers.ollama]\n"
'base_url = "http://localhost:11434/v1"\n'
)
merged = connect._merge_codex_config(existing, BASE)
parsed = _parse_toml(merged)
assert parsed["model"] == "gpt-5"
assert parsed["model_providers"]["unsloth_api"]["base_url"] == f"{BASE}/v1"
assert parsed["model_providers"]["unsloth_api"]["wire_api"] == "responses"
assert "http_headers" not in parsed["model_providers"]["unsloth_api"]
assert parsed["model_providers"]["ollama"]["base_url"] == "http://localhost:11434/v1"
assert connect._merge_codex_config(merged, BASE) == merged
def test_merge_codex_config_keeps_user_oss_provider():
merged = connect._merge_codex_config('oss_provider = "ollama"\n', BASE)
assert _parse_toml(merged)["oss_provider"] == "ollama"
def test_write_codex_config_profile(tmp_path, monkeypatch):
monkeypatch.setenv("CODEX_HOME", str(tmp_path))
connect.write_codex_config(BASE, MODEL)
profile = _parse_toml((tmp_path / "unsloth_api.config.toml").read_text())
assert profile["oss_provider"] == "unsloth_api"
assert profile["model_provider"] == "unsloth_api"
assert profile["model"] == MODEL["id"]
assert profile["model_context_window"] == 131072
config = _parse_toml((tmp_path / "config.toml").read_text())
assert config["model_providers"]["unsloth_api"]["env_key"] == "UNSLOTH_STUDIO_AUTH_TOKEN"
@pytest.fixture()
def fake_studio(tmp_path, monkeypatch, claude_settings):
calls = []
state = {"models": [MODEL]}
def http_json(
method,
url,
token,
payload = None,
timeout = 30,
error = None,
):
calls.append((method, url, payload))
if url.endswith("/v1/models"):
return {"object": "list", "data": state["models"]}
if url.endswith("/api/inference/status"):
return {"is_gguf": True, "model_identifier": state["models"][0]["id"]}
if url.endswith("/api/auth/api-keys"):
return {"key": "sk-unsloth-feedfacefeedface"}
if url.endswith("/api/inference/load"):
state["models"] = [{"id": payload["model_path"], "context_length": 4096}]
return {}
raise AssertionError(f"unexpected request: {method} {url}")
monkeypatch.setattr(connect, "find_studio_server", lambda: BASE)
# Identity handshake has its own tests; trust the loopback server here.
monkeypatch.setattr(connect, "verify_studio_identity", lambda base: True)
# _studio_token / api-keys are faked so the mint flow stays offline.
monkeypatch.setattr(connect, "_studio_token", lambda: "jwt-token")
monkeypatch.setattr(connect, "_http_json", http_json)
monkeypatch.setattr(connect, "_key_cache_path", lambda: tmp_path / "agent_api_key.json")
# No `claude` on PATH, so _claude_cache_flags never probes the real binary.
monkeypatch.setattr(connect.shutil, "which", lambda _: None)
monkeypatch.setenv("CODEX_HOME", str(tmp_path / "codex"))
monkeypatch.delenv("UNSLOTH_API_KEY", raising = False)
return calls
def test_connect_claude_no_launch(fake_studio, claude_settings):
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 0, result.output
_assert_env_unset(result.output, "ANTHROPIC_API_KEY")
_assert_env_unset(result.output, "CLAUDE_CODE_OAUTH_TOKEN")
_assert_env_set(result.output, "ANTHROPIC_BASE_URL", BASE)
_assert_env_set(result.output, "ANTHROPIC_AUTH_TOKEN", "sk-unsloth-feedfacefeedface")
_assert_env_set(result.output, "ANTHROPIC_MODEL", MODEL["id"])
_assert_env_set(result.output, "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC", "1")
_assert_env_set(result.output, "CLAUDE_CODE_DISABLE_EXPERIMENTAL_BETAS", "1")
assert f"claude --model {MODEL['id']} --exclude-dynamic-system-prompt-sections" in result.output
settings = json.loads(claude_settings.read_text())
assert settings["env"]["CLAUDE_CODE_ATTRIBUTION_HEADER"] == "0"
def test_connect_claude_launch_scrubs_conflicting_auth_env(fake_studio, monkeypatch):
captured = {}
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-anthropic-stale")
monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "oauth-stale")
monkeypatch.setattr(connect.shutil, "which", lambda _: "/usr/local/bin/claude")
monkeypatch.setattr(connect, "_claude_cache_flags", lambda: [])
def run(command, env):
captured["command"] = command
captured["env"] = env
return SimpleNamespace(returncode = 0)
monkeypatch.setattr(connect.subprocess, "run", run)
result = CliRunner().invoke(connect.connect_app, ["claude"])
assert result.exit_code == 0, result.output
assert captured["command"] == ["/usr/local/bin/claude", "--model", MODEL["id"]]
assert "ANTHROPIC_API_KEY" not in captured["env"]
assert "CLAUDE_CODE_OAUTH_TOKEN" not in captured["env"]
assert captured["env"]["ANTHROPIC_AUTH_TOKEN"] == "sk-unsloth-feedfacefeedface"
assert captured["env"]["ANTHROPIC_BASE_URL"] == BASE
assert captured["env"]["ANTHROPIC_MODEL"] == MODEL["id"]
@pytest.mark.skipif(
os.name == "nt",
reason = "WSL-from-Linux scenario (calling a Windows agent .exe from inside WSL); "
"os.name is 'posix' under WSL, so this path can't run on a native Windows runner.",
)
def test_connect_claude_windows_shim_from_wsl_bridges_env(fake_studio, monkeypatch):
captured = {}
monkeypatch.setenv("WSL_DISTRO_NAME", "Ubuntu")
monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-anthropic-stale")
monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN", "oauth-stale")
monkeypatch.setattr(
connect.shutil, "which", lambda _: "/mnt/c/Users/samle/AppData/Roaming/npm/claude"
)
monkeypatch.setattr(connect, "_claude_cache_flags", lambda: [])
def run(command, env):
captured["command"] = command
captured["env"] = env
return SimpleNamespace(returncode = 0)
monkeypatch.setattr(connect.subprocess, "run", run)
result = CliRunner().invoke(connect.connect_app, ["claude"])
assert result.exit_code == 0, result.output
assert captured["command"] == [
"/mnt/c/Users/samle/AppData/Roaming/npm/claude",
"--model",
MODEL["id"],
]
assert captured["env"]["ANTHROPIC_API_KEY"] == ""
assert captured["env"]["CLAUDE_CODE_OAUTH_TOKEN"] == ""
assert captured["env"]["ANTHROPIC_AUTH_TOKEN"] == "sk-unsloth-feedfacefeedface"
assert captured["env"]["ANTHROPIC_BASE_URL"] == BASE
assert captured["env"]["ANTHROPIC_MODEL"] == MODEL["id"]
for name in (
"ANTHROPIC_AUTH_TOKEN",
"ANTHROPIC_BASE_URL",
"ANTHROPIC_MODEL",
"ANTHROPIC_API_KEY",
"CLAUDE_CODE_OAUTH_TOKEN",
):
assert name in captured["env"]["WSLENV"].split(":")
@pytest.mark.skipif(
os.name == "nt",
reason = "WSL-from-Linux scenario (calling a Windows agent .exe from inside WSL); "
"os.name is 'posix' under WSL, so this path can't run on a native Windows runner.",
)
def test_connect_claude_no_launch_windows_shim_from_wsl_prints_wslenv(fake_studio, monkeypatch):
monkeypatch.setenv("WSL_DISTRO_NAME", "Ubuntu")
monkeypatch.setattr(
connect.shutil, "which", lambda _: "/mnt/c/Users/samle/AppData/Roaming/npm/claude"
)
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 0, result.output
assert "export ANTHROPIC_API_KEY=" in result.output
assert "export CLAUDE_CODE_OAUTH_TOKEN=" in result.output
assert "export WSLENV=" in result.output
assert "ANTHROPIC_AUTH_TOKEN" in result.output
assert "CLAUDE_CODE_OAUTH_TOKEN" in result.output
def test_connect_codex_no_launch(fake_studio, tmp_path):
result = CliRunner().invoke(connect.connect_app, ["codex", "--no-launch"])
assert result.exit_code == 0, result.output
_assert_env_set(result.output, "UNSLOTH_STUDIO_AUTH_TOKEN", "sk-unsloth-feedfacefeedface")
assert "codex --oss --profile unsloth_api" in result.output
assert (tmp_path / "codex" / "config.toml").exists()
assert (tmp_path / "codex" / "unsloth_api.config.toml").exists()
def test_connect_key_minted_once_then_cached(fake_studio, tmp_path):
CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
# First run mints; second reuses the minted key cached for this server.
mints = [c for c in fake_studio if c[1].endswith("/api/auth/api-keys")]
assert len(mints) == 1
cached = json.loads((tmp_path / "agent_api_key.json").read_text())
assert cached["servers"][BASE]["minted"] == ["sk-unsloth-feedfacefeedface"]
def test_connect_explicit_key_remembered_for_keyless_runs(fake_studio, tmp_path):
CliRunner().invoke(
connect.connect_app,
["claude", "--no-launch", "--api-key", "sk-unsloth-deadbeefdeadbeef"],
)
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 0, result.output
# Reused, not re-minted (a mint would return the feedface stand-in).
_assert_env_set(result.output, "ANTHROPIC_AUTH_TOKEN", "sk-unsloth-deadbeefdeadbeef")
cached = json.loads((tmp_path / "agent_api_key.json").read_text())
# An explicit key is remembered as "saved" so it replays without the handshake.
assert cached["servers"][BASE]["saved"] == ["sk-unsloth-deadbeefdeadbeef"]
def test_connect_skips_cached_keys_the_server_rejects(fake_studio, tmp_path, monkeypatch):
cache = tmp_path / "agent_api_key.json"
cache.write_text(
json.dumps(
{"servers": {BASE: {"minted": ["sk-unsloth-stale", "sk-unsloth-feedfacefeedface"]}}}
)
)
inner = connect._http_json
def http_json(
method,
url,
token,
payload = None,
timeout = 30,
error = None,
):
if url.endswith("/v1/models") and token == "sk-unsloth-stale":
raise urllib.error.HTTPError(url, 401, "Unauthorized", None, None)
return inner(method, url, token, payload, timeout, error)
monkeypatch.setattr(connect, "_http_json", http_json)
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 0, result.output
_assert_env_set(result.output, "ANTHROPIC_AUTH_TOKEN", "sk-unsloth-feedfacefeedface")
# The working key moves to the front so the next run tries it first.
cached = json.loads(cache.read_text())
assert cached["servers"][BASE]["minted"] == ["sk-unsloth-feedfacefeedface", "sk-unsloth-stale"]
def test_connect_legacy_unscoped_cache_not_replayed(fake_studio, tmp_path):
# Legacy unscoped caches have no server binding (could leak across servers),
# so they're ignored: a fresh key is minted and stored scoped to this server.
(tmp_path / "agent_api_key.json").write_text(json.dumps({"key": "sk-unsloth-oldformat"}))
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 0, result.output
_assert_env_set(result.output, "ANTHROPIC_AUTH_TOKEN", "sk-unsloth-feedfacefeedface")
cached = json.loads((tmp_path / "agent_api_key.json").read_text())
assert cached["servers"][BASE]["minted"] == ["sk-unsloth-feedfacefeedface"]
assert "key" not in cached # legacy field collapsed away
def test_connect_model_flag_loads_on_server(fake_studio):
result = CliRunner().invoke(
connect.connect_app, ["claude", "--no-launch", "--model", "unsloth/Qwen3.5-35B-A3B"]
)
assert result.exit_code == 0, result.output
loads = [c for c in fake_studio if c[1].endswith("/api/inference/load")]
assert loads == [
("POST", f"{BASE}/api/inference/load", {"model_path": "unsloth/Qwen3.5-35B-A3B"})
]
_assert_env_set(result.output, "ANTHROPIC_MODEL", "unsloth/Qwen3.5-35B-A3B")
def test_connect_model_flag_matches_canonical_id(fake_studio, monkeypatch):
# Studio registers a loaded model under a canonical id (resolved identifier
# / casing) that can differ from the path we passed. The agent must connect
# to that model, not silently fall through to the first loaded one.
requested = "Unsloth/Qwen3.5-35B-A3B"
canonical = "unsloth/Qwen3.5-35B-A3B"
inner = connect._http_json
def http_json(
method,
url,
token,
payload = None,
timeout = 30,
error = None,
):
if url.endswith("/api/inference/load"):
return {"model": canonical, "display_name": canonical}
if url.endswith("/v1/models"):
# Decoy sorts first, so models[0] is the wrong pick on the old code.
return {"object": "list", "data": [MODEL, {"id": canonical, "context_length": 4096}]}
return inner(method, url, token, payload, timeout, error)
monkeypatch.setattr(connect, "_http_json", http_json)
result = CliRunner().invoke(
connect.connect_app, ["claude", "--no-launch", "--model", requested]
)
assert result.exit_code == 0, result.output
_assert_env_set(result.output, "ANTHROPIC_MODEL", canonical)
def test_connect_no_model_loaded_errors(fake_studio, monkeypatch):
monkeypatch.setattr(
connect,
"_http_json",
lambda method, url, token, payload = None, timeout = 30, error = None: (
{"key": "sk-unsloth-feedfacefeedface"}
if url.endswith("/api/auth/api-keys")
else {"object": "list", "data": []}
),
)
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 1
assert "No model is loaded" in result.output
def test_connect_requested_model_not_loaded_fails(fake_studio, monkeypatch):
# Studio never surfaces the requested model; fail loudly rather than
# silently connecting to whatever else happens to be loaded.
inner = connect._http_json
def http_json(
method,
url,
token,
payload = None,
timeout = 30,
error = None,
):
if url.endswith("/api/inference/load"):
return {}
if url.endswith("/v1/models"):
return {"object": "list", "data": [MODEL]} # decoy; request never appears
return inner(method, url, token, payload, timeout, error)
monkeypatch.setattr(connect, "_http_json", http_json)
result = CliRunner().invoke(
connect.connect_app, ["claude", "--no-launch", "--model", "unsloth/Missing-7B"]
)
assert result.exit_code == 1
assert "unsloth/Missing-7B" in result.output
def test_connect_codex_rejects_non_gguf_model(fake_studio, monkeypatch):
inner = connect._http_json
def http_json(
method,
url,
token,
payload = None,
timeout = 30,
error = None,
):
if url.endswith("/api/inference/status"):
return {"is_gguf": False, "model_identifier": "unsloth/Qwen3-0.6B"}
return inner(method, url, token, payload, timeout, error)
monkeypatch.setattr(connect, "_http_json", http_json)
result = CliRunner().invoke(connect.connect_app, ["codex", "--no-launch"])
assert result.exit_code == 1
assert "GGUF" in result.output
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 0, result.output
def test_connect_nonloopback_keyless_refuses_to_send_credential(fake_studio, monkeypatch):
# A server known only by URL + health check is unverified: keyless connect
# must refuse and make no request at all.
monkeypatch.setattr(connect, "find_studio_server", lambda: "http://studio.evil.example:8888")
result = CliRunner().invoke(connect.connect_app, ["opencode", "--no-launch"])
assert result.exit_code == 1
assert "Settings → API" in result.output
assert "--api-key" in result.output
assert fake_studio == [] # no HTTP request of any kind (no mint, no /v1/models)
def test_connect_nonloopback_explicit_key_is_allowed(fake_studio, monkeypatch):
# User named both server and key, so it's their choice; only auto-send is blocked.
monkeypatch.setattr(connect, "find_studio_server", lambda: "http://studio.example:8888")
result = CliRunner().invoke(
connect.connect_app,
["opencode", "--no-launch", "--api-key", "sk-unsloth-deadbeefdeadbeef"],
)
assert result.exit_code == 0, result.output
def test_connect_nonloopback_replays_saved_key(fake_studio, tmp_path, monkeypatch):
# A key saved for a remote (non-loopback) Studio is replayed on keyless runs;
# auto-minting stays blocked for non-loopback.
remote = "http://studio.example:8888"
monkeypatch.setattr(connect, "find_studio_server", lambda: remote)
(tmp_path / "agent_api_key.json").write_text(
json.dumps({"servers": {remote: {"saved": ["sk-unsloth-deadbeefdeadbeef"]}}})
)
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 0, result.output
_assert_env_set(result.output, "ANTHROPIC_AUTH_TOKEN", "sk-unsloth-deadbeefdeadbeef")
assert not any(c[1].endswith("/api/auth/api-keys") for c in fake_studio) # never minted
def test_connect_studio_server_errors_on_explicit_remote(monkeypatch):
# A user who pointed UNSLOTH_STUDIO_URL at a remote Studio should get an
# error, not a silent local model load (which they did not ask for).
import typer
import unsloth_cli._inference as inference
monkeypatch.setenv("UNSLOTH_STUDIO_URL", "http://studio.example:8888")
monkeypatch.setattr(
inference, "find_studio_server", lambda *a, **k: "http://studio.example:8888"
)
with pytest.raises(typer.Exit):
inference.connect_studio_server("m", hf_token = None, max_seq_length = 4096, load_in_4bit = False)
def test_connect_studio_server_falls_back_locally_on_default_discovery(monkeypatch):
# Opportunistic local discovery (no UNSLOTH_STUDIO_URL): if the loopback
# server can't be verified, fall back to a local load rather than erroring.
import unsloth_cli._inference as inference
monkeypatch.delenv("UNSLOTH_STUDIO_URL", raising = False)
monkeypatch.setattr(inference, "find_studio_server", lambda *a, **k: "http://127.0.0.1:8888")
monkeypatch.setattr(inference, "verify_studio_identity", lambda *a, **k: False)
assert (
inference.connect_studio_server("m", hf_token = None, max_seq_length = 4096, load_in_4bit = False)
is None
)
def test_connect_unverified_loopback_without_cached_key_refuses_to_mint(
fake_studio, tmp_path, monkeypatch
):
# With no saved key, the next step would auto-mint; an unverified loopback
# server (port squatter) must be refused, with nothing sent.
monkeypatch.setattr(connect, "verify_studio_identity", lambda base: False)
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 1
assert "--api-key" in result.output
assert not any(c[1].endswith("/api/auth/api-keys") for c in fake_studio) # never minted
def test_connect_replays_saved_key_without_identity_check(fake_studio, tmp_path, monkeypatch):
# A "saved" key (e.g. for an SSH-tunnelled Studio the handshake can't match)
# replays on keyless runs without the handshake, scoped to its own base.
cache = tmp_path / "agent_api_key.json"
cache.write_text(json.dumps({"servers": {BASE: {"saved": ["sk-unsloth-deadbeefdeadbeef"]}}}))
monkeypatch.setattr(connect, "verify_studio_identity", lambda base: False)
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 0, result.output
_assert_env_set(result.output, "ANTHROPIC_AUTH_TOKEN", "sk-unsloth-deadbeefdeadbeef")
assert not any(c[1].endswith("/api/auth/api-keys") for c in fake_studio) # reused, not minted
def test_connect_minted_cache_requires_identity_check(fake_studio, tmp_path, monkeypatch):
# A "minted" key is NOT replayed to an unverified loopback server: minting and
# minted-key replay both sit behind the handshake, so a squatter can't grab it.
cache = tmp_path / "agent_api_key.json"
cache.write_text(json.dumps({"servers": {BASE: {"minted": ["sk-unsloth-feedfacefeedface"]}}}))
monkeypatch.setattr(connect, "verify_studio_identity", lambda base: False)
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 1
assert "--api-key" in result.output
assert not any(c[1].endswith("/v1/models") for c in fake_studio) # minted key never sent
def test_connect_explicit_key_skips_identity_check(fake_studio, monkeypatch):
# An explicit key is the user's deliberate choice, so it does not require
# the automatic identity handshake.
monkeypatch.setattr(connect, "verify_studio_identity", lambda base: False)
result = CliRunner().invoke(
connect.connect_app,
["claude", "--no-launch", "--api-key", "sk-unsloth-deadbeefdeadbeef"],
)
assert result.exit_code == 0, result.output
_assert_env_set(result.output, "ANTHROPIC_AUTH_TOKEN", "sk-unsloth-deadbeefdeadbeef")
def _serve_identity(proof_for):
"""Start a localhost HTTP server answering /api/auth/identity with
proof_for(nonce_bytes). Returns (base_url, shutdown)."""
import base64
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import parse_qs, urlparse
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path != "/api/auth/identity":
self.send_response(404)
self.end_headers()
return
nonce = base64.urlsafe_b64decode(parse_qs(parsed.query)["nonce"][0])
host, port = self.server.server_address[0], self.server.server_address[1]
body = json.dumps({"proof": proof_for(nonce, host, port)}).encode()
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(body)
def log_message(self, *a):
pass
server = HTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target = server.serve_forever, daemon = True).start()
base = f"http://127.0.0.1:{server.server_address[1]}"
return base, server.shutdown
def test_verify_studio_identity_end_to_end(tmp_path, monkeypatch):
# Real crypto end to end: verify_studio_identity reads the install secret from
# an isolated DB; a "good" server proves the same secret, a spoofing one can't.
import unsloth_cli._inference as inference
inference.ensure_studio_backend_path()
try:
from studio.backend.auth import storage
except Exception as exc: # backend not importable here (e.g. missing deps)
pytest.skip(f"studio backend not importable: {exc}")
monkeypatch.setattr(storage, "DB_PATH", tmp_path / "auth.db")
monkeypatch.setattr(storage, "_identity_secret_cache", None)
good = lambda nonce, host, port: storage.compute_identity_proof(
nonce, host, port
) # real secret
bad = lambda nonce, host, port: "00" * 32 # spoofer without the secret
base_ok, stop_ok = _serve_identity(good)
base_bad, stop_bad = _serve_identity(bad)
try:
assert inference.verify_studio_identity(base_ok) is True
assert inference.verify_studio_identity(base_bad) is False
finally:
stop_ok()
stop_bad()
def _serve_redirect(target):
"""Start a localhost server that 302-redirects every GET to target+path."""
import threading
from http.server import BaseHTTPRequestHandler, HTTPServer
class Handler(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(302)
self.send_header("Location", target + self.path)
self.end_headers()
def log_message(self, *a):
pass
server = HTTPServer(("127.0.0.1", 0), Handler)
threading.Thread(target = server.serve_forever, daemon = True).start()
base = f"http://127.0.0.1:{server.server_address[1]}"
return base, server.shutdown
def test_verify_studio_identity_rejects_redirect(tmp_path, monkeypatch):
# A squatter could 302 /api/auth/identity to the real Studio and relay its
# proof; redirects must be refused so the squatter's base isn't accepted.
import unsloth_cli._inference as inference
inference.ensure_studio_backend_path()
try:
from studio.backend.auth import storage
except Exception as exc:
pytest.skip(f"studio backend not importable: {exc}")
monkeypatch.setattr(storage, "DB_PATH", tmp_path / "auth.db")
monkeypatch.setattr(storage, "_identity_secret_cache", None)
real_base, stop_real = _serve_identity(
lambda nonce, host, port: storage.compute_identity_proof(nonce, host, port)
)
squatter_base, stop_squatter = _serve_redirect(real_base)
try:
assert inference.verify_studio_identity(real_base) is True # direct: ok
assert inference.verify_studio_identity(squatter_base) is False # relayed: refused
finally:
stop_real()
stop_squatter()
def test_verify_studio_identity_rejects_relayed_proof(tmp_path, monkeypatch):
# A squatter that proxies the nonce to the real Studio on another port gets a
# proof bound to *that* port; the client expects one bound to the port it
# connected to, so the relayed proof is rejected.
import unsloth_cli._inference as inference
inference.ensure_studio_backend_path()
try:
from studio.backend.auth import storage
except Exception as exc:
pytest.skip(f"studio backend not importable: {exc}")
monkeypatch.setattr(storage, "DB_PATH", tmp_path / "auth.db")
monkeypatch.setattr(storage, "_identity_secret_cache", None)
real_base, stop_real = _serve_identity(
lambda nonce, host, port: storage.compute_identity_proof(nonce, host, port)
)
real_port = int(real_base.rsplit(":", 1)[1])
# The squatter answers on its own port but returns the proof for the real port.
squatter_base, stop_squatter = _serve_identity(
lambda nonce, host, port: storage.compute_identity_proof(nonce, host, real_port)
)
try:
assert inference.verify_studio_identity(real_base) is True
assert inference.verify_studio_identity(squatter_base) is False
finally:
stop_real()
stop_squatter()
@pytest.mark.parametrize(
"url, loopback",
[
("http://127.0.0.1:8888", True),
("http://localhost:8888", True),
("http://[::1]:8888", True),
("http://127.0.0.5:9001", True), # SSH tunnels can land anywhere in 127/8
("http://0.0.0.0:8888", False),
("http://10.0.0.5:8888", False),
("http://studio.evil.example:8888", False),
("https://studio.example.com", False),
],
)
def test_is_loopback_url(url, loopback):
assert connect.is_loopback_url(url) is loopback
def test_connect_no_studio_errors(fake_studio, monkeypatch):
monkeypatch.setattr(connect, "find_studio_server", lambda: None)
result = CliRunner().invoke(connect.connect_app, ["claude", "--no-launch"])
assert result.exit_code == 1
assert "No running Studio server" in result.output
def test_connect_explicit_api_key_skips_mint(fake_studio):
result = CliRunner().invoke(
connect.connect_app,
["claude", "--no-launch", "--api-key", "sk-unsloth-deadbeefdeadbeef"],
)
assert result.exit_code == 0, result.output
_assert_env_set(result.output, "ANTHROPIC_AUTH_TOKEN", "sk-unsloth-deadbeefdeadbeef")
assert not any(c[1].endswith("/api/auth/api-keys") for c in fake_studio)
# ── OpenClaw (Anthropic /v1/messages) ────────────────────────────────
def test_write_openclaw_config_fresh(tmp_path, monkeypatch):
path = tmp_path / "openclaw.json"
monkeypatch.setattr(connect, "openclaw_config_path", lambda: path)
connect.write_openclaw_config(BASE, "sk-unsloth-abc", MODEL)
config = json.loads(path.read_text())
provider = config["models"]["providers"]["unsloth"]
assert provider["baseUrl"] == f"{BASE}/v1"
assert provider["apiKey"] == "sk-unsloth-abc"
assert provider["api"] == "openai-completions"
assert provider["models"] == [
{"id": MODEL["id"], "name": MODEL["id"], "contextWindow": MODEL["context_length"]}
]
# The default model must be pinned or OpenClaw has nothing active.
assert config["agents"]["defaults"]["model"]["primary"] == f"unsloth/{MODEL['id']}"
assert config["gateway"]["mode"] == "local"
assert config["gateway"]["auth"]["mode"] == "none" # unauth loopback gateway
if os.name != "nt": # the file holds an API key
assert path.stat().st_mode & 0o777 == 0o600
def test_write_openclaw_config_preserves_and_idempotent(tmp_path, monkeypatch):
path = tmp_path / "openclaw.json"
monkeypatch.setattr(connect, "openclaw_config_path", lambda: path)
path.write_text(
json.dumps(
{
"theme": "dark",
"agents": {"defaults": {"temperature": 0.5}},
"models": {"mode": "replace", "providers": {"openrouter": {"baseUrl": "x"}}},
}
)
)
connect.write_openclaw_config(BASE, "sk-unsloth-abc", MODEL)
config = json.loads(path.read_text())
assert config["theme"] == "dark"
assert config["agents"]["defaults"]["temperature"] == 0.5 # other agent defaults kept
assert config["agents"]["defaults"]["model"]["primary"] == f"unsloth/{MODEL['id']}"
assert config["models"]["mode"] == "replace" # user's mode is left as-is
assert config["models"]["providers"]["openrouter"]["baseUrl"] == "x"
assert config["models"]["providers"]["unsloth"]["baseUrl"] == f"{BASE}/v1"
before = path.read_text()
connect.write_openclaw_config(BASE, "sk-unsloth-abc", MODEL)
assert path.read_text() == before
def test_write_openclaw_config_corrupt_left_alone(tmp_path, monkeypatch, capsys):
path = tmp_path / "openclaw.json"
monkeypatch.setattr(connect, "openclaw_config_path", lambda: path)
path.write_text("{not json")
connect.write_openclaw_config(BASE, "sk-unsloth-abc", MODEL)
assert path.read_text() == "{not json"
assert "couldn't parse" in capsys.readouterr().err
def test_connect_openclaw_no_launch(fake_studio, tmp_path, monkeypatch):
path = tmp_path / "openclaw.json"
monkeypatch.setattr(connect, "openclaw_config_path", lambda: path)
result = CliRunner().invoke(connect.connect_app, ["openclaw", "--no-launch"])
assert result.exit_code == 0, result.output
assert "openclaw" in result.output
assert "export" not in result.output # key lives in the config, not the env
config = json.loads(path.read_text())
assert config["models"]["providers"]["unsloth"]["apiKey"] == "sk-unsloth-feedfacefeedface"
assert config["agents"]["defaults"]["model"]["primary"] == f"unsloth/{MODEL['id']}"
# OpenAI /v1/chat/completions works on either backend — no GGUF gate.
assert not any(c[1].endswith("/api/inference/status") for c in fake_studio)
# ── OpenCode (OpenAI /v1/chat/completions) ───────────────────────────
def test_write_opencode_config_fresh(tmp_path, monkeypatch):
path = tmp_path / "opencode.json"
monkeypatch.setattr(connect, "opencode_config_path", lambda: path)
connect.write_opencode_config(BASE, "sk-unsloth-abc", MODEL)
config = json.loads(path.read_text())
provider = config["provider"]["unsloth"]
assert provider["npm"] == "@ai-sdk/openai-compatible"
assert provider["options"] == {"baseURL": f"{BASE}/v1", "apiKey": "sk-unsloth-abc"}
assert provider["models"] == {MODEL["id"]: {"name": MODEL["id"]}}
assert config["model"] == f"unsloth/{MODEL['id']}"
def test_write_opencode_config_preserves_and_idempotent(tmp_path, monkeypatch):
path = tmp_path / "opencode.json"
monkeypatch.setattr(connect, "opencode_config_path", lambda: path)
path.write_text(
json.dumps({"theme": "tokyonight", "provider": {"anthropic": {"name": "Anthropic"}}})
)
connect.write_opencode_config(BASE, "sk-unsloth-abc", MODEL)
config = json.loads(path.read_text())
assert config["theme"] == "tokyonight"
assert config["provider"]["anthropic"]["name"] == "Anthropic"
assert config["provider"]["unsloth"]["options"]["baseURL"] == f"{BASE}/v1"
before = path.read_text()
connect.write_opencode_config(BASE, "sk-unsloth-abc", MODEL)
assert path.read_text() == before
def test_connect_opencode_no_launch(fake_studio, tmp_path, monkeypatch):
path = tmp_path / "opencode.json"
monkeypatch.setattr(connect, "opencode_config_path", lambda: path)
result = CliRunner().invoke(connect.connect_app, ["opencode", "--no-launch"])
assert result.exit_code == 0, result.output
assert "opencode" in result.output
config = json.loads(path.read_text())
assert config["provider"]["unsloth"]["options"]["apiKey"] == "sk-unsloth-feedfacefeedface"
assert config["model"] == f"unsloth/{MODEL['id']}"
assert not any(c[1].endswith("/api/inference/status") for c in fake_studio)
# ── Hermes (OpenAI /v1/chat/completions, key via env) ────────────────
@pytest.fixture()
def hermes_config(tmp_path, monkeypatch):
path = tmp_path / "config.yaml"
monkeypatch.setattr(connect, "hermes_config_path", lambda: path)
return path
def test_write_hermes_config_fresh(hermes_config):
yaml = pytest.importorskip("yaml")
connect.write_hermes_config(BASE, MODEL)
config = yaml.safe_load(hermes_config.read_text())
# Hermes only honors the key for a *named* custom provider, so the endpoint
# is registered under providers.* and model.provider points at it.
assert config["model"]["provider"] == "custom:unsloth"
assert config["model"]["default"] == MODEL["id"]
assert config["model"]["api_mode"] == "openai"
provider = config["providers"]["unsloth"]
assert provider["base_url"] == f"{BASE}/v1"
assert provider["api_mode"] == "openai"
assert provider["key_env"] == "UNSLOTH_API_KEY"
# The key is resolved from the launch env, never written to disk.
assert "sk-unsloth" not in hermes_config.read_text()
def test_write_hermes_config_preserves_and_idempotent(hermes_config):
yaml = pytest.importorskip("yaml")
hermes_config.write_text(
yaml.safe_dump(
{
"terminal": {"backend": "local"},
"model": {"temperature": 0.7},
"providers": {"openrouter": {"base_url": "https://openrouter.ai/api/v1"}},
}
)
)
connect.write_hermes_config(BASE, MODEL)
config = yaml.safe_load(hermes_config.read_text())
assert config["terminal"] == {"backend": "local"} # unrelated sections kept
assert config["model"]["temperature"] == 0.7 # unrelated model keys kept
assert config["model"]["provider"] == "custom:unsloth"
assert config["providers"]["openrouter"]["base_url"] == "https://openrouter.ai/api/v1"
assert config["providers"]["unsloth"]["base_url"] == f"{BASE}/v1"
before = hermes_config.read_text()
connect.write_hermes_config(BASE, MODEL)
assert hermes_config.read_text() == before
def test_write_hermes_config_preserves_non_mapping_file(hermes_config, capsys):
pytest.importorskip("yaml")
original = "- just\n- a\n- list\n" # valid YAML, but not a mapping
hermes_config.write_text(original)
connect.write_hermes_config(BASE, MODEL)
assert hermes_config.read_text() == original # user-managed file left untouched
assert "couldn't parse" in capsys.readouterr().err
def test_connect_hermes_no_launch(fake_studio, hermes_config):
yaml = pytest.importorskip("yaml")
result = CliRunner().invoke(connect.connect_app, ["hermes", "--no-launch"])
assert result.exit_code == 0, result.output
_assert_env_set(result.output, "UNSLOTH_API_KEY", "sk-unsloth-feedfacefeedface")
assert "hermes" in result.output
config = yaml.safe_load(hermes_config.read_text())
assert config["model"]["provider"] == "custom:unsloth"
assert config["providers"]["unsloth"]["base_url"] == f"{BASE}/v1"
assert config["model"]["default"] == MODEL["id"]
assert not any(c[1].endswith("/api/inference/status") for c in fake_studio)

1848
unsloth_cli/tests/test_start.py Normal file
View file

File diff suppressed because it is too large Load diff