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

Merge 34f1e9b089 into 3212710a4a

Browse source
This commit is contained in:
oobabooga 2026-07-29 09:25:42 +00:00 • committed by GitHub
commit e1269d778a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 813 additions and 127 deletions
Download patch file Download diff file Expand all files Collapse all files

15
studio/backend/requirements/no-torch-runtime.txt
View file

@ -2,23 +2,22 @@
# Installed with --no-deps to prevent transitive torch resolution
# from packages like accelerate, peft, trl, sentence-transformers.
#
# Includes unsloth's own direct deps (typer, pydantic, pyyaml,
# nest-asyncio) since unsloth is also installed with --no-deps
# Includes unsloth's own direct deps (typer, click, rich, pydantic,
# pyyaml, nest-asyncio) since unsloth is also installed with --no-deps
# (current PyPI metadata still declares torch as a hard dep).
# unsloth direct deps (from pyproject.toml [project].dependencies)
typer>=0.12.0
# typer's full runtime dep tree. Required explicitly because this
click>=8.0
rich>=13.0
# typer's remaining runtime dep tree. Required explicitly because this
# file is installed with --no-deps. On Linux/Mac CI runners these
# are often cached transitively; on a fresh windows-latest venv they
# are not, and `unsloth studio setup` crashes with
# `ModuleNotFoundError: No module named 'click'`, then 'annotated_doc',
# then 'rich', etc. as each is hit. Pin the full chain so the
# are not, and `unsloth studio setup` crashes on the first missing
# module. Pin the full chain so the
# no-torch path works cleanly on every fresh venv.
click>=8.0
shellingham>=1.5
annotated-doc>=0.0.3
rich>=13.0
markdown-it-py>=3.0
mdurl>=0.1
pygments>=2.0

381
unsloth_cli/commands/start.py
View file

@ -385,6 +385,7 @@ def _opencode_supports_native_auto() -> bool:
text = True,
timeout = 10,
stderr = subprocess.DEVNULL,
env = _probe_env(),
)
except Exception:
return False
@ -439,6 +440,23 @@ def _hermes_install_hint() -> str:
return _HERMES_WINDOWS_INSTALL_HINT if os.name == "nt" else _HERMES_POSIX_INSTALL_HINT
def _npm_install_hint(package: str, *, ignore_scripts: bool = False) -> str:
parts = ["npm", "install", "-g"]
if os.name != "nt":
# No resolvable home (container under a bare UID): fall back to npm's own prefix
# rather than failing a launch whose agent may already be installed.
try:
parts.extend(("--prefix", str(Path.home() / ".local")))
except (RuntimeError, OSError):
pass
if ignore_scripts:
parts.append("--ignore-scripts")
parts.append(package)
if os.name == "nt":
return " ".join(_powershell_quote(part) for part in parts)
return shlex.join(parts)
def _hermes_resume_oneshot_args(args: list[str]) -> list[str]:
"""Route resumed one-shot prompts through Hermes' session-aware chat command."""
has_resume = any(
@ -1213,10 +1231,13 @@ def _require_studio(
)
def _studio_auth_root() -> Path:
from unsloth_cli.commands.studio import STUDIO_HOME
return STUDIO_HOME / "auth"
def _key_cache_path() -> Path:
ensure_studio_backend_path()
from utils.paths import auth_root
return auth_root() / "agent_api_key.json"
return _studio_auth_root() / "agent_api_key.json"
def _read_cache(cache: Path) -> dict:
@ -1657,7 +1678,7 @@ def _claude_version() -> Optional[tuple]:
return None
try:
result = subprocess.run(
[executable, "--version"], capture_output = True, text = True, timeout = 10
[executable, "--version"], capture_output = True, text = True, timeout = 10, env = _probe_env()
)
# Pull the X.Y.Z out of the output rather than assuming it is the first token.
# claude prints it first today ("2.1.98 (Claude Code)"), but a format change
@ -1734,7 +1755,11 @@ def _codex_supports_model_catalog() -> bool:
return True
try:
output = subprocess.check_output(
[executable, "--version"], text = True, timeout = 10, stderr = subprocess.DEVNULL
[executable, "--version"],
text = True,
timeout = 10,
stderr = subprocess.DEVNULL,
env = _probe_env(),
)
except Exception:
return False
@ -2005,20 +2030,6 @@ def write_codex_parent_overlay(overlay: Path) -> Path:
return overlay
@contextlib.contextmanager
def _codex_parent_overlay(session_home: Path, *, launch: bool, persist: bool):
if launch and not persist:
temp_root = _agents_config_root() / ".tmp"
temp_root.mkdir(parents = True, exist_ok = True, mode = 0o700)
overlay = Path(tempfile.mkdtemp(prefix = "codex-parent-", dir = temp_root))
try:
yield write_codex_parent_overlay(overlay)
finally:
shutil.rmtree(overlay, ignore_errors = True)
else:
yield write_codex_parent_overlay(session_home / "parent")
def _agent_config_path(path: Path, command: list) -> str:
"""Translate a generated config path when a Windows agent runs through WSL."""
return _wsl_windows_path(path) if _wsl_windows_executable(command) else str(path)
@ -2068,8 +2079,7 @@ def _opencode_subagent_inline_config(path: Path, permission: dict) -> dict:
err = True,
)
else:
env = os.environ.copy()
env["OPENCODE_CONFIG"] = _agent_config_path(path, ["opencode"])
env = _probe_env(OPENCODE_CONFIG = _agent_config_path(path, ["opencode"]))
try:
resolved = subprocess.run(
[executable, "debug", "config"],
@ -2408,19 +2418,51 @@ def _refresh_windows_path() -> None:
os.environ["PATH"] = os.pathsep.join(entries)
def _managed_node_tools() -> Optional[tuple[Path, Path, bool]]:
# Best-effort probe on the launch path: reaching the backend for the managed Node must
# never break a launch, so any failure (missing package, unreadable home, degraded env)
# just means "no managed Node" and falls back to the system runtime.
try:
ensure_studio_backend_path()
from utils.node_runtime import managed_node_binary, resolve_node_executable
node = Path(managed_node_binary())
except (ImportError, OSError, RuntimeError, TypeError, ValueError):
return None
npm = node.with_name("npm.cmd" if os.name == "nt" else "npm")
try:
usable = node.is_file() and npm.is_file()
if os.name != "nt":
usable = usable and os.access(node, os.X_OK) and os.access(npm, os.X_OK)
except OSError:
return None
if not usable:
return None
try:
resolved = resolve_node_executable()
preferred = bool(resolved) and Path(resolved).resolve() == node.resolve()
except (OSError, RuntimeError, TypeError, ValueError):
preferred = False
return node, npm, preferred
def _augment_path_with_install_dirs() -> None:
# Append known install dirs to PATH so a freshly installed agent resolves without a new
# shell: some installers write the binary but not PATH (claude drops ~/.local/bin and
# only prints a note; npm -g shims land in %APPDATA%\npm). Appended, so precedence holds.
# Add known install dirs to PATH so a freshly installed agent resolves without a new
# shell. User agent dirs are appended so existing tools keep precedence. A managed Node
# selected over the system runtime is prepended so Node-backed shims use that same runtime.
# Only the user install dirs need a home; a missing one must not also drop the managed
# Node, or a shim resolved here fails on `env node` under a bare container UID.
try:
home = Path.home()
except (RuntimeError, OSError):
return
candidates = [home / ".local" / "bin"]
home = None
candidates = [home / ".local" / "bin"] if home is not None else []
if os.name == "nt":
appdata = os.environ.get("APPDATA")
if appdata:
candidates.append(Path(appdata) / "npm")
managed_node = _managed_node_tools()
if managed_node is not None and not managed_node[2]:
candidates.append(managed_node[0].parent)
current = os.environ.get("PATH")
if current is None:
# PATH unset: shutil.which() and exec*p* fall back to os.defpath (e.g. /bin:/usr/bin), so
@ -2428,14 +2470,42 @@ def _augment_path_with_install_dirs() -> None:
# system-installed agent and strip the launched child's normal PATH). An explicitly empty
# PATH is left as-is: like shutil.which, it means "search nothing", not os.defpath.
current = os.defpath
preferred_node = (
str(managed_node[0].parent) if managed_node is not None and managed_node[2] else None
)
if preferred_node:
preferred_key = os.path.normcase(preferred_node)
current = os.pathsep.join(
entry
for entry in current.split(os.pathsep)
if not entry or os.path.normcase(entry) != preferred_key
)
seen = {os.path.normcase(entry) for entry in current.split(os.pathsep) if entry}
additions = [
str(directory)
for directory in candidates
if directory.is_dir() and os.path.normcase(str(directory)) not in seen
]
if additions:
os.environ["PATH"] = os.pathsep.join([current, *additions] if current else additions)
if preferred_node or additions:
parts = [part for part in (preferred_node, current, *additions) if part]
os.environ["PATH"] = os.pathsep.join(parts)
def _probe_env(**extra: str) -> dict:
"""Environment for probes that RUN a resolved shim.
_which_with_install_dirs restores PATH before returning, so a shim backed by Studio's
managed Node would not find that node when executed.
"""
original = os.environ.get("PATH")
_augment_path_with_install_dirs()
env = os.environ.copy()
if original is None:
os.environ.pop("PATH", None)
else:
os.environ["PATH"] = original
env.update(extra)
return env
def _which_with_install_dirs(name: str) -> Optional[str]:
@ -2471,6 +2541,71 @@ def _pinned_raw_github_commit(source: str) -> Optional[str]:
return match.group(1).lower() if match else None
def _npm_executable() -> Optional[str]:
managed_node = _managed_node_tools()
if not (managed_node and managed_node[2]):
executable = shutil.which("npm")
if executable and not _wsl_windows_executable([executable]):
return executable
if executable:
# WSL inherits the Windows PATH, so the shim just rejected may be shadowing a
# usable native npm further along it. Keep looking rather than giving up here.
for directory in os.get_exec_path():
candidate = shutil.which("npm", path = directory)
if candidate and not _wsl_windows_executable([candidate]):
return candidate
return str(managed_node[1]) if managed_node is not None else None
def _install_command(install_hint: str) -> tuple[list[str], Optional[dict]]:
if not re.match(r"^\s*npm(?:\s|$)", install_hint):
if os.name == "nt":
return (
[
"powershell",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-Command",
install_hint,
],
None,
)
return ["/bin/sh", "-c", install_hint], None
npm = _npm_executable()
if npm is None:
_fail(
"npm is required to install this agent, but no native system npm or usable "
"Unsloth-managed Node installation was found. Install Node.js with npm, "
"then re-run."
)
args = shlex.split(install_hint)
env = dict(os.environ)
# dirname, not Path().parent: Path picks its flavour from os.name, so a caller that
# overrides it (the tests do) builds the wrong one. Empty means npm is a bare name,
# and prepending "" would put the cwd on PATH.
npm_dir = os.path.dirname(npm)
current_path = env.get("PATH", "")
if npm_dir:
env["PATH"] = os.pathsep.join([npm_dir, current_path]) if current_path else npm_dir
if os.name == "nt":
command = "& " + " ".join(_powershell_quote(arg) for arg in [npm, *args[1:]])
return (
[
"powershell",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-Command",
command,
],
env,
)
return [npm, *args[1:]], env
def _install_agent(name: str, install_hint: str) -> Optional[str]:
# Missing agent under --launch: offer to run its documented install command, then
# re-resolve it on PATH. Consent-based (we never auto-run a remote install script
@ -2507,22 +2642,15 @@ def _install_agent(name: str, install_hint: str) -> Optional[str]:
typer.secho(warning, fg = "yellow", err = True)
if not typer.confirm(f"Install `{name}` now with `{install_hint}`?", default = False):
return None
# Run each hint through its shell: PowerShell on Windows, /bin/sh elsewhere.
# -ExecutionPolicy Bypass is process-scoped (nothing persistent) so npm's npm.ps1 and
# irm | iex run under the Windows default Restricted policy instead of failing with a
# PSSecurityException.
if os.name == "nt":
install_command = [
"powershell",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-Command",
install_hint,
]
else:
install_command = ["/bin/sh", "-c", install_hint]
if subprocess.run(install_command).returncode != 0:
install_command, install_env = _install_command(install_hint)
try:
result = subprocess.run(install_command, env = install_env)
except OSError as exc:
_fail(
f"Could not run the install command: {exc}. "
f"Run it yourself, then re-run: {install_hint}"
)
if result.returncode != 0:
message = f"Install command failed. Run it yourself, then re-run: {install_hint}"
if os.name == "nt":
# A hand-run retry can still hit the policy; point at the one-time per-user fix.
@ -2545,6 +2673,19 @@ def _install_agent(name: str, install_hint: str) -> Optional[str]:
return executable
def _resolve_or_install_agent(name: str, install_hint: str, resolver) -> str:
executable = resolver(name) or _install_agent(name, install_hint)
if executable is None:
_fail(f"`{name}` not found on PATH. Install it with: {install_hint}")
return executable
def _require_agent_for_launch(name: str, install_hint: str, launch: bool) -> Optional[str]:
if not launch:
return None
return _resolve_or_install_agent(name, install_hint, _which_with_install_dirs)
def _wsl_shim_env(command: list, env: dict, unset_env: tuple) -> tuple[dict, tuple]:
wsl_env_bridge = _wsl_bridge_names(env, unset_env) if _wsl_windows_executable(command) else ()
if not wsl_env_bridge:
@ -2564,9 +2705,7 @@ def _launch(
# Resolve well-known install dirs (e.g. ~/.local/bin) first, so an already-installed
# agent not yet on PATH is found instead of prompting a needless reinstall.
_augment_path_with_install_dirs()
executable = shutil.which(command[0]) or _install_agent(command[0], install_hint)
if executable is None:
_fail(f"`{command[0]}` not found on PATH. Install it with: {install_hint}")
executable = _resolve_or_install_agent(command[0], install_hint, shutil.which)
env, wsl_env_bridge = _wsl_shim_env(command, env, unset_env)
child_env = dict(os.environ)
if wsl_env_bridge:
@ -2678,14 +2817,36 @@ def _run(
def _agents_config_root() -> Path:
ensure_studio_backend_path()
from utils.paths import auth_root
return auth_root() / "agents"
return _studio_auth_root() / "agents"
@contextlib.contextmanager
def _temporary_agent_config(prefix: str):
# These homes live under Studio's auth tree, which nothing else prunes, so reuse the
# locked session helper: a wrapper killed before its finally runs leaves a home that
# the next launch reclaims, and the lock keeps a live session from being swept.
temp_root = _agents_config_root() / ".tmp"
with contextlib.ExitStack() as stack:
try:
temp_root.mkdir(parents = True, exist_ok = True, mode = 0o700)
path = stack.enter_context(_short_ephemeral_session(temp_root, prefix))
except OSError:
# Attaching to a remote or already-running Studio does not need a local auth
# tree, so it may be absent, read-only, or owned by someone else; the key cache
# degrades the same way. Fall back to the system temp dir, where these homes
# lived before. No reclamation there, but the OS prunes it.
path = Path(tempfile.mkdtemp(prefix = prefix))
stack.callback(shutil.rmtree, path, ignore_errors = True)
yield path
# codex-subagent nests CODEX_HOME under <home>/parent, so it needs the short root too.
_CODEX_SHORT_HOME_AGENTS = ("codex", "codex-subagent")
def _ephemeral_session_parent(agent: str) -> Optional[Path]:
"""Return a non-system-temp parent when an agent needs one."""
if os.name != "nt" or agent != "codex":
if os.name != "nt" or agent not in _CODEX_SHORT_HOME_AGENTS:
return None
# Codex creates a deeply nested curated-plugin checkout below CODEX_HOME.
# A normal %TEMP%\unsloth-codex-* home can exceed legacy Windows path
@ -2699,7 +2860,9 @@ def _ephemeral_session_parent(agent: str) -> Optional[Path]:
def _ephemeral_session_prefix(agent: str, parent: Optional[Path]) -> str:
"""Return the platform-specific prefix for an ephemeral agent home."""
return "u-codex-" if agent == "codex" and parent is not None else f"unsloth-{agent}-"
if agent in _CODEX_SHORT_HOME_AGENTS and parent is not None:
return "u-codex-"
return f"unsloth-{agent}-"
@contextlib.contextmanager
@ -2749,9 +2912,9 @@ def _locked_file(path: Path, blocking: bool = True):
handle.close()
def _reclaim_stale_ephemeral_sessions(parent: Path) -> None:
"""Remove abandoned short Codex homes while preserving locked live sessions."""
for path in parent.glob("u-codex-*"):
def _reclaim_stale_ephemeral_sessions(parent: Path, prefix: str) -> None:
"""Remove abandoned session homes while preserving locked live sessions."""
for path in parent.glob(f"{prefix}*"):
if not path.is_dir():
continue
active_lock = path / ".active.lock"
@ -2782,8 +2945,8 @@ def _refresh_ephemeral_session_marker(path: Path, stop: threading.Event) -> None
@contextlib.contextmanager
def _short_ephemeral_session(parent: Path):
"""Create a short Codex home whose lock makes crash cleanup concurrency-safe."""
def _short_ephemeral_session(parent: Path, prefix: str = "u-codex-"):
"""Create a session home whose lock makes crash cleanup concurrency-safe."""
path = None
active_lock = contextlib.ExitStack()
heartbeat_stop = None
@ -2792,8 +2955,8 @@ def _short_ephemeral_session(parent: Path):
with _locked_file(parent / ".cleanup.lock") as cleanup_lock:
if not cleanup_lock: # The blocking acquisition should always succeed.
raise RuntimeError(f"Could not lock ephemeral session root: {parent}")
_reclaim_stale_ephemeral_sessions(parent)
path = Path(tempfile.mkdtemp(prefix = "u-codex-", dir = parent))
_reclaim_stale_ephemeral_sessions(parent, prefix)
path = Path(tempfile.mkdtemp(prefix = prefix, dir = parent))
locked = active_lock.enter_context(_locked_file(path / ".active.lock"))
if not locked:
raise RuntimeError(f"Could not lock ephemeral session home: {path}")
@ -2801,7 +2964,7 @@ def _short_ephemeral_session(parent: Path):
heartbeat = threading.Thread(
target = _refresh_ephemeral_session_marker,
args = (path / ".active.lock", heartbeat_stop),
name = "unsloth-codex-home-heartbeat",
name = "unsloth-agent-home-heartbeat",
daemon = True,
)
heartbeat.start()
@ -2839,16 +3002,16 @@ def _session_config(
resumed next time. Either way the user's real ~/.<agent> config is left untouched.
"""
if launch and not persist:
# Windows codex keeps #7519's short, locked home (MAX_PATH + stale reclaim);
# every other agent uses the Studio-private root.
parent = _ephemeral_session_parent(agent)
prefix = _ephemeral_session_prefix(agent, parent)
if parent is not None:
with _short_ephemeral_session(parent) as path:
with _short_ephemeral_session(parent, prefix) as path:
yield path
else:
path = Path(tempfile.mkdtemp(prefix = _ephemeral_session_prefix(agent, parent)))
try:
with _temporary_agent_config(prefix) as path:
yield path
finally:
shutil.rmtree(path, ignore_errors = True)
else:
# Never wipe this dir: a previously printed recipe may still be running
# an agent whose sessions/state live here, and every config writer
@ -3235,6 +3398,12 @@ def claude(
"""Point Claude Code at the running Unsloth server and start it."""
# Route a leading `org/name` positional to --model; forward the rest to the agent.
model, ctx.args[:] = _consume_positional_model(model, ctx.args)
install_hint = (
"irm https://claude.ai/install.ps1 | iex"
if os.name == "nt"
else "curl -fsSL https://claude.ai/install.sh | bash"
)
_require_agent_for_launch("claude", install_hint, launch)
base, key, entry = _connect(
api_key,
model,
@ -3255,11 +3424,6 @@ def claude(
),
)
model_id = entry["id"]
install_hint = (
"irm https://claude.ai/install.ps1 | iex"
if os.name == "nt"
else "curl -fsSL https://claude.ai/install.sh | bash"
)
if as_subagent:
subagent_id = _subagent_model_id(base, key, entry, model, gguf_variant)
subagent_model = {**entry, "id": subagent_id}
@ -3355,6 +3519,8 @@ def codex(
"""Point OpenAI Codex at the running Unsloth server and start it."""
# Route a leading `org/name` positional to --model; forward the rest to the agent.
model, ctx.args[:] = _consume_positional_model(model, ctx.args)
install_hint = _npm_install_hint("@openai/codex")
_require_agent_for_launch("codex", install_hint, launch)
base, key, entry = _connect(
api_key,
model,
@ -3393,25 +3559,25 @@ def codex(
home,
yolo = yolo,
)
with _codex_parent_overlay(home, launch = launch, persist = persist) as parent_home:
command = [
"codex",
*_codex_subagent_flags(bridge_config),
*_yolo_command_flags("codex", yolo),
*ctx.args,
]
typer.echo(
"Unsloth is available as a local agent. "
"Ask Codex to spawn an Unsloth or local agent."
)
_run(
base,
subagent_model,
{"CODEX_HOME": str(parent_home)},
command,
launch = launch,
install_hint = "npm install -g @openai/codex",
)
parent_home = write_codex_parent_overlay(home / "parent")
command = [
"codex",
*_codex_subagent_flags(bridge_config),
*_yolo_command_flags("codex", yolo),
*ctx.args,
]
typer.echo(
"Unsloth is available as a local agent. "
"Ask Codex to spawn an Unsloth or local agent."
)
_run(
base,
subagent_model,
{"CODEX_HOME": str(parent_home)},
command,
launch = launch,
install_hint = install_hint,
)
return
command = [
"codex",
@ -3424,7 +3590,7 @@ def codex(
with _session_config("codex", launch, persist = persist) as home:
write_codex_config(base, entry, home)
env = {_CODEX_ENV_KEY: key, "CODEX_HOME": str(home)}
_run(base, entry, env, command, launch = launch, install_hint = "npm install -g @openai/codex")
_run(base, entry, env, command, launch = launch, install_hint = install_hint)
@start_app.command("openclaw", cls = _PassthroughCommand, context_settings = _PASSTHROUGH)
@ -3456,6 +3622,12 @@ def openclaw(
# Route a leading `org/name` positional to --model; forward the rest to the agent.
model, ctx.args[:] = _consume_positional_model(model, ctx.args)
_reject_as_subagent("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"
)
_require_agent_for_launch("openclaw", install_hint, launch)
base, key, entry = _connect(
api_key,
model,
@ -3486,11 +3658,6 @@ def openclaw(
if not openclaw_args:
openclaw_args = ["tui", "--local"]
command = ["openclaw", *openclaw_args]
install_hint = (
"iwr -useb https://openclaw.ai/install.ps1 | iex"
if os.name == "nt"
else "curl -fsSL https://openclaw.ai/install.sh | bash"
)
with _session_config("openclaw", launch, persist = persist) as cfg:
config_path = cfg / "openclaw.json"
workspace_path = None
@ -3539,6 +3706,8 @@ def opencode(
"""Point OpenCode at the running Unsloth server and start it."""
# Route a leading `org/name` positional to --model; forward the rest to the agent.
model, ctx.args[:] = _consume_positional_model(model, ctx.args)
install_hint = _npm_install_hint("opencode-ai")
_require_agent_for_launch("opencode", install_hint, launch)
base, key, entry = _connect(
api_key,
model,
@ -3577,11 +3746,6 @@ def opencode(
as_subagent = True,
)
env = {"OPENCODE_CONFIG": str(config_path)}
if launch and _which_with_install_dirs("opencode") is None:
# Provider-filter inspection needs the binary; offer the install now so
# a global/project allowlist is honored on this first launch instead of
# being read only after _launch installs OpenCode.
_install_agent("opencode", "npm install -g opencode-ai")
inline_config = _opencode_subagent_inline_config(config_path, session_permission)
# A project opencode.json outranks the session file and could field-merge its
# own agent.unsloth over ours. Pin ours in the inline overlay so it wins.
@ -3599,7 +3763,7 @@ def opencode(
env,
command,
launch = launch,
install_hint = "npm install -g opencode-ai",
install_hint = install_hint,
)
return
opencode_model = f"{_OPENCODE_PROVIDER}/{entry['id']}"
@ -3670,7 +3834,7 @@ def opencode(
"OPENCODE_CONFIG": str(config_path),
"OPENCODE_CONFIG_CONTENT": json.dumps(inline_config),
}
_run(base, entry, env, command, launch = launch, install_hint = "npm install -g opencode-ai")
_run(base, entry, env, command, launch = launch, install_hint = install_hint)
@start_app.command("hermes", cls = _PassthroughCommand, context_settings = _PASSTHROUGH)
@ -3704,6 +3868,8 @@ def hermes(
_reject_as_subagent("hermes", ctx.args)
native_args = [*_yolo_command_flags("hermes", yolo), *ctx.args]
command = ["hermes", *_hermes_resume_oneshot_args(native_args)]
install_hint = _hermes_install_hint()
_require_agent_for_launch("hermes", install_hint, launch)
base, key, entry = _connect(
api_key,
model,
@ -3723,7 +3889,6 @@ def hermes(
presence_penalty = presence_penalty,
),
)
install_hint = _hermes_install_hint()
with _session_config("hermes", launch, persist = persist) as home:
# HERMES_HOME relocates hermes' whole home dir (config.yaml, sessions, state)
# like CODEX_HOME, so the user's ~/.hermes is left untouched for the session.
@ -3761,6 +3926,13 @@ def pi(
"""Point Pi (coding agent) at the running Unsloth server and start it."""
# Route a leading `org/name` positional to --model; forward the rest to the agent.
model, ctx.args[:] = _consume_positional_model(model, ctx.args)
install_hint = _npm_install_hint(
"@earendil-works/pi-coding-agent",
ignore_scripts = True,
)
if as_subagent and not _PI_SUBAGENT_EXTENSION.is_file():
_fail(f"Missing Pi subagent extension: {_PI_SUBAGENT_EXTENSION}")
_require_agent_for_launch("pi", install_hint, launch)
base, key, entry = _connect(
api_key,
model,
@ -3780,10 +3952,7 @@ def pi(
presence_penalty = presence_penalty,
),
)
install_hint = "npm install -g --ignore-scripts @earendil-works/pi-coding-agent"
if as_subagent:
if not _PI_SUBAGENT_EXTENSION.is_file():
_fail(f"Missing Pi subagent extension: {_PI_SUBAGENT_EXTENSION}")
subagent_id = _subagent_model_id(base, key, entry, model, gguf_variant)
subagent_model = {**entry, "id": subagent_id}
extension = _agent_config_path(_PI_SUBAGENT_EXTENSION, ["pi"])

544
unsloth_cli/tests/test_start.py
View file

@ -13,7 +13,7 @@ import sys
import time
import urllib.error
from pathlib import Path
from types import SimpleNamespace
from types import ModuleType, SimpleNamespace
_REPO_ROOT = Path(__file__).resolve().parents[2]
if str(_REPO_ROOT) not in sys.path:
@ -137,6 +137,7 @@ def test_install_agent_prompts_then_installs(monkeypatch):
monkeypatch.setattr(start.os, "name", "posix")
monkeypatch.setattr(start.sys, "stdin", SimpleNamespace(isatty = lambda: True))
monkeypatch.setattr(start.typer, "confirm", lambda *a, **k: True)
monkeypatch.setattr(start, "_npm_executable", lambda: "/usr/local/bin/npm")
ran = []
monkeypatch.setattr(
start.subprocess,
@ -148,7 +149,7 @@ def test_install_agent_prompts_then_installs(monkeypatch):
monkeypatch.setattr(start.shutil, "which", lambda _: "/usr/local/bin/codex")
executable = start._install_agent("codex", "npm install -g @openai/codex")
assert executable == "/usr/local/bin/codex"
assert ran == [["/bin/sh", "-c", "npm install -g @openai/codex"]]
assert ran == [["/usr/local/bin/npm", "install", "-g", "@openai/codex"]]
def test_install_agent_uses_powershell_on_windows(monkeypatch):
@ -178,9 +179,12 @@ def test_install_agent_windows_failure_hints_execution_policy(monkeypatch, capsy
# A failed install on Windows points the user at the per-user execution-policy fix:
# our subprocess bypasses the policy, but their own shell may still block npm.ps1
# (PSSecurityException) when they run the install by hand.
monkeypatch.setattr(start.os, "name", "nt")
_simulate_windows(monkeypatch)
monkeypatch.setattr(start.sys, "stdin", SimpleNamespace(isatty = lambda: True))
monkeypatch.setattr(start.typer, "confirm", lambda *a, **k: True)
monkeypatch.setattr(
start, "_npm_executable", lambda: r"C:\Users\me\AppData\Roaming\npm\npm.cmd"
)
monkeypatch.setattr(
start.subprocess,
"run",
@ -196,6 +200,23 @@ def test_install_agent_windows_failure_hints_execution_policy(monkeypatch, capsy
assert "Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned" in err
def test_install_command_uses_resolved_npm_cmd_on_windows(monkeypatch):
_simulate_windows(monkeypatch)
monkeypatch.setattr(start, "_npm_executable", lambda: r"C:\Managed Node\npm.cmd")
command, env = start._install_command(start._npm_install_hint("@openai/codex"))
assert command == [
"powershell",
"-NoProfile",
"-ExecutionPolicy",
"Bypass",
"-Command",
"& 'C:\\Managed Node\\npm.cmd' install -g '@openai/codex'",
]
assert env is not None
def test_install_agent_posix_failure_omits_execution_policy_hint(monkeypatch, capsys):
# The execution-policy hint is Windows-only; a POSIX install failure must not mention it.
monkeypatch.setattr(start.os, "name", "posix")
@ -215,6 +236,227 @@ def test_install_agent_posix_failure_omits_execution_policy_hint(monkeypatch, ca
assert "Set-ExecutionPolicy" not in err
def test_npm_install_hint_uses_user_prefix_on_posix(monkeypatch, tmp_path):
monkeypatch.setattr(start.os, "name", "posix")
monkeypatch.setattr(start.Path, "home", lambda: tmp_path)
hint = start._npm_install_hint("@openai/codex")
assert shlex.split(hint) == [
"npm",
"install",
"-g",
"--prefix",
str(tmp_path / ".local"),
"@openai/codex",
]
@pytest.mark.skipif(os.name == "nt", reason = "managed node layout is POSIX")
def test_npm_executable_uses_studio_managed_node(monkeypatch, tmp_path):
start.ensure_studio_backend_path()
from utils import node_runtime
managed_bin = tmp_path / "node" / "bin"
managed_bin.mkdir(parents = True)
node = managed_bin / "node"
npm = managed_bin / "npm"
node.touch()
npm.touch()
node.chmod(0o755)
npm.chmod(0o755)
monkeypatch.setattr(start.shutil, "which", lambda _: None)
monkeypatch.setattr(node_runtime, "managed_node_binary", lambda: node)
monkeypatch.setattr(node_runtime, "resolve_node_executable", lambda: str(node))
assert start._npm_executable() == str(npm)
@pytest.mark.skipif(os.name == "nt", reason = "WSL scenario")
def test_npm_executable_skips_wsl_shim_and_finds_native_npm(monkeypatch):
# WSL inherits the Windows PATH, so a shim can precede a usable native npm.
native = "/usr/bin/npm"
shim = "/mnt/c/Program Files/nodejs/npm"
monkeypatch.setenv("WSL_DISTRO_NAME", "Ubuntu")
monkeypatch.setenv("PATH", "/mnt/c/Program Files/nodejs:/usr/bin")
monkeypatch.setattr(start, "_managed_node_tools", lambda: None)
found = {"/mnt/c/Program Files/nodejs": shim, "/usr/bin": native}
def fake_which(name, path = None):
if os.path.isabs(name):
return name
# No path given means search all of PATH, so the shim wins as it does in WSL.
for entry in (path or os.environ["PATH"]).split(os.pathsep):
if entry in found:
return found[entry]
return None
monkeypatch.setattr(start.shutil, "which", fake_which)
assert start._npm_executable() == native
@pytest.mark.skipif(os.name == "nt", reason = "POSIX hint form")
def test_npm_install_hint_without_resolvable_home(monkeypatch):
# A bare container UID has no home; the hint must still build so an already
# installed agent can launch.
def no_home():
raise RuntimeError("no home directory")
monkeypatch.setattr(start.Path, "home", staticmethod(no_home))
assert start._npm_install_hint("@openai/codex") == "npm install -g @openai/codex"
def test_managed_node_probe_tolerates_unsupported_path_flavour(monkeypatch):
def unsupported_backend_path():
raise RuntimeError("unsupported path flavour")
monkeypatch.setattr(start, "ensure_studio_backend_path", unsupported_backend_path)
assert start._managed_node_tools() is None
@pytest.mark.skipif(os.name == "nt", reason = "managed node layout is POSIX")
def test_npm_executable_uses_managed_npm_when_system_node_has_none(monkeypatch, tmp_path):
start.ensure_studio_backend_path()
from utils import node_runtime
managed_bin = tmp_path / "node" / "bin"
managed_bin.mkdir(parents = True)
node = managed_bin / "node"
npm = managed_bin / "npm"
node.touch()
npm.touch()
node.chmod(0o755)
npm.chmod(0o755)
monkeypatch.setattr(
start.shutil,
"which",
lambda name: "/usr/bin/node" if name == "node" else None,
)
monkeypatch.setattr(node_runtime, "managed_node_binary", lambda: node)
monkeypatch.setattr(node_runtime, "resolve_node_executable", lambda: str(node))
assert start._npm_executable() == str(npm)
@pytest.mark.skipif(os.name == "nt", reason = "WSL scenario")
def test_npm_executable_prefers_managed_npm_over_windows_npm_in_wsl(monkeypatch, tmp_path):
start.ensure_studio_backend_path()
from utils import node_runtime
managed_bin = tmp_path / "node" / "bin"
managed_bin.mkdir(parents = True)
node = managed_bin / "node"
npm = managed_bin / "npm"
node.touch(mode = 0o755)
npm.touch(mode = 0o755)
windows_npm = "/mnt/c/Program Files/nodejs/npm"
monkeypatch.setenv("WSL_DISTRO_NAME", "Ubuntu")
monkeypatch.setattr(
start.shutil,
"which",
lambda name: windows_npm if name in ("npm", windows_npm) else None,
)
monkeypatch.setattr(node_runtime, "managed_node_binary", lambda: node)
monkeypatch.setattr(node_runtime, "resolve_node_executable", lambda: str(node))
assert start._npm_executable() == str(npm)
@pytest.mark.skipif(os.name == "nt", reason = "managed node layout is POSIX")
def test_augment_path_includes_studio_managed_node(monkeypatch, tmp_path):
start.ensure_studio_backend_path()
from utils import node_runtime
managed_bin = tmp_path / "node" / "bin"
managed_bin.mkdir(parents = True)
node = managed_bin / "node"
npm = managed_bin / "npm"
node.touch(mode = 0o755)
npm.touch(mode = 0o755)
monkeypatch.setattr(start.Path, "home", lambda: tmp_path / "home")
monkeypatch.setattr(node_runtime, "managed_node_binary", lambda: node)
monkeypatch.setattr(node_runtime, "resolve_node_executable", lambda: str(node))
monkeypatch.setenv("PATH", os.pathsep.join([os.defpath, str(managed_bin)]))
start._augment_path_with_install_dirs()
path = os.environ["PATH"].split(os.pathsep)
assert path[0] == str(managed_bin)
assert path.count(str(managed_bin)) == 1
def test_install_agent_missing_npm_names_node_requirement(monkeypatch, capsys):
monkeypatch.setattr(start.sys, "stdin", SimpleNamespace(isatty = lambda: True))
monkeypatch.setattr(start.typer, "confirm", lambda *a, **k: True)
monkeypatch.setattr(start, "_npm_executable", lambda: None)
monkeypatch.setattr(
start.subprocess,
"run",
lambda *a, **k: pytest.fail("should not run an installer without npm"),
)
with pytest.raises(start.typer.Exit):
start._install_agent("codex", "npm install -g @openai/codex")
err = capsys.readouterr().err
assert "npm is required" in err
assert "Unsloth-managed Node" in err
assert "Install Node.js with npm" in err
def test_install_agent_reports_os_error_without_traceback(monkeypatch, capsys):
monkeypatch.setattr(start.sys, "stdin", SimpleNamespace(isatty = lambda: True))
monkeypatch.setattr(start.typer, "confirm", lambda *a, **k: True)
monkeypatch.setattr(start, "_npm_executable", lambda: "/broken/npm")
monkeypatch.setattr(
start.subprocess,
"run",
lambda *args, **kwargs: (_ for _ in ()).throw(PermissionError("permission denied")),
)
with pytest.raises(start.typer.Exit):
start._install_agent("codex", "npm install -g @openai/codex")
err = capsys.readouterr().err
assert "Could not run the install command: permission denied" in err
assert "Run it yourself, then re-run" in err
@pytest.mark.skipif(os.name == "nt", reason = "POSIX install command")
def test_install_agent_runs_managed_npm_with_its_node_on_path(monkeypatch, tmp_path):
monkeypatch.setattr(start.os, "name", "posix")
monkeypatch.setattr(start.Path, "home", lambda: tmp_path)
monkeypatch.setattr(start.sys, "stdin", SimpleNamespace(isatty = lambda: True))
monkeypatch.setattr(start.typer, "confirm", lambda *a, **k: True)
npm = tmp_path / "managed-node" / "bin" / "npm"
monkeypatch.setattr(start, "_npm_executable", lambda: str(npm))
monkeypatch.setattr(start.shutil, "which", lambda _: "/usr/local/bin/codex")
captured = {}
def run(command, **kwargs):
captured["command"] = command
captured["env"] = kwargs["env"]
return SimpleNamespace(returncode = 0)
monkeypatch.setattr(start.subprocess, "run", run)
hint = start._npm_install_hint("@openai/codex")
assert start._install_agent("codex", hint) == "/usr/local/bin/codex"
assert captured["command"] == [
str(npm),
"install",
"-g",
"--prefix",
str(tmp_path / ".local"),
"@openai/codex",
]
assert captured["env"]["PATH"].split(os.pathsep)[0] == str(npm.parent)
def test_install_agent_warns_remote_installer_is_unverified_third_party(monkeypatch, capsys):
# Before the confirm, a remote installer must name the URL it fetches so the
# user consents to a specific source rather than blindly accepting.
@ -531,6 +773,25 @@ def _parse_toml(text: str) -> dict:
return tomllib.loads(text)
def test_project_declares_direct_click_dependency():
project = _parse_toml((_REPO_ROOT / "pyproject.toml").read_text(encoding = "utf-8"))
assert "click>=8.0" in project["project"]["dependencies"]
def test_agent_paths_use_cli_studio_home_without_backend_imports(monkeypatch, tmp_path):
studio = ModuleType("unsloth_cli.commands.studio")
studio.STUDIO_HOME = tmp_path
monkeypatch.setitem(sys.modules, studio.__name__, studio)
monkeypatch.setattr(
start,
"ensure_studio_backend_path",
lambda: pytest.fail("agent paths should not import backend runtime packages"),
)
assert start._key_cache_path() == tmp_path / "auth" / "agent_api_key.json"
assert start._agents_config_root() == tmp_path / "auth" / "agents"
def test_merge_codex_config_fresh():
merged = start._merge_codex_config("", BASE)
parsed = _parse_toml(merged)
@ -797,20 +1058,34 @@ def test_write_codex_parent_overlay_uses_windows_home_for_windows_codex(tmp_path
assert (overlay / "auth.json").read_text() == '{"auth": "windows"}\n'
def test_codex_parent_overlay_launch_uses_private_temp_root_and_cleans_up(tmp_path, monkeypatch):
def test_codex_parent_overlay_can_use_session_home(tmp_path, monkeypatch):
source = tmp_path / "user-codex"
source.mkdir()
(source / "auth.json").write_text("{}\n")
monkeypatch.setenv("CODEX_HOME", str(source))
session_home = tmp_path / "session"
overlay = start.write_codex_parent_overlay(session_home / "parent")
assert overlay == session_home / "parent"
assert start._CODEX_SUBAGENT_ROUTING_INSTRUCTIONS in (overlay / "AGENTS.md").read_text()
assert overlay.exists()
def test_ephemeral_codex_parent_overlay_is_cleaned_with_session(tmp_path, monkeypatch):
source = tmp_path / "user-codex"
source.mkdir()
monkeypatch.setenv("CODEX_HOME", str(source))
agents_root = tmp_path / "agents"
monkeypatch.setattr(start, "_agents_config_root", lambda: agents_root)
with start._codex_parent_overlay(tmp_path / "session", launch = True, persist = False) as overlay:
assert overlay.parent == agents_root / ".tmp"
assert start._CODEX_SUBAGENT_ROUTING_INSTRUCTIONS in (overlay / "AGENTS.md").read_text()
with start._session_config("codex-subagent", launch = True) as session_home:
overlay = start.write_codex_parent_overlay(session_home / "parent")
assert overlay.exists()
assert session_home.exists()
assert not overlay.exists()
assert not session_home.exists()
@pytest.mark.skipif(os.name == "nt", reason = "WSL scenario")
@ -895,6 +1170,115 @@ def test_unsupported_agents_reject_as_subagent(agent, flag):
assert f"--as-subagent is not supported for {agent}." in result.output
@pytest.mark.parametrize("agent", ["claude", "codex", "openclaw", "opencode", "hermes", "pi"])
def test_launch_preflights_agent_before_connect(agent, monkeypatch):
events = []
def require(name, hint, launch):
assert name == agent
assert hint
assert launch is True
events.append("agent")
def connect(*args, **kwargs):
events.append("connect")
raise RuntimeError("stop after ordering check")
monkeypatch.setattr(start, "_require_agent_for_launch", require)
monkeypatch.setattr(start, "_connect", connect)
result = CliRunner().invoke(start.start_app, [agent])
assert result.exit_code == 1
assert events == ["agent", "connect"]
def test_declined_opencode_subagent_install_stops_before_connect(monkeypatch):
installs = []
monkeypatch.setattr(start, "_which_with_install_dirs", lambda _: None)
monkeypatch.setattr(
start,
"_install_agent",
lambda name, hint: installs.append((name, hint)),
)
monkeypatch.setattr(
start,
"_connect",
lambda *a, **k: pytest.fail("declined install must stop before model connection"),
)
result = CliRunner().invoke(start.start_app, ["opencode", "--as-subagent"])
assert result.exit_code == 1
assert len(installs) == 1
assert installs[0][0] == "opencode"
@pytest.mark.parametrize("agent", ["claude", "codex", "openclaw", "opencode", "hermes", "pi"])
def test_noninteractive_missing_agent_stops_before_connect(agent, monkeypatch):
monkeypatch.setattr(start, "_which_with_install_dirs", lambda _: None)
monkeypatch.setattr(
start.subprocess,
"run",
lambda *args, **kwargs: pytest.fail("non-interactive launch must not install"),
)
monkeypatch.setattr(
start,
"_connect",
lambda *args, **kwargs: pytest.fail("missing agent must stop before connection"),
)
result = CliRunner().invoke(start.start_app, [agent])
assert result.exit_code == 1
assert f"`{agent}` not found on PATH" in result.output
@pytest.mark.parametrize("agent", ["claude", "codex", "openclaw", "opencode", "hermes", "pi"])
def test_no_launch_skips_agent_resolution(agent, monkeypatch):
monkeypatch.setattr(
start,
"_which_with_install_dirs",
lambda _: pytest.fail("--no-launch must not resolve an agent"),
)
monkeypatch.setattr(
start,
"_install_agent",
lambda *args: pytest.fail("--no-launch must not install an agent"),
)
def stop_at_connect(*args, **kwargs):
raise RuntimeError
monkeypatch.setattr(start, "_connect", stop_at_connect)
result = CliRunner().invoke(start.start_app, [agent, "--no-launch"])
assert result.exit_code == 1
assert isinstance(result.exception, RuntimeError)
def test_missing_pi_subagent_extension_fails_before_install_or_connect(monkeypatch, tmp_path):
monkeypatch.setattr(start, "_PI_SUBAGENT_EXTENSION", tmp_path / "missing.ts")
monkeypatch.setattr(
start,
"_require_agent_for_launch",
lambda *args: pytest.fail("local prerequisites must be checked before installation"),
)
monkeypatch.setattr(
start,
"_connect",
lambda *args, **kwargs: pytest.fail(
"local prerequisites must be checked before connection"
),
)
result = CliRunner().invoke(start.start_app, ["pi", "--as-subagent"])
assert result.exit_code == 1
assert "Missing Pi subagent extension" in result.output
@pytest.fixture()
def fake_studio(tmp_path, monkeypatch):
calls = []
@ -934,6 +1318,7 @@ def fake_studio(tmp_path, monkeypatch):
monkeypatch.setattr(start, "_key_cache_path", lambda: tmp_path / "agent_api_key.json")
# --no-launch session configs land under tmp instead of the real Unsloth dir.
monkeypatch.setattr(start, "_agents_config_root", lambda: tmp_path / "agents")
monkeypatch.setattr(start, "_require_agent_for_launch", lambda *args: None)
# No `claude` on PATH, so _claude_flags never probes the real binary.
monkeypatch.setattr(start.shutil, "which", lambda _: None)
monkeypatch.delenv("UNSLOTH_API_KEY", raising = False)
@ -3678,12 +4063,12 @@ def test_opencode_subagent_installs_binary_before_filter_inspection(fake_studio,
lambda name: "/usr/local/bin/opencode" if installed.get("done") else None,
)
def install(name, hint):
def require(name, hint, launch):
assert launch is True
installed["done"] = True
installed["name"] = name
return "/usr/local/bin/opencode"
monkeypatch.setattr(start, "_install_agent", install)
monkeypatch.setattr(start, "_require_agent_for_launch", require)
inspected = {}
def inline(path, permission):
@ -4732,12 +5117,15 @@ def test_session_config_persist_uses_stable_dir_and_survives(monkeypatch, tmp_pa
assert (home / "marker").read_text() == "kept"
def test_session_config_default_launch_is_ephemeral():
# Default launch (no --persist) still uses a throwaway temp dir wiped on exit.
def test_session_config_default_launch_is_ephemeral(monkeypatch, tmp_path):
agents_root = tmp_path / "agents"
monkeypatch.setattr(start, "_agents_config_root", lambda: agents_root)
with start._session_config("codex", launch = True) as home:
assert home.exists()
parent = start._ephemeral_session_parent("codex")
assert home.name.startswith(start._ephemeral_session_prefix("codex", parent))
if parent is None:
assert home.parent == agents_root / ".tmp"
assert not home.exists()
@ -4811,6 +5199,135 @@ def test_session_config_reclaims_old_short_homes_but_keeps_recent_and_live(monke
assert not first.exists()
@pytest.mark.parametrize("agent", ["codex", "codex-subagent"])
def test_windows_codex_homes_use_the_short_parent(monkeypatch, tmp_path, agent):
# codex-subagent nests CODEX_HOME under <home>/parent, so the long Studio auth
# path would eat even more of the legacy MAX_PATH budget than a plain launch.
monkeypatch.setattr(start.os, "name", "nt")
monkeypatch.setattr(start.Path, "home", staticmethod(lambda: tmp_path))
assert start._ephemeral_session_parent(agent) == tmp_path / ".unsloth" / ".tmp"
parent = start._ephemeral_session_parent(agent)
assert start._ephemeral_session_prefix(agent, parent) == "u-codex-"
def test_non_codex_agents_keep_the_studio_private_root(monkeypatch, tmp_path):
monkeypatch.setattr(start.os, "name", "nt")
monkeypatch.setattr(start.Path, "home", staticmethod(lambda: tmp_path))
assert start._ephemeral_session_parent("claude") is None
assert start._ephemeral_session_prefix("claude", None) == "unsloth-claude-"
def test_session_config_falls_back_when_existing_temp_root_is_unwritable(monkeypatch, tmp_path):
# mkdir(exist_ok = True) succeeds on a root that already exists but cannot be written,
# so the lock file is the first thing to fail.
agents = tmp_path / "agents"
temp_root = agents / ".tmp"
temp_root.mkdir(parents = True)
os.chmod(temp_root, 0o500)
monkeypatch.setattr(start, "_agents_config_root", lambda: agents)
try:
with start._session_config("claude", launch = True) as home:
assert home.exists()
assert temp_root not in home.parents
finally:
os.chmod(temp_root, 0o700)
assert not home.exists()
def test_augment_path_keeps_managed_node_without_a_home(monkeypatch, tmp_path):
# A bare container UID has no home, but the managed Node is still the runtime that
# resolves a Node-backed shim, so it must stay on PATH.
managed_bin = tmp_path / "node" / "bin"
managed_bin.mkdir(parents = True)
monkeypatch.setattr(
start,
"_managed_node_tools",
lambda: (managed_bin / "node", managed_bin / "npm", True),
)
monkeypatch.setattr(
start.Path,
"home",
staticmethod(lambda: (_ for _ in ()).throw(RuntimeError("no home directory"))),
)
monkeypatch.setenv("PATH", "/usr/bin")
start._augment_path_with_install_dirs()
assert os.environ["PATH"].split(os.pathsep)[0] == str(managed_bin)
def test_augment_path_leaves_path_alone_when_nothing_to_add(monkeypatch):
monkeypatch.setattr(start, "_managed_node_tools", lambda: None)
monkeypatch.setattr(
start.Path,
"home",
staticmethod(lambda: (_ for _ in ()).throw(RuntimeError("no home directory"))),
)
monkeypatch.setenv("PATH", "/usr/bin")
start._augment_path_with_install_dirs()
assert os.environ["PATH"] == "/usr/bin"
def test_probe_env_carries_install_dirs_and_restores_path(monkeypatch, tmp_path):
# A shim resolved via Studio's managed Node needs that node on PATH when it runs.
managed_bin = tmp_path / "node" / "bin"
managed_bin.mkdir(parents = True)
monkeypatch.setattr(
start,
"_managed_node_tools",
lambda: (managed_bin / "node", managed_bin / "npm", True),
)
before = os.environ.get("PATH")
env = start._probe_env(OPENCODE_CONFIG = "/tmp/cfg.json")
assert str(managed_bin) in env["PATH"]
assert env["OPENCODE_CONFIG"] == "/tmp/cfg.json"
assert os.environ.get("PATH") == before
def test_session_config_falls_back_when_studio_auth_root_is_unwritable(monkeypatch, tmp_path):
# Attaching to a remote or already-running Studio does not need a local auth tree, so
# an absent or read-only one must not stop the launch.
readonly = tmp_path / "readonly"
readonly.mkdir(mode = 0o500)
monkeypatch.setattr(start, "_agents_config_root", lambda: readonly / "agents")
with start._session_config("claude", launch = True) as home:
assert home.exists()
assert readonly not in home.parents
assert not home.exists()
def test_session_config_reclaims_abandoned_homes_for_non_codex_agents(monkeypatch, tmp_path):
# These homes sit under Studio's auth tree, which nothing else prunes, so a wrapper
# killed before its finally runs must be reclaimed by the next launch.
agents_root = tmp_path / "agents"
temp_root = agents_root / ".tmp"
temp_root.mkdir(parents = True)
monkeypatch.setattr(start, "_agents_config_root", lambda: agents_root)
abandoned = temp_root / "unsloth-claude-abandoned"
abandoned.mkdir()
(abandoned / ".active.lock").write_bytes(b"\0")
(abandoned / "state.json").write_text("left behind")
old = time.time() - start._CODEX_EPHEMERAL_STALE_SECONDS - 1
os.utime(abandoned / ".active.lock", (old, old))
recent = temp_root / "unsloth-claude-still-running"
recent.mkdir()
(recent / ".active.lock").write_bytes(b"\0")
with start._session_config("claude", launch = True) as home:
assert not abandoned.exists()
assert recent.exists()
assert home.parent == temp_root
assert not home.exists()
def test_session_config_serializes_normal_short_home_deletion(monkeypatch, tmp_path):
short_parent = tmp_path / "u"
short_parent.mkdir()
@ -4875,7 +5392,8 @@ def test_default_launch_home_is_ephemeral(agent, fake_studio, tmp_path, monkeypa
home = captured["env"][_RESUME_ENV_VAR[agent]]
parent = start._ephemeral_session_parent(agent)
assert start._ephemeral_session_prefix(agent, parent) in home
assert str(tmp_path / "agents") not in home
if parent is None:
assert Path(home).parent == tmp_path / "agents" / ".tmp"
def test_resume_opencode_config_in_stable_dir(fake_studio, tmp_path, monkeypatch):